Skip to content

Device latest-sample projection (fast restart seed)

  • Date: 2026-07-13
  • Surface: backend (crates/xylolabs-server ingest manager + crates/xylolabs-db)
  • Trigger (P3, Wave 8): the in-memory latest_samples cache (latest reading per (device, stream), powering the operator records / device-detail pages) starts empty on every deploy. seed_latest_samples_from_db() repopulates it at boot by scanning the ENTIRE metadata_chunks table (DISTINCT ON (device, stream_name), 552k+ pairs in prod, capped at 5000) AND downloading + zstd-decoding an S3 chunk per pair to extract the latest value. Slow and costly on every boot.

Goal

Persist the latest sample per (device, stream) in a small projection table so restart seeding is a single cheap SELECT — no metadata_chunks scan, no S3 download/decode. Additive (the in-memory cache + its read path are unchanged) and hot-path-safe (the flush gains one bounded, best-effort, non-fatal upsert).

Schema — device_latest_samples

CREATE TABLE device_latest_samples (
    device_id    uuid NOT NULL REFERENCES devices(id) ON DELETE CASCADE,
    stream_name  text NOT NULL,
    stream_id    uuid NOT NULL,
    stream_index integer NOT NULL,
    timestamp_us bigint NOT NULL,
    value        jsonb NOT NULL,
    updated_at   timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (device_id, stream_name)
);
CREATE INDEX ix_device_latest_samples_updated ON device_latest_samples (updated_at DESC);
  • Natural key (device_id, stream_name) matches the cache key and the name-keyed reseed rule (every session mints fresh metadata_streams rows, so keying on stream_id explodes the pair space — P741). stream_id/stream_index are stored to reconstruct LatestSampleEvent on seed.
  • value is the LatestSampleEvent.value (serde_json::Value) verbatim — the seed needs no S3 decode because the value is already here.
  • FK to devices with ON DELETE CASCADE keeps it clean; a flush's device_id is always a live device, so the FK never fails on the hot path.
  • timestamp_us is u64 in code, stored as bigint (i64) — realistic µs timestamps are far below i64::MAX.

Repo — crates/xylolabs-db/src/repo/device_latest_sample.rs

  • struct DeviceLatestSampleRow { device_id, stream_name, stream_id, stream_index, timestamp_us: i64, value: JsonValue }
  • upsert_batch(pool, rows: &[DeviceLatestSampleUpsert]) — one INSERT … SELECT FROM UNNEST(…) ON CONFLICT (device_id, stream_name) DO UPDATE … WHERE EXCLUDED.timestamp_us > device_latest_samples.timestamp_us. The monotonic guard keeps the persisted latest from regressing when a device backfills old data. The in-memory cache applies the SAME guard (insert_latest_if_newer in manager.rs, loop3-c1 AGG-1) so both representations agree: the projection is the durable source of truth for "newest"; the cache is its monotonic serving mirror.
  • list_recent(pool, limit) — ORDER BY updated_at DESC LIMIT $1 for the seed.

Hot-path hook — latest-sample block (manager.rs, the newest block ~1330)

After the existing loop that inserts newest into the in-memory cache (and after the write lock is released), build the batch from the same newest and upsert_batch it. This block runs on EVERY process_batch call (every incoming batch), not only on flush, so the DB upsert is spawned under a bounded semaphore (LATEST_PROJECTION_CONCURRENCY = 8, try_acquire_owned → drop-with-warn on exhaustion), mirroring the anomaly-dispatch pattern. It therefore never blocks the WS read loop or delays the next batch's flush/backpressure checks. Best-effort: a dropped-under-load or failed upsert never affects ingestion — the projection self-heals on the next batch, and the cache (its monotonic serving mirror) keeps serving. Bounded by streams-per-batch (~10).

Boot seed — seed_latest_samples_from_db() (encapsulated fallback)

Add seed_latest_samples_from_projection() — list_recent(pool, 5000) → reconstruct LatestSampleEvent per row → insert into the cache's vacant slots. Change seed_latest_samples_from_db() to try the projection FIRST; if it returns 0 rows (empty projection = the very first deploy of this feature), fall back to the existing metadata_chunks + S3 scan (unchanged). main.rs is untouched — it still calls seed_latest_samples_from_db().

The S3-scan fallback also BACKFILLS the projection (one upsert_batch of its results). This is what makes the fallback complete: after the first boot the projection contains every pair the S3 scan found — including streams that stopped reporting before the feature deployed and will never flush again. Without the backfill, a projected > 0 (partially populated) projection would wrongly suppress the S3 fallback and leave un-flushed/stale pairs unseeded.

Transition: first deploy → projection empty → S3 scan populates the cache AND backfills the projection → every subsequent boot uses the fast, complete projection seed and never runs the S3 scan again. No backfill migration needed.

Testing

  • Repo integration test (Postgres): upsert_batch inserts, a second upsert with a NEWER timestamp_us advances the row, a second upsert with an OLDER timestamp_us is a no-op (monotonic guard); list_recent returns rows newest-first.
  • Manager: after a flush, the projection has the flushed latest per stream; a fresh manager's seed_latest_samples_from_db populates the cache from the projection (no S3) and latest_samples_for_device returns them.
  • Migration applies on a fresh DB (monotonic prefix 20260713100000).

Out of scope / guardrails

  • No change to the in-memory cache's read path, the SSE broadcast, or latest_sample_event construction.
  • No change to is_online/health/alert logic.
  • The projection is a cache/optimization, not a system of record — the metadata_chunks + S3 data remains the source of truth.