perf(dq): #1c snapshot pagination + #5b incremental rollup (chaos-proven)

internal/materializer/materializer.go

@@ -153,10 +153,11 @@ func (r *Runner) Run(ctx context.Context) error {
 			failures.record(true, time.Now()) // any success clears the failure streak
 			triedSessionRecycle = false       // a healthy pass proves the session pool recovered
 			if processed > 0 {
-				// Still draining. signals_latest is maintained off this path
-				// (FlushRollup), so a long catch-up doesn't block the writer; bound the
-				// view's staleness with a periodic flush (steady state only — backfill
-				// defers to the single catch-up flush so the drain runs flat-out).
+				// Still draining. signals_latest is maintained INCREMENTALLY at commit
+				// (#5b), so FlushRollup here only recomputes the events_latest rollup and
+				// any backfill-dirtied signal subjects — cheap in steady state. Interval-
+				// gated so a long catch-up doesn't stall the drain (backfill defers to the
+				// single catch-up flush so the drain runs flat-out).
 				if !r.cfg.BackfillMode {
 					r.maybeFlushRollup(ctx, &lastRollup)
 				}

internal/materializer/metrics.go

@@ -167,6 +167,13 @@ func observeLakeLag(events []cloudevent.RawEvent) {
 			oldest = ts
 		}
 	}
+	observeLakeLagAt(oldest)
+}
+
+// observeLakeLagAt sets the decode-lag gauge from the already-computed oldest pending
+// event time (the windowed decode path tracks the running min as it pages, so it never
+// needs the whole delta resident just to measure lag). A zero time means caught up.
+func observeLakeLagAt(oldest time.Time) {
 	if oldest.IsZero() {
 		lagSeconds.WithLabelValues(lakeMetricType).Set(0)
 		return

internal/service/duck/events.go

@@ -157,7 +157,22 @@ func (q *Queries) GetEventCountsForRanges(ctx context.Context, subject string, r
 // falling back to the base scan until the rollup table exists (getEventSummariesLake).
 func (q *Queries) GetEventSummaries(ctx context.Context, subject string) ([]*qtypes.EventSummary, error) {
 	if q.eventsRollupAvailable(ctx) {
-		return q.getEventSummariesRollup(ctx, subject)
+		summaries, err := q.getEventSummariesRollup(ctx, subject)
+		if err != nil {
+			return nil, err
+		}
+		if len(summaries) > 0 {
+			return summaries, nil
+		}
+		// An empty rollup for this subject is ambiguous: the vehicle genuinely has no
+		// events, OR events_latest hasn't been populated for it yet — the query pod
+		// (a separate release) can observe the table existing but empty during the
+		// one-time first-create migration rebuild, or for a brand-new subject before the
+		// next FlushEventRollup. Falling back to the live base scan makes both cases
+		// correct (a genuinely-eventless vehicle just does one extra, bucket-pruned,
+		// empty scan) rather than transiently reporting zero events for a vehicle that
+		// has them. The rollup fast-path still serves every populated subject.
+		return q.getEventSummariesLake(ctx, subject)
 	}
 	return q.getEventSummariesLake(ctx, subject)
 }

internal/service/duck/lake_latest.go

@@ -247,15 +247,17 @@ func (q *Queries) getSignalSummariesLake(ctx context.Context, subject string, fi
 }
 
 // locationGapFillLookback bounds how far before a requested timestamp LocationAt /
-// LocationsAt will reach for the nearest non-origin fix; a fix older than this is
-// treated as "no fix" and the caller substitutes the (0,0) no-data sentinel. Without
-// a floor a GPS-sparse or fix-less vehicle forces a full reverse scan of its entire
-// retained subject_bucket partition on every point lookup (finding #8). 90d is
-// generous enough that a real trip's prior fix is virtually always inside it, while
-// capping the deep scan and letting day-partition pruning drop older files. LocationAt
-// and LocationsAt MUST share this floor so the batched path stays exactly equivalent
-// to the per-point path.
-const locationGapFillLookback = 90 * 24 * time.Hour
+// LocationsAt will reach for the nearest non-origin fix. This floor only exists to give
+// DuckLake a finite lower bound for the reverse scan (so day-partition pruning has a stop);
+// it is deliberately sized WELL BEYOND any realistic decoded retention so it NEVER drops a
+// fix that is still in the retained base. An earlier 90d value was a behavior regression vs
+// the pre-#8 unbounded LocationAt: a vehicle whose last GPS fix was, say, 120 days ago (but
+// still within LAKE_DECODED_RETENTION, default 8760h/~1y) would get (0,0) instead of its
+// last real coordinate. The base is already retention-pruned, so a floor >= retention costs
+// nothing (there is no data below it) yet preserves the old semantics. Must stay >=
+// LAKE_DECODED_RETENTION. LocationAt and LocationsAt MUST share it so the batched path stays
+// exactly equivalent to the per-point path.
+const locationGapFillLookback = 10 * 365 * 24 * time.Hour
 
 // LocationAt returns the nearest non-origin currentLocationCoordinates fix at or
 // before ts — a point lookup that reaches back up to locationGapFillLookback before

internal/service/duck/segments_source_test.go

@@ -469,8 +469,9 @@ func TestLakeQueries_LocationsAt_MatchesLocationAt(t *testing.T) {
 	// (aaa) must win the tie-break in both paths.
 	insertLoc("zzz", at(30), 99.0, 99.0, 9.9)
 	insertLoc("aaa", at(30), 42.0, -77.0, 1.3)
-	// A fix 180d before base — older than the 90d floor of the far probe below, so both
-	// paths must return nil for that probe (lookback-floor parity).
+	// A fix 180d before base, 100d before the probe below — well within the (retention-
+	// sized) gap-fill lookback, so BOTH paths must return it (the batched path equals the
+	// per-point path; the equivalence is derived from LocationAt at assert time).
 	staleProbe := base.Add(-80 * 24 * time.Hour)
 	insertLoc("stale", base.Add(-180*24*time.Hour), 10.0, 10.0, 5.0)
 
@@ -481,7 +482,7 @@ func TestLakeQueries_LocationsAt_MatchesLocationAt(t *testing.T) {
 		at(25),     // → l2 (origin (0,0) skipped)
 		at(35),     // → aaa (tie-break at ts=30)
 		at(10),     // exactly on l2 (>= boundary)
-		staleProbe, // only the 180d-old stale fix is prior → beyond 90d floor → nil
+		staleProbe, // the 180d-old stale fix is prior and within the lookback → returned by both
 	}
 
 	batched, err := q.LocationsAt(ctx, subject, probes)

tests/VERIFY-CAMPAIGN.md

@@ -0,0 +1,19 @@
+# Verification campaign — #1c pagination + #5b incremental rollup (2026-07-07)
+
+10 adversarial loops. Contract every loop: incremental `signals_latest` == full `RecomputeRollup`, and base rows exactly-once. Result: **10/10 PASS, zero product defects.**
+(Loop 1's first run failed on a harness assumption — location signals are not NULL `value_number` — the differential itself passed; product correct.)
+
+| # | Vector | Where |
+|---|--------|-------|
+| 1 | location signal (value_number/loc fold) | TestVerify01 |
+| 2 | intermittent location — loc_ts stays newest despite later older fix | TestVerify02 |
+| 3 | fat single snapshot, many (subject,name) | TestVerify03 |
+| 4 | pagination driven by the BYTE budget (row cap off) | TestVerify04 |
+| 5 | crash at the LAST intermediate window → restart exact | TestVerify05 |
+| 6 | span mixing decodable + non-decodable (wrong-chain) rows | TestVerify06 |
+| 7 | 100-day out-of-order arrival | TestVerify07 |
+| 8 | RecomputeRollup interleaved with incremental folds (self-healing) | TestVerify08 |
+| 9 | THREE concurrent writers, paginated fat snapshot (real PG) | TestVerify09_PG |
+| 10 | one writer crashes mid-span + supervisor restart, two race it (real PG) | TestVerify10_PG |
+
+Embedded loops run in `go test ./tests/`; PG loops need `PG_CATALOG_DSN`. `-race` clean.

tests/ducklake_event_rollup_test.go

@@ -7,6 +7,7 @@ package tests
 
 import (
 	"context"
+	"database/sql"
 	"encoding/json"
 	"fmt"
 	"testing"
@@ -209,3 +210,67 @@ func TestDuckLake_EventRollup_FirstCreateBackfill(t *testing.T) {
 	require.Contains(t, got, evEngineBlock, "dormant subject must be backfilled into events_latest on first create")
 	assert.EqualValues(t, 2, got[evEngineBlock].count)
 }
+
+// TestGetEventSummaries_FallsBackWhenRollupEmpty pins the #5a serving fix: while
+// events_latest exists but isn't yet populated for a subject (first-create migration
+// window, or a brand-new subject before the next flush), GetEventSummaries must fall back
+// to the live base scan instead of returning an empty summary.
+func TestGetEventSummaries_FallsBackWhenRollupEmpty(t *testing.T) {
+	ctx := context.Background()
+	svc := newLakeService(t, t.TempDir())
+	db := svc.DB()
+	subject := fmt.Sprintf("did:erc721:137:%s:14", vehicleNFT.Hex())
+	day := time.Now().UTC().AddDate(0, 0, -2).Truncate(24 * time.Hour)
+	seedRawEvent(t, svc, deviceEvents("fb-1", subject, day.Add(time.Hour), eventAt(evEngineBlock, day.Add(time.Hour))))
+	runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).
+		WithDuckLake(mustMat(t, ctx, db))
+	require.Equal(t, 1, drainRunner(t, ctx, runner))
+
+	// Simulate the not-yet-populated window: the table exists but has no rows for anyone.
+	_, err := db.ExecContext(ctx, "DELETE FROM lake.events_latest")
+	require.NoError(t, err)
+
+	q := duck.NewLakeQueries(svc)
+	got := eventSummaryByName(t, ctx, q, subject)
+	require.Contains(t, got, evEngineBlock, "must fall back to the base scan when the rollup is empty for the subject")
+	assert.EqualValues(t, 1, got[evEngineBlock].count)
+}
+
+// TestDuckLake_EventRollup_CrashMidMigrationRestart pins the Q3b crash-safe marker: if a
+// pod is killed after creating events_latest but before its first-create rebuild finishes,
+// the RESTART must still detect the incomplete rollup (events present, rollup empty) and
+// rebuild — not silently skip it because the table now exists.
+func TestDuckLake_EventRollup_CrashMidMigrationRestart(t *testing.T) {
+	ctx := context.Background()
+	svc := newLakeService(t, t.TempDir())
+	db := svc.DB()
+	subject := fmt.Sprintf("did:erc721:137:%s:15", vehicleNFT.Hex())
+	day := time.Now().UTC().AddDate(0, 0, -2).Truncate(24 * time.Hour)
+	seedRawEvent(t, svc, deviceEvents("cm-1", subject, day.Add(time.Hour), eventAt(evEngineBlock, day.Add(time.Hour))))
+	runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).
+		WithDuckLake(mustMat(t, ctx, db))
+	require.Equal(t, 1, drainRunner(t, ctx, runner))
+
+	// Crash mid-migration: events_latest exists (created by a prior boot) but is empty
+	// (the rebuild never finished). lake.events still holds the dormant subject's events.
+	_, err := db.ExecContext(ctx, "DELETE FROM lake.events_latest")
+	require.NoError(t, err)
+
+	// Restart: a fresh materializer over the same catalog must re-detect the incomplete
+	// rollup from DB state (not an in-memory flag) and rebuild on the next flush.
+	mat2 := mustMat(t, ctx, db)
+	require.NoError(t, mat2.FlushEventRollup(ctx))
+
+	// Read directly from the rollup (not via the fallback) to prove it was actually repopulated.
+	var cnt int
+	require.NoError(t, db.QueryRowContext(ctx,
+		"SELECT count(*) FROM lake.events_latest WHERE subject = ?", subject).Scan(&cnt))
+	assert.Positive(t, cnt, "restart after a crash mid-migration must rebuild events_latest for dormant subjects")
+}
+
+func mustMat(t *testing.T, ctx context.Context, db *sql.DB) *materializer.DuckLakeMaterializer {
+	t.Helper()
+	mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop())
+	require.NoError(t, err)
+	return mat
+}