materializer: replace per-pass signals_latest fold with a daily watermarked refresh

[elffjs]

Problem

The per-pass signals_latest upkeep (captureRollupDelta + foldSignalsRollup, run inside every span transaction) dominates materializer pass cost: ~300s at afternoon peak vs ~10s at dawn (dq_materializer_phase_seconds shows commit at ~1800–2200s/hour; decode is ~10s/hour). The fold is O(batch) in rows but O(day-partition files × 3) in I/O:

• The recency/locrec CTEs re-derive latest-per-(subject, name) from lake.signals with a correlated bound (s.timestamp >= p.prev_ts), which cannot drive static partition pruning — no constant day = predicate exists in the query — so each pass re-opens roughly the affected buckets' current-day files.
• The scan happens three times per pass (recency, locrec, and captureRollupDelta's NOT-EXISTS probe).
• File opens are ~0.5s-class over httpfs, and the day partition's file count grows all day.

Result: the daily pass-duration sawtooth, decode ceiling below business-hours ingest (~255 snapshots/h), and the 2026-08-05/06 backlogs. Vehicle count is irrelevant to the cost — it's priced in file opens.

The fold's correctness machinery (recompute-equivalence, cloud_event_id tie-breaks against the base, QUALIFY dedup matching the read path, self-healing recency) exists to keep the table continuously exact — but the table stopped being a serving surface when the KV shipped: LATEST_KV_READ_MODE=serve in prod, negative path included. Its remaining jobs are KV bootstrap/reconcile source and fallback. We're paying the old system's ongoing cost for a backup copy.

Design

Remove the fold from the span transaction; refresh the rollup once daily from a watermark.

1. Span commit keeps only: base inserts (signals/events) + KV publish + cursor advance. No rollup reads/writes.
2. Daily refresh step (in the materializer loop — preserves the single-writer invariant; NOT a separate pod/cron):
   • Runs after the UTC-midnight partition rollover once the cursor has settled past it (~03:30–04:00 UTC — also this node's traffic trough).
   • Folds lake.signals WHERE timestamp >= <watermark> into the rollup with a constant watermark literal → static file pruning, one complete settled partition in steady state.
   • Watermark-driven and idempotent: a missed run folds two days next time; no operator memory needed. Watermark stored alongside the rollup (same transaction).
   • Simple arg_max/QUALIFY over (rollup ∪ tail) — the per-pass fold's tie-break/delta machinery can be dropped or reused; equivalence bar is unchanged (differential test stays).
3. Coverage reconcile becomes watermark-aware — the sharp edge. It currently diffs KV against the rollup assuming freshness; against a daily rollup it must diff against (rollup ∪ signals-since-watermark), or bound comparisons at the watermark. Otherwise reconcile "repairs" the KV backwards up to a day. Same care as the CAS/multi-writer rules on the bucket.
4. BootstrapFromRollup gains a tail replay: bootstrap from rollup (≤24h stale), then fold signals since watermark — constant predicate, 1–2 partitions, minutes, and stays bounded forever (vs O(history) pure-drop rebuild that grows with retention).

Cadence rationale (why daily, not finer/coarser)

The day partition is the cost quantum. Hourly refreshes rescan today-so-far repeatedly during business hours (recreating the problem, milder); weekly saves nothing per-day-covered and grows the recovery/reconcile tail 7x. Daily anchored to rollover folds exactly one complete settled partition. The rollup has no freshness SLA of its own anymore — its only consumer is disaster recovery, which is minutes-fast at any cadence.

Expected wins

• Span passes lose their dominant, day-length-dependent term → sawtooth flattens, decode ceiling stops depending on time of day. MATERIALIZER_MAX_SNAPSHOT_SPAN=32 (dimo-node#123 stopgap) can return to 16.
• signals_latest churn drops from ~288 rewrite cycles/day to 1 → the delete-churn fragmentation problem (din#13's rewrite_data_files motivation) essentially disappears at the source; din maintenance cycles get cheaper.

Out of scope / later

• events_latest: same pattern, no KV equivalent yet — untouched here; measure its share and consider the same treatment (or a KV) separately.
• Dropping the table entirely: viable phase 2 once the KV has solo-serving history; the daily job and lake_rollup.go fallback removal make the eventual drop trivial. Not now — the rollup-as-recovery-snapshot keeps KV rebuild O(1 day) instead of O(history).
• LAKE_REBUILD_ROLLUP_ON_BOOT / RecomputeRollup stay as the deep-DR path (unchanged cost class).

Migration order (each step independently shippable)

1. Add watermark column/row + daily refresh step (flag-gated), per-pass fold still on: differential-compare daily output vs incremental for a few days.
2. Make coverage reconcile + bootstrap watermark-aware.
3. Flip the flag: fold off, daily on. Watch dq_lake_latest_kv_* coverage metrics and span durations.
4. Remove the per-pass fold code + MATERIALIZER_MAX_SNAPSHOT_SPAN back to default; retire lake_rollup.go fallback when confident.

[elffjs]

Correction to "Problem" ¶3: signals_latest is not just DR/fallback — four read paths still serve from it (lake_rollup.go), and a daily-stale rollup changes what they return. Per-path disposition, which amends the migration order:

Path	Today	Under daily refresh
signalsLatest (no source filter)	KV (serve)	unchanged
signalsLatest (source-filtered)	scan path, bypasses rollup	unchanged
getLatestSignalsRollup (KV error/version fallback)	fresh rollup	day-stale answers on the rare fallback — acceptable; miss is already authoritative under negative-serve
allLatest (getAllLatestSignalsRollup)	rollup only	must move to KV first — same Entry, all names (the "future work" item from the serve rollout)
availableSignals	rollup only	move to KV (keys of Entry.Signals); staleness alternative means a newly-added vehicle reads empty for up to a day
summaries	rollup only	hybrid: KV Entry has no count/first_seen (checked — latestkv.Row/SignalValue carry value+loc+CEID only). But those two fields are monotone slow-movers where ≤24h staleness is benign; lastSeen is the freshness-critical one and Entry.LastSeen() already exists. Serve count/firstSeen from the daily rollup + lastSeen overlay from KV. Deliberately do NOT add count arithmetic to the KV fold — incremental count needs the NOT-EXISTS dedup probe (redelivery would double-count), which is exactly the machinery this issue removes; the settled one-substrate rule keeps KV a dumb serving copy.

Revised order: the KV-serving moves for allLatest/availableSignals + summaries hybrid become a new step between reconcile-watermark-awareness and the flip — the per-pass fold stays on until they ship.

[elffjs]

Amending the summaries disposition above: not the KV-overlay hybrid — serve it as (rollup ∪ signals-since-watermark), full-outer-joined by name: count = coalesce(rollup.count, 0) + count(DISTINCT timestamp) over the tail, least/greatest for first/last seen. The watermark splits base rows exactly (rollup counts < W, tail counts >= W), so counts stay exact — including names first seen after the watermark, which the overlay design had no count source for. count(DISTINCT timestamp) self-dedups redeliveries with no NOT-EXISTS probe. The tail leg is subject_bucket = const + constant timestamp bound → one bucket × 1–2 day partitions, riding on a query that already pays the lake floor. Requires the watermark reach the query as a literal (cached in-process; a subquery on ingest_progress may defeat static pruning — assert pruning in the differential test).

Considered and rejected: counts in the KV. A raw counter double-counts under window replay, CAS lost-ack retry, and redelivery (the fold's idempotency comes from newerThan; arithmetic has no equivalent), and exactness needs the NOT-EXISTS probe on the publish path — the cost this issue removes. The one replay-safe variant (increment only when the reading advances the name's latest timestamp) systematically never counts a lagging source's readings on multi-source vehicles — a structural undercount with no self-heal and no coherent merge rule for reconcile. Confirms the one-substrate rule: the KV holds only merge-repairable projections of lake rows, never arithmetic state. This also drops the "KV Entry lastSeen overlay" — the tail leg's max(timestamp) covers freshness, so summaries stays a pure DuckLake query.

Net effect on the plan: the summaries item moves out of the KV-serving step — it ships with the (rollup ∪ tail) fragment alongside the reconcile/bootstrap watermark-awareness work, which needs the same fragment.

[elffjs]

Late-arriving data breaks a pure timestamp watermark. The refresh predicate timestamp >= W folds by event time, but rows become foldable at arrival, and devices buffer offline and upload later. Rows stamped < W that land after the refresh are invisible to every subsequent refresh: their count contribution is lost, and — for a subject whose newest data is in the buffered upload — the rollup's recency stays stale not ≤24h but indefinitely, until something recomputes that subject. The per-pass fold never had this problem (it folds by arrival); the KV is immune for the same reason. But the rollup is the DR substrate, and bounded staleness is its one remaining contract.

Fix, reusing machinery that already exists: at span commit, if any batch row has timestamp < W (one min() over the in-memory batch, no I/O), mark its subject dirty — markDirtyFromBatch + FlushRollup's per-subject recompute already do exactly this for backfill mode. The daily refresh becomes: constant-predicate fold of the settled partition plus a full recompute of the (normally tiny) late set. The summaries tail leg (comment above) shares the blind spot with the same bound — wrong for a late subject only until the next refresh heals it, matching the rollup's own accuracy contract. This slightly widens step 1's scope: the refresh step ships with the late-set hook from the start, or the differential test fails on the first buffered upload it replays.

Correction to "Design" §3: reconcile cannot repair the KV backwards — mergeFromRollup folds through FoldValue, which is strictly newer-wins under CAS, so day-stale rollup rows lose. The real hazard is the completeness proof: a successful reconcile certifies "every lake subject has a key," which is what lets readers serve authoritative "no data" on a miss — but a subject whose first-ever reading arrived after W is invisible to a rollup-only scan, so a reconcile after a lost publish could certify a bucket with a real hole. Same prescribed fix (scan rollup ∪ tail), different test: manufacture a first-reading-after-W subject, lose its publish, reconcile, and assert the negative path doesn't answer "no data."

Two small ones for the record: W is pinned to the UTC-midnight partition boundary, never "time of refresh," so the rollup/tail split always lands on a partition edge (named constant + test, not convention). And shadowCompareLatest uses the rollup as its baseline, so post-flip shadow mode degrades into permanent kv_newer noise — harmless rollout scaffolding, but it goes on the step-5 cleanup list.

[elffjs]

Done, deployed, and measured. All four migration steps (plus the step-5 cleanup) shipped between 2026-08-07 and 2026-08-10: #56 (watermark + shadow refresh + diff), #59 (seed hardening after the inlined-reader crash), #60 (reconcile/bootstrap tail replay), #61 (reader moves: allLatest/availableSignals → KV, summaries → the exact (rollup ∪ tail) union), #63 (the flip: mode on + verified promote), #64 (per-refresh cardinality probe), #65 (fold deleted, shadow scaffolding retired). Final deploy: dimo-node#131, image c67a18f, MATERIALIZER_MAX_SNAPSHOT_SPAN back to 16.

Measured outcome vs the problem statement:

• Span cost at business hours: 275–515s → 6–19s (the residual mild day-dependence is the anti-join dedup probe, as expected). Sawtooth flat; decode ceiling no longer depends on time of day; the dimo-node#123 span stopgap is retired.
• signals_latest churn: ~288 rewrite cycles/day → 1 bounded fold/day at the ~03:30 UTC trough (87–237s observed), plus late-set heals (10–89 subjects/day of buffered uploads — the mechanism added in the comment thread, earning its keep daily).
• Read serving: signalsLatest/signalsSnapshot/availableSignals sub-ms off the KV (467 shadow comparisons, zero mismatch, gated the serve flip); dataSummary exact via the union (tail-born names included — the resolution of the summaries disposition above). End-to-end freshness unchanged: ~89% of readings queryable within 5 min of event time, bounded by din's WAL flush timer, not dq.

What the migration's own evidence caught along the way: the shadow diff refused to read zero, and the cardinality probe (#62) showed why — the live rollup held 823,520 visible rows over 7,666 keys (~266 duplicate copies per active key, growing ~180k/day). Root cause per the evidence: the per-pass fold's DELETE racing din's rewrite_data_files compaction (din's "attempting to delete from table with index" conflict is the same collision from the other side); to be filed upstream against ducklake. The #63 promote discarded the corruption wholesale, and the probe has held rows == keys through every refresh and boot since — the once-daily fold removes the race window that produced it.

Phase 2 (dropping the table entirely once the KV has solo-serving history) remains future work as scoped above; events_latest likewise. Closing.