diff --git a/internal/materializer/ducklake.go b/internal/materializer/ducklake.go index 8e01379..f12d023 100644 --- a/internal/materializer/ducklake.go +++ b/internal/materializer/ducklake.go @@ -104,6 +104,25 @@ type DuckLakeMaterializer struct { // maxDirtySubjects is the overflow cap (defaultMaxDirtySubjects). Shared by both // dirty sets. maxDirtySubjects int + + // windowByteBudget / maxRowsPerWindow bound the resident working set of a SINGLE + // snapshot's decode+write (finding #1c). maxSnapshotSpan count-bounds the + // MULTI-snapshot backlog, but a single fat snapshot (span can't drop below 1) with + // a large blob fan-out would still materialize whole and OOM the writer. Instead + // the (cursor, to] insert delta is read+decoded+written in row-key-ordered WINDOWS: + // intermediate windows are written idempotently WITHOUT advancing the cursor, and + // only the FINAL window's transaction advances it — so "cursor advanced ⟺ every + // window durable" still holds and a crash mid-span re-reads from the un-advanced + // cursor, the anti-join collapsing already-written windows. A window is flushed once + // its resident raw-payload bytes reach windowByteBudget OR its row count reaches + // maxRowsPerWindow. <= 0 disables that bound (the other still applies). + windowByteBudget int64 + maxRowsPerWindow int + // windowCommitHook is a TEST-ONLY seam: called with the 0-based index after each + // INTERMEDIATE window commits (never for the final cursor-advancing commit). Returning + // an error aborts the pass after that window is durable but before the cursor advances, + // deterministically reproducing a crash mid-span. nil in production. + windowCommitHook func(windowIndex int) error } // WithBackfillMode toggles backfill tuning (skip the cross-batch dedup anti-join; @@ -169,6 +188,8 @@ func NewDuckLakeMaterializer(ctx context.Context, db *sql.DB, log zerolog.Logger dirtySubjects: map[string]struct{}{}, dirtyEventSubjects: map[string]struct{}{}, maxDirtySubjects: defaultMaxDirtySubjects, + windowByteBudget: defaultWindowByteBudget, + maxRowsPerWindow: defaultWindowMaxRows, } if err := m.ensureSchema(ctx); err != nil { return nil, err @@ -183,6 +204,41 @@ func (m *DuckLakeMaterializer) WithMaxSnapshotSpan(n int64) *DuckLakeMaterialize return m } +// defaultWindowByteBudget / defaultWindowMaxRows bound one intra-snapshot window's +// resident working set (finding #1c). 64 MiB of raw payloads keeps a window well inside +// the pod's non-DuckDB headroom even under a large blob fan-out; the row cap bounds the +// count when payloads are tiny (so the decoded-row and temp-parquet working set stays +// bounded too). windowReadChunk is how many change-feed rows are read per query while +// filling a window. +const ( + defaultWindowByteBudget = 64 << 20 // 64 MiB + defaultWindowMaxRows = 100_000 + windowReadChunk = 512 +) + +// WithWindowByteBudget overrides the per-window resident-byte bound (finding #1c). +// A non-positive n disables the byte bound (the row cap still applies). Returns m. +func (m *DuckLakeMaterializer) WithWindowByteBudget(n int64) *DuckLakeMaterializer { + m.windowByteBudget = n + return m +} + +// WithMaxRowsPerWindow overrides the per-window row cap (finding #1c). A non-positive n +// disables the row bound (the byte budget still applies). Tests use a tiny value to force +// multi-window pagination of a small snapshot. Returns m. +func (m *DuckLakeMaterializer) WithMaxRowsPerWindow(n int) *DuckLakeMaterializer { + m.maxRowsPerWindow = n + return m +} + +// WithWindowCommitHook installs the TEST-ONLY crash seam (see the field doc): the hook +// runs after each intermediate window commits, and a returned error aborts the pass +// before the cursor advances. Returns m. +func (m *DuckLakeMaterializer) WithWindowCommitHook(fn func(windowIndex int) error) *DuckLakeMaterializer { + m.windowCommitHook = fn + return m +} + // WithBlobStore configures blob-payload resolution: when a raw_events row has // no inline data but a data_index_key under BlobKeyPrefix, the materializer // downloads the payload from bucket via getter before decoding. Mirrors the @@ -277,14 +333,24 @@ func (m *DuckLakeMaterializer) ensureSchema(ctx context.Context) error { m.rollupFullRebuild = true } } - // events_latest first-create over a PRE-EXISTING lake.events base is a migration - // (finding #5a): an existing catalog getting the events rollup for the first time - // must backfill every already-decoded subject, or dormant vehicles (no new events) - // would read an empty summary. Escalate the first FlushEventRollup to a full - // rebuild. A brand-new catalog creates events + events_latest in the same pass - // (exists["events"] is false), so this does NOT fire there — no 256-bucket - // snapshot churn on a fresh boot; the per-batch dirty tracking fills it instead. - if !exists["events_latest"] && exists["events"] { + // events_latest completeness marker (crash-safe, mirrors the loc_ts pattern above). + // A first-create migration over a PRE-EXISTING lake.events base must backfill every + // already-decoded subject, or dormant vehicles (no new events) read an empty summary. + // Deriving that from PRE-DDL existence (an in-memory flag) is NOT crash-safe: a pod + // killed after events_latest is created but before RecomputeEventRollup finishes would, + // on restart, see the table exists and silently skip the rebuild — stranding dormant + // subjects forever (Q3b). Instead re-derive it from DB STATE each boot: "the event base + // is non-empty but the rollup has NO rows" precisely means "rollup incomplete", and it + // survives any crash mid-rebuild. A truly fresh catalog (events not yet flowing) has an + // empty base, so this does not fire; once the base has data, one rebuild populates the + // rollup and it never fires again. (The per-subject GetEventSummaries fallback keeps + // reads correct during the window.) + var eventsPending bool + if err := m.db.QueryRowContext(ctx, + `SELECT EXISTS (SELECT 1 FROM lake.events) AND NOT EXISTS (SELECT 1 FROM lake.events_latest)`).Scan(&eventsPending); err != nil { + return fmt.Errorf("checking events_latest backfill marker: %w", err) + } + if eventsPending { m.eventRollupFullRebuild = true } return nil @@ -512,40 +578,39 @@ func (m *DuckLakeMaterializer) processChunk(ctx context.Context, cur, head int64 to = cur + m.maxSnapshotSpan } - events, err := m.readDelta(ctx, cur, to) + // Decode+write the (cur, to] insert delta in byte/row-bounded windows so a single + // fat snapshot can't materialize whole and OOM the writer (#1c). Only the final + // window advances the cursor, so a crash mid-span re-reads from cur and the + // anti-join collapses already-written windows — exactly-once preserved. + res, err := m.decodeAndWriteSpan(ctx, cur, to, dec) if err != nil { - // Any feed-read failure might mean din's maintenance expired the cursor - // range. Decide on retention (the oldest retained snapshot), not on the - // error text — so a real expiry with unmatched wording can't wedge us - // forever, and a transient error that merely looks like expiry can't make - // us skip retained data. + if errors.Is(err, errSnapshotMoved) { + return 0, cur, true, nil // another decoder won this range; retry next pass + } + // A feed-read failure might mean din's maintenance expired the cursor range. + // Decide on retention (the oldest retained snapshot), not on the error text — + // so a real expiry with unmatched wording can't wedge us forever, and a + // transient error that merely looks like expiry can't make us skip retained data. if n, handled, rerr := m.maybeRecoverExpired(ctx, cur, err); handled { return n, cur, true, rerr } return 0, cur, true, err } - if len(events) > 0 { - observeLakeLag(events) // decode lag = age of the oldest pending event - decoded := dec.decodeEvents(ctx, events) - if cerr := m.commit(ctx, decoded, cur, to); cerr != nil { - if errors.Is(cerr, errSnapshotMoved) { - return 0, cur, true, nil // another decoder won this range; retry next pass - } - return 0, cur, true, cerr - } - // A batch committed: feed the freshness/throughput alerts (CHD-12). + if res.rawRows > 0 { + observeLakeLagAt(res.oldest) // decode lag = age of the oldest pending event + // A span committed: feed the freshness/throughput alerts (CHD-12). batchesTotal.WithLabelValues(lakeMetricType).Inc() - rowsTotal.WithLabelValues("signals").Add(float64(decoded.signalCount)) - rowsTotal.WithLabelValues("events").Add(float64(decoded.eventCount)) - errorsTotal.Add(float64(decoded.errorCount)) + rowsTotal.WithLabelValues("signals").Add(float64(res.signalRows)) + rowsTotal.WithLabelValues("events").Add(float64(res.eventRows)) + errorsTotal.Add(float64(res.errRows)) cursorSnapshotID.Set(float64(to)) - // Report progress to din's snapshot-expiry floor. Throttled: the batch is + // Report progress to din's snapshot-expiry floor. Throttled: the span is // already durable, and din only needs the floor within its retention window // — a lagging report just holds expiry back slightly (conservative, never // unsafe), so it needn't be a catalog txn on every batch. m.maybeReportProgress(ctx, to) - return len(events), to, true, nil + return res.rawRows, to, true, nil } // Empty span. If it reached head, the decoder is caught up: head advanced only @@ -783,12 +848,207 @@ func (m *DuckLakeMaterializer) headSnapshot(ctx context.Context) (int64, error) } // readDelta reconstructs the raw events inserted in (from, to]. -func (m *DuckLakeMaterializer) readDelta(ctx context.Context, from, to int64) ([]cloudevent.RawEvent, error) { +// rawRowKey is the unique, totally-ordered key used to paginate the (cur, to] insert +// change feed: (subject, time, id). id (the cloud_event_id) is globally unique, so the +// triple is a strict total order — a `> after` cursor + ORDER BY partitions the feed +// into windows with no row skipped or repeated. +type rawRowKey struct { + subject string + ts time.Time + id string +} + +// spanCounts accumulates what a span (or window) processed, for metrics and the +// caught-up/empty-span decision. rawRows counts raw_events rows READ (before any poison +// drop) so the cursor and lag stay correct even when a payload is dropped. +type spanCounts struct { + rawRows int + signalRows int + eventRows int + errRows int + oldest time.Time // oldest raw event Time seen (for decode-lag) +} + +// readDeltaWindow reads up to `limit` raw_events INSERT rows from the (from, to] change +// feed, ordered by the unique row key and strictly after `after` when hasAfter, then +// resolves their blob payloads. It returns the resolved rows (poison rows dropped), the +// key of the LAST row read (captured from the raw scan, before any drop, so pagination +// never skips a row whose payload was dropped), and got = the number of rows actually +// read (0 ⇒ the feed is exhausted past `after`). +func (m *DuckLakeMaterializer) readDeltaWindow(ctx context.Context, from, to int64, after rawRowKey, hasAfter bool, limit int) (events []cloudevent.RawEvent, last rawRowKey, got int, err error) { + where := "" + var args []any + if hasAfter { + where = ` AND (subject, "time", id) > (?, ?, ?)` + args = []any{after.subject, after.ts.UTC(), after.id} + } q := fmt.Sprintf( - "SELECT %s FROM ducklake_table_changes('lake', 'main', 'raw_events', %d, %d) WHERE change_type = 'insert'", - duck.RawEventColumns, from+1, to) - // Don't classify a query error here — RunOnce decides expired-vs-transient on retention. - return m.scanAndResolveRaw(ctx, q, "reading raw_events delta") + `SELECT %s FROM ducklake_table_changes('lake', 'main', 'raw_events', %d, %d) `+ + `WHERE change_type = 'insert'%s ORDER BY subject, "time", id LIMIT %d`, + duck.RawEventColumns, from+1, to, where, limit) + rows, qerr := m.db.QueryContext(ctx, q, args...) + if qerr != nil { + // Don't classify here — processChunk decides expired-vs-transient on retention. + return nil, rawRowKey{}, 0, fmt.Errorf("reading raw_events delta window: %w", qerr) + } + defer rows.Close() //nolint:errcheck + + var out []cloudevent.RawEvent + var blobKeys []string + for rows.Next() { + ev, blobKey, serr := scanRawEvent(rows) + if serr != nil { + return nil, rawRowKey{}, 0, serr + } + out = append(out, ev) + blobKeys = append(blobKeys, blobKey) + last = rawRowKey{subject: ev.Subject, ts: ev.Time, id: ev.ID} + got++ + } + if rerr := rows.Err(); rerr != nil { + return nil, rawRowKey{}, 0, rerr + } + resolved, berr := m.resolveBlobs(ctx, out, blobKeys) + if berr != nil { + return nil, rawRowKey{}, 0, berr + } + return resolved, last, got, nil +} + +// rawResidentBytes estimates the resident payload bytes of a batch of raw events (the +// working set the window byte-budget bounds — blob fan-out is the OOM risk in #1c). +func rawResidentBytes(events []cloudevent.RawEvent) int64 { + var n int64 + for i := range events { + n += int64(len(events[i].Data)) + int64(len(events[i].DataBase64)) + } + return n +} + +// appendDecoded folds src into dst (rows and counts). +func appendDecoded(dst, src *decodedBatch) { + dst.signals = append(dst.signals, src.signals...) + dst.events = append(dst.events, src.events...) + dst.signalCount += src.signalCount + dst.eventCount += src.eventCount + dst.errorCount += src.errorCount +} + +// earlier returns the earlier of two times, ignoring the zero value. +func earlier(a, b time.Time) time.Time { + switch { + case a.IsZero(): + return b + case b.IsZero(): + return a + case b.Before(a): + return b + default: + return a + } +} + +// nextWindow accumulates one byte/row-bounded window of the (from, to] delta, advancing +// *after past the rows it consumed. It reads the feed in windowReadChunk-sized pages and +// decodes each page, stopping once the window's resident bytes reach windowByteBudget or +// its row count reaches maxRowsPerWindow (or the feed is exhausted). c.rawRows == 0 means +// no rows remain past *after. +func (m *DuckLakeMaterializer) nextWindow(ctx context.Context, from, to int64, after *rawRowKey, hasAfter *bool, dec eventDecoder) (*decodedBatch, spanCounts, error) { + win := &decodedBatch{} + var c spanCounts + var winBytes int64 + for { + readLimit := windowReadChunk + if m.maxRowsPerWindow > 0 { + rem := m.maxRowsPerWindow - c.rawRows + if rem <= 0 { + break // row cap reached: window is full + } + if rem < readLimit { + readLimit = rem + } + } + resolved, last, got, err := m.readDeltaWindow(ctx, from, to, *after, *hasAfter, readLimit) + if err != nil { + return nil, c, err + } + if got == 0 { + break // feed exhausted past `after` + } + *after = last + *hasAfter = true + c.rawRows += got + winBytes += rawResidentBytes(resolved) + for i := range resolved { + c.oldest = earlier(c.oldest, resolved[i].Time) + } + d := dec.decodeEvents(ctx, resolved) + appendDecoded(win, d) + c.signalRows += d.signalCount + c.eventRows += d.eventCount + c.errRows += d.errorCount + if got < readLimit { + break // fewer rows than asked ⇒ feed exhausted (tail window) + } + if m.windowByteBudget > 0 && winBytes >= m.windowByteBudget { + break // byte budget reached: window is full + } + } + return win, c, nil +} + +// decodeAndWriteSpan decodes and writes the (from, to] insert delta in byte/row-bounded +// windows (finding #1c). Every window but the last is written idempotently WITHOUT +// advancing the cursor; the final window's commit advances the cursor from→to coupled to +// its insert, exactly as the pre-#1c monolithic commit did. So the invariant holds: +// cursor advanced ⟺ every window durable. A crash mid-span leaves the cursor at `from`; +// restart re-reads the whole span and the cloud_event_id anti-join collapses the windows +// that already landed. Returns 0 rawRows (and does NOT touch the cursor) for an empty +// span, leaving the caller to run the empty-span/caught-up handling. +func (m *DuckLakeMaterializer) decodeAndWriteSpan(ctx context.Context, from, to int64, dec eventDecoder) (spanCounts, error) { + var total spanCounts + var after rawRowKey + hasAfter := false + var pending *decodedBatch + havePending := false + windowIdx := 0 + for { + win, c, err := m.nextWindow(ctx, from, to, &after, &hasAfter, dec) + if err != nil { + return total, err + } + if c.rawRows == 0 { + break // no more rows: `pending`, if any, is the final window + } + total.rawRows += c.rawRows + total.signalRows += c.signalRows + total.eventRows += c.eventRows + total.errRows += c.errRows + total.oldest = earlier(total.oldest, c.oldest) + if havePending { + // The previous window is now known NOT to be the last, so write it as an + // intermediate window (idempotent, no cursor advance). + if err := m.writeWindow(ctx, pending); err != nil { + return total, err + } + if m.windowCommitHook != nil { + if err := m.windowCommitHook(windowIdx); err != nil { + return total, err + } + } + windowIdx++ + } + pending = win + havePending = true + } + if !havePending { + return total, nil // empty span + } + // The final window advances the cursor, coupled to its insert. + if err := m.commit(ctx, pending, from, to); err != nil { + return total, err + } + return total, nil } // scanAndResolveRaw runs a raw_events SELECT (projected with duck.RawEventColumns so @@ -1131,27 +1391,10 @@ func (m *DuckLakeMaterializer) commit(ctx context.Context, dec *decodedBatch, fr } defer func() { _ = tx.Rollback() }() - if len(dec.signals) > 0 { - tmp, err := writeTempParquet(m.tempDir, writeSignalParquet, dec.signals) - if err != nil { - return err - } - cleanup = append(cleanup, tmp) - tsMin, tsMax := timeRange(dec.signals, func(r SignalRow) time.Time { return r.Timestamp }) - if _, err := tx.ExecContext(ctx, antiJoinInsert("lake.signals", tmp, tsMin, tsMax, m.backfillMode)); err != nil { - return fmt.Errorf("insert signals: %w", err) - } - } - if len(dec.events) > 0 { - tmp, err := writeTempParquet(m.tempDir, writeEventParquet, dec.events) - if err != nil { - return err - } - cleanup = append(cleanup, tmp) - tsMin, tsMax := timeRange(dec.events, func(r EventRow) time.Time { return r.Timestamp }) - if _, err := tx.ExecContext(ctx, antiJoinInsert("lake.events", tmp, tsMin, tsMax, m.backfillMode)); err != nil { - return fmt.Errorf("insert events: %w", err) - } + files, err := m.insertDecodedSteady(ctx, tx, dec) + cleanup = append(cleanup, files...) + if err != nil { + return err } // NOTE: the lake.signals_latest rollup is no longer maintained here. It is a @@ -1192,26 +1435,80 @@ func (m *DuckLakeMaterializer) commit(ctx context.Context, dec *decodedBatch, fr m.log.Warn().Err(err).Int64("to", to). Msg("commit returned an error but the cursor advanced to the target; treating as committed (lost ack)") } - // The base rows are durable. Mark the subjects this batch wrote so the - // decoupled FlushRollup recomputes only their rollup rows (B2). Done after + // The base rows are durable. Mark the subjects this batch wrote so the decoupled + // FlushRollup/FlushEventRollup recompute only their rollup rows (B2/#5a). Done after // commit so a rolled-back batch never dirties the rollup. - for i := range dec.signals { - m.dirtySubjects[dec.signals[i].Subject] = struct{}{} + m.markDirtyFromBatch(dec) + return nil +} + +// insertDecodedSteady stages dec's signals+events as temp parquet and issues the +// idempotent (cloud_event_id anti-join) INSERTs into lake.signals/events within tx, +// clamping the dedup probe to dedupProbeFloor (the steady-state live-decode window — +// redeliveries are recent). Returns the temp-file paths the caller must remove after +// the transaction completes. Shared by commit (the cursor-advancing final window) and +// writeWindow (a non-cursor-advancing intermediate window); backfillWrite keeps its own +// unclamped probe for arbitrarily-old data. +func (m *DuckLakeMaterializer) insertDecodedSteady(ctx context.Context, tx *sql.Tx, dec *decodedBatch) ([]string, error) { + var cleanup []string + if len(dec.signals) > 0 { + tmp, err := writeTempParquet(m.tempDir, writeSignalParquet, dec.signals) + if err != nil { + return cleanup, err + } + cleanup = append(cleanup, tmp) + tsMin, tsMax := timeRange(dec.signals, func(r SignalRow) time.Time { return r.Timestamp }) + // #5b: steady-state lake.signals_latest is maintained INCREMENTALLY here + // (O(batch), not an O(history) recompute per flush). The count delta must be + // captured BEFORE the base insert — afterwards the batch rows are in the base and + // the NOT-EXISTS probe finds them, yielding delta 0 (that is exactly what makes a + // replayed window idempotent). Backfill (bulk/arbitrarily-old) skips the fold and + // defers to the end-of-catch-up recompute (markDirtyFromBatch marks it dirty). + if !m.backfillMode { + if err := m.captureRollupDelta(ctx, tx, tmp); err != nil { + return cleanup, err + } + } + if _, err := tx.ExecContext(ctx, antiJoinInsert("lake.signals", tmp, tsMin, tsMax, m.backfillMode)); err != nil { + return cleanup, fmt.Errorf("insert signals: %w", err) + } + if !m.backfillMode { + if err := m.foldSignalsRollup(ctx, tx, tmp); err != nil { + return cleanup, err + } + } + } + if len(dec.events) > 0 { + tmp, err := writeTempParquet(m.tempDir, writeEventParquet, dec.events) + if err != nil { + return cleanup, err + } + cleanup = append(cleanup, tmp) + tsMin, tsMax := timeRange(dec.events, func(r EventRow) time.Time { return r.Timestamp }) + if _, err := tx.ExecContext(ctx, antiJoinInsert("lake.events", tmp, tsMin, tsMax, m.backfillMode)); err != nil { + return cleanup, fmt.Errorf("insert events: %w", err) + } + } + return cleanup, nil +} + +// markDirtyFromBatch records the subjects dec wrote so the decoupled rollups refresh +// only their rows, escalating to a full rebuild if a dirty set overflows its cap (a +// fleet-wide catch-up would otherwise grow the maps unbounded — see the field docs). +// signals_latest is now maintained incrementally at commit time (#5b), so signals are +// dirtied ONLY under backfillMode (the deferred bulk catch-up recomputes them at the +// end); events_latest still uses the decoupled recompute, so events are always dirtied. +// Single-writer: mutated only on the decode-loop goroutine. +func (m *DuckLakeMaterializer) markDirtyFromBatch(dec *decodedBatch) { + if m.backfillMode { + for i := range dec.signals { + m.dirtySubjects[dec.signals[i].Subject] = struct{}{} + } + if len(m.dirtySubjects) > m.maxDirtySubjects { + m.dirtySubjects = map[string]struct{}{} + m.rollupFullRebuild = true + } } - // Bound the dirty set: a fleet-wide catch-up (initial backfill defers the - // flush until fully drained) would otherwise grow this map with every - // subject in the fleet — ~1GB+ at 10M vehicles, inside the pod's ~2GB - // non-DuckDB headroom. Past the cap, per-subject tracking is no cheaper - // than a full rebuild anyway, so escalate: clear the map and let the next - // flush run the bucket-chunked, memory-bounded RecomputeRollup. - if len(m.dirtySubjects) > m.maxDirtySubjects { - m.dirtySubjects = map[string]struct{}{} - m.rollupFullRebuild = true - } - // Same dirty-tracking for the events_latest rollup (finding #5a): a batch can - // write events for subjects with no signals (and vice versa), so track them - // independently. Post-commit like the signals set, so a rolled-back batch never - // dirties the rollup. for i := range dec.events { m.dirtyEventSubjects[dec.events[i].Subject] = struct{}{} } @@ -1219,6 +1516,145 @@ func (m *DuckLakeMaterializer) commit(ctx context.Context, dec *decodedBatch, fr m.dirtyEventSubjects = map[string]struct{}{} m.eventRollupFullRebuild = true } +} + +// captureRollupDelta stages, into the per-connection temp table _rollup_delta, the number +// of NEWLY-DISTINCT (subject, name, timestamp) tuples this batch adds — i.e. the exact +// increment to lake.signals_latest.count (#5b). It must run BEFORE the base insert: it +// probes lake.signals for tuples that DON'T already exist, so a redelivery or a +// same-(subject,name,timestamp) collision (which the count must not double) contributes 0, +// and a replayed window (rows already at rest) yields 0 — the idempotency the crash- +// recovery path relies on. The probe matches on the EXACT timestamp (partition-pruned by +// subject_bucket + the day partition), deliberately NOT clamped to the anti-join's 30d +// dedup window: an ancient (>30d) redelivery must still find its existing row so count +// stays equal to a full RecomputeRollup, even though the physical anti-join may re-insert a +// duplicate row (the read-path QUALIFY dedup collapses that, and count must match it). +func (m *DuckLakeMaterializer) captureRollupDelta(ctx context.Context, tx *sql.Tx, sigParquet string) error { + q := fmt.Sprintf(`CREATE OR REPLACE TEMPORARY TABLE _rollup_delta AS +SELECT b.subject, b.name, CAST(count(*) AS BIGINT) AS delta +FROM (SELECT DISTINCT subject, name, subject_bucket, "timestamp" FROM read_parquet(%[1]s)) b +WHERE NOT EXISTS ( + SELECT 1 FROM lake.signals s + WHERE s.subject_bucket = b.subject_bucket AND s.subject = b.subject AND s.name = b.name + AND s."timestamp" = b."timestamp" +) +GROUP BY b.subject, b.name`, sqlLit(sigParquet)) + if _, err := tx.ExecContext(ctx, q); err != nil { + return fmt.Errorf("capture rollup count delta: %w", err) + } + return nil +} + +// foldSignalsRollup folds this batch into lake.signals_latest incrementally, EXACTLY as a +// full recompute would (proven by the differential test), but O(batch) instead of +// O(history) (#5b). It runs AFTER the base insert, inside the same transaction: +// - RECENCY (timestamp, value_*, loc_*, loc_ts) is recomputed from the base but BOUNDED +// to "timestamp >= the row's prior latest" — the new latest is either in this batch +// (newer) or unchanged (the prior row still qualifies), so the bound is exact yet +// prunes every day-partition older than the prior latest. This makes recency +// SELF-HEALING (recomputed from the base each batch), matching the recompute's deduped +// arg_max via ORDER BY timestamp DESC, cloud_event_id ASC. +// - COUNT is prev.count + the captured NOT-EXISTS delta; FIRST_SEEN is min(prev, batch). +// These carry forward (idempotent on replay), and self-heal via the boot rebuild +// (RecomputeRollup / LAKE_REBUILD_ROLLUP_ON_BOOT) if a rollup row is ever lost. +// A (subject,name) with no prior rollup row folds against zero — correct in steady state +// (a newly-seen signal has no prior base); a mass-loss (dropped rollup) is the boot +// rebuild's job, exactly as before. +func (m *DuckLakeMaterializer) foldSignalsRollup(ctx context.Context, tx *sql.Tx, sigParquet string) error { + build := fmt.Sprintf(`CREATE OR REPLACE TEMPORARY TABLE _rollup_new AS +WITH affected AS ( + SELECT subject, name, any_value(subject_bucket) AS subject_bucket, min("timestamp") AS batch_min + FROM read_parquet(%[1]s) GROUP BY subject, name +), +prev AS ( + SELECT l.subject, l.name, l."timestamp" AS prev_ts, l.loc_ts AS prev_loc_ts, l.count AS prev_count, l.first_seen AS prev_first + FROM lake.signals_latest l + WHERE EXISTS (SELECT 1 FROM affected a WHERE a.subject = l.subject AND a.name = l.name) +), +recency AS ( + SELECT s.subject, s.name, s."timestamp" AS ts, s.value_number, s.value_string + FROM lake.signals s + JOIN affected a ON s.subject = a.subject AND s.name = a.name AND s.subject_bucket = a.subject_bucket + LEFT JOIN prev p ON p.subject = s.subject AND p.name = s.name + WHERE s."timestamp" >= coalesce(p.prev_ts, make_timestamp(0)) + QUALIFY row_number() OVER (PARTITION BY s.subject, s.name ORDER BY s."timestamp" DESC, s.cloud_event_id ASC) = 1 +), +locrec AS ( + SELECT s.subject, s.name, s."timestamp" AS loc_ts, s.loc_lat, s.loc_lon, s.loc_hdop, s.loc_heading + FROM lake.signals s + JOIN affected a ON s.subject = a.subject AND s.name = a.name AND s.subject_bucket = a.subject_bucket + LEFT JOIN prev p ON p.subject = s.subject AND p.name = s.name + WHERE (s.loc_lat != 0 OR s.loc_lon != 0) AND s."timestamp" >= coalesce(p.prev_loc_ts, make_timestamp(0)) + QUALIFY row_number() OVER (PARTITION BY s.subject, s.name ORDER BY s."timestamp" DESC, s.cloud_event_id ASC) = 1 +) +SELECT a.subject, a.subject_bucket, a.name, + r.ts AS "timestamp", r.value_number, r.value_string, + coalesce(lr.loc_lat, 0) AS loc_lat, coalesce(lr.loc_lon, 0) AS loc_lon, + coalesce(lr.loc_hdop, 0) AS loc_hdop, coalesce(lr.loc_heading, 0) AS loc_heading, + coalesce(lr.loc_ts, make_timestamp(0)) AS loc_ts, + coalesce(p.prev_count, 0) + coalesce(d.delta, 0) AS count, + LEAST(coalesce(p.prev_first, a.batch_min), a.batch_min) AS first_seen, + r.ts AS last_seen +FROM affected a +JOIN recency r ON r.subject = a.subject AND r.name = a.name +LEFT JOIN locrec lr ON lr.subject = a.subject AND lr.name = a.name +LEFT JOIN prev p ON p.subject = a.subject AND p.name = a.name +LEFT JOIN _rollup_delta d ON d.subject = a.subject AND d.name = a.name`, sqlLit(sigParquet)) + if _, err := tx.ExecContext(ctx, build); err != nil { + return fmt.Errorf("build incremental rollup rows: %w", err) + } + if _, err := tx.ExecContext(ctx, + `DELETE FROM lake.signals_latest WHERE EXISTS (SELECT 1 FROM _rollup_new n WHERE n.subject = lake.signals_latest.subject AND n.name = lake.signals_latest.name)`); err != nil { + return fmt.Errorf("delete superseded rollup rows: %w", err) + } + if _, err := tx.ExecContext(ctx, + `INSERT INTO lake.signals_latest (subject, subject_bucket, name, "timestamp", value_number, value_string, loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen) + SELECT subject, subject_bucket, name, "timestamp", value_number, value_string, loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen FROM _rollup_new`); err != nil { + return fmt.Errorf("insert incremental rollup rows: %w", err) + } + return nil +} + +// writeWindow writes an INTERMEDIATE pagination window (finding #1c): the decoded rows +// go into lake.signals/events in their own transaction that does NOT advance the ingest +// cursor. It is idempotent (the cloud_event_id anti-join) and cursor-independent, so — +// unlike the final commit — ANY commit error, lost ack included, is safe to treat as a +// failure: the pass aborts, and restart re-reads from the un-advanced cursor while the +// anti-join collapses whatever landed. A window with no decoded rows (e.g. all payloads +// were poison-dropped) is a no-op. +func (m *DuckLakeMaterializer) writeWindow(ctx context.Context, dec *decodedBatch) error { + if len(dec.signals) == 0 && len(dec.events) == 0 { + return nil + } + var cleanup []string + defer func() { + for _, f := range cleanup { + _ = os.Remove(f) + } + }() + conn, err := m.db.Conn(ctx) + if err != nil { + return fmt.Errorf("acquiring conn: %w", err) + } + defer conn.Close() //nolint:errcheck + tx, err := conn.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("begin window: %w", err) + } + defer func() { _ = tx.Rollback() }() + + files, err := m.insertDecodedSteady(ctx, tx, dec) + cleanup = append(cleanup, files...) + if err != nil { + return err + } + if err := tx.Commit(); err != nil { + if isCommitConflict(err) { + return errSnapshotMoved + } + return fmt.Errorf("commit window: %w", err) + } + m.markDirtyFromBatch(dec) return nil } @@ -1446,32 +1882,25 @@ func (m *DuckLakeMaterializer) PruneDecoded(ctx context.Context, retention time. const signalsLatestColumns = ` (subject, subject_bucket, name, "timestamp", value_number, ` + `value_string, loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen) ` -// TODO(load-review #5b): rollupSelectSQL still recomputes each dirty subject's rollup -// over its ENTIRE retained history every flush (the WHERE carries only subject_bucket + -// subject IN, no timestamp floor), so the cost grows with LAKE_DECODED_RETENTION × active -// fleet, independent of how few rows the batch added. A naive `timestamp >= floor` is NOT -// safe: `count`, `first_seen` (and `last_seen`) are full-history aggregates — a floor -// undercounts `count` and moves `first_seen` forward. Folding only the batch's new maxima -// is also not exactly-correct for `count`: the write anti-join keys on cloud_event_id, so a -// different-cloud_event_id duplicate of an existing (subject,name,timestamp) IS stored and -// only the read-path QUALIFY dedup collapses it — an incremental `count += len(batch)` would -// double-count exactly those, breaking the "each rollup row is exactly the deduped-base -// aggregate" invariant the rollup tests assert (tests/ducklake_latest_rollup_test.go, -// ducklake_dedup_test.go). Deferred rather than shipped unproven. +// rollupSelectSQL is the FULL-history recompute of a set of rollup rows. Steady-state +// maintenance no longer uses it — lake.signals_latest is folded INCREMENTALLY at commit +// time (foldSignalsRollup, #5b), O(batch) not O(history). This recompute is retained for +// the paths that genuinely need a from-scratch rebuild: the disaster-recovery / boot +// rebuild (RecomputeRollup / LAKE_REBUILD_ROLLUP_ON_BOOT) and the deferred bulk-backfill +// catch-up (FlushRollup over backfill-dirtied subjects). Kept byte-identical to the fold's +// result — the differential test (tests/ducklake_incremental_rollup_test.go) asserts the +// incremental path equals this recompute across redelivery, same-timestamp collision, +// out-of-order arrival, multi-window spans, and crash-replay. // -// Safe design for a follow-up: split the rollup's RECENCY columns from its CUMULATIVE -// columns. (a) Maintain (timestamp, value_*, loc_*, loc_ts, last_seen) by folding the -// batch's per-(subject,name) max against the existing rollup row — recency-only, so -// arg_max over {existing latest row, batch rows} equals arg_max over full history because -// the existing row already holds the prior max (no base scan, exact). (b) Keep (count, -// first_seen) exact with a bounded scan, not a full one: store a per-(subject,name) -// "counted-through" high-watermark = the max timestamp already folded into count; each -// flush counts only DEDUPED base rows in (watermark, now] (a recent-partition scan bounded -// like the write anti-join's dedupProbeFloor) and advances the watermark; first_seen -// min-folds the batch min. Retention's existing orphan-prune already handles deletes. This -// keeps count/first_seen exact while making the steady-state flush O(batch), not O(history). -// It needs its own exactness tests (incremental == full-recompute under redelivery) before -// it can replace the recompute below. +// Why the fold is exact where a naive `timestamp >= floor` recompute would not be: +// RECENCY (timestamp/value_*/loc_*/loc_ts) IS recomputed each batch, but bounded to +// `timestamp >= the row's prior latest` — the new latest is either in the batch or the +// prior row still qualifies, so the bound prunes old day-partitions without dropping the +// answer (self-healing + exact). COUNT and FIRST_SEEN are the full-history aggregates a +// floor would corrupt, so they are NOT floored: count carries forward as prev.count + a +// NOT-EXISTS delta over DISTINCT (subject,name,timestamp) — collisions and redeliveries +// contribute 0, and a replayed window contributes 0 (idempotent) — and first_seen min-folds +// the batch min. Both self-heal via this recompute on boot if a rollup row is ever lost. func rollupSelectSQL(whereClause string) string { const locNonzero = "(loc_lat != 0 OR loc_lon != 0)" return fmt.Sprintf(`SELECT subject, any_value(subject_bucket) AS subject_bucket, name, diff --git a/internal/materializer/materializer.go b/internal/materializer/materializer.go index 16f607b..ade79e3 100644 --- a/internal/materializer/materializer.go +++ b/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) } diff --git a/internal/materializer/metrics.go b/internal/materializer/metrics.go index 42701b1..a3732e5 100644 --- a/internal/materializer/metrics.go +++ b/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 diff --git a/internal/service/duck/events.go b/internal/service/duck/events.go index 589ac4b..07ea5b8 100644 --- a/internal/service/duck/events.go +++ b/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) } diff --git a/internal/service/duck/lake_latest.go b/internal/service/duck/lake_latest.go index 8ef73b1..8e251c9 100644 --- a/internal/service/duck/lake_latest.go +++ b/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 diff --git a/internal/service/duck/segments_source_test.go b/internal/service/duck/segments_source_test.go index 6f456ff..37f0c09 100644 --- a/internal/service/duck/segments_source_test.go +++ b/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) diff --git a/tests/VERIFY-CAMPAIGN.md b/tests/VERIFY-CAMPAIGN.md new file mode 100644 index 0000000..48a7eab --- /dev/null +++ b/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. diff --git a/tests/ducklake_event_rollup_test.go b/tests/ducklake_event_rollup_test.go index d5b2eaf..aa6e540 100644 --- a/tests/ducklake_event_rollup_test.go +++ b/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 +} diff --git a/tests/ducklake_incremental_rollup_test.go b/tests/ducklake_incremental_rollup_test.go new file mode 100644 index 0000000..3966f52 --- /dev/null +++ b/tests/ducklake_incremental_rollup_test.go @@ -0,0 +1,309 @@ +// ducklake_incremental_rollup_test.go proves finding #5b: lake.signals_latest is +// maintained INCREMENTALLY at decode-commit time (O(batch), not an O(history) recompute +// per flush), and the incremental result is EXACTLY the full RecomputeRollup — the +// differential invariant, checked across redelivery, same-timestamp collision, +// out-of-order arrival, multi-window (#1c) spans, location updates, and dormant re-report. +package tests + +import ( + "context" + "database/sql" + "fmt" + "math/rand" + "sort" + "testing" + "time" + + "github.com/DIMO-Network/dq/internal/materializer" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type rollupRow struct { + subject, name string + bucket int + timestamp time.Time + valueNumber sql.NullFloat64 + valueString sql.NullString + locLat, locLon float64 + locHdop, locHeading float64 + locTS time.Time + count int64 + firstSeen, lastSeen time.Time +} + +func dumpRollupMap(t *testing.T, ctx context.Context, db *sql.DB) map[string]rollupRow { + t.Helper() + rows, err := db.QueryContext(ctx, + `SELECT subject, name, subject_bucket, "timestamp", value_number, value_string, + loc_lat, loc_lon, loc_hdop, loc_heading, loc_ts, count, first_seen, last_seen + FROM lake.signals_latest ORDER BY subject, name`) + require.NoError(t, err) + defer rows.Close() //nolint:errcheck + out := map[string]rollupRow{} + for rows.Next() { + var r rollupRow + require.NoError(t, rows.Scan(&r.subject, &r.name, &r.bucket, &r.timestamp, &r.valueNumber, &r.valueString, + &r.locLat, &r.locLon, &r.locHdop, &r.locHeading, &r.locTS, &r.count, &r.firstSeen, &r.lastSeen)) + out[r.subject+"|"+r.name] = r + } + require.NoError(t, rows.Err()) + return out +} + +// assertMatchesRecompute snapshots the incrementally-maintained rollup, rebuilds it from +// scratch with RecomputeRollup, and asserts every column of every (subject,name) row is +// identical. This is the exactness contract for #5b. +func assertMatchesRecompute(t *testing.T, ctx context.Context, db *sql.DB, mat *materializer.DuckLakeMaterializer) { + t.Helper() + incremental := dumpRollupMap(t, ctx, db) + require.NoError(t, mat.RecomputeRollup(ctx)) + recomputed := dumpRollupMap(t, ctx, db) + + keys := map[string]bool{} + for k := range incremental { + keys[k] = true + } + for k := range recomputed { + keys[k] = true + } + sorted := make([]string, 0, len(keys)) + for k := range keys { + sorted = append(sorted, k) + } + sort.Strings(sorted) + for _, k := range sorted { + inc, okI := incremental[k] + rec, okR := recomputed[k] + require.Truef(t, okI, "%s present after RecomputeRollup but MISSING from the incremental rollup", k) + require.Truef(t, okR, "%s present in the incremental rollup but MISSING after RecomputeRollup", k) + assert.Equalf(t, rec.count, inc.count, "%s count", k) + assert.Truef(t, rec.timestamp.Equal(inc.timestamp), "%s timestamp: recompute=%s incremental=%s", k, rec.timestamp, inc.timestamp) + assert.Equalf(t, rec.valueNumber, inc.valueNumber, "%s value_number", k) + assert.Equalf(t, rec.valueString, inc.valueString, "%s value_string", k) + assert.Truef(t, rec.firstSeen.Equal(inc.firstSeen), "%s first_seen: recompute=%s incremental=%s", k, rec.firstSeen, inc.firstSeen) + assert.Truef(t, rec.lastSeen.Equal(inc.lastSeen), "%s last_seen: recompute=%s incremental=%s", k, rec.lastSeen, inc.lastSeen) + assert.InDeltaf(t, rec.locLat, inc.locLat, 1e-9, "%s loc_lat", k) + assert.InDeltaf(t, rec.locLon, inc.locLon, 1e-9, "%s loc_lon", k) + assert.Truef(t, rec.locTS.Equal(inc.locTS), "%s loc_ts: recompute=%s incremental=%s", k, rec.locTS, inc.locTS) + } +} + +func incrRunner(t *testing.T, ctx context.Context, db *sql.DB, opts ...func(*materializer.DuckLakeMaterializer)) (*materializer.Runner, *materializer.DuckLakeMaterializer) { + t.Helper() + mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + for _, o := range opts { + o(mat) + } + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) + return runner, mat +} + +// drainNoFlush runs the decode loop to completion WITHOUT calling FlushRollup, so +// lake.signals_latest is populated ONLY by the commit-time incremental fold (#5b) — the +// whole point of the differential test. +func drainNoFlush(t *testing.T, ctx context.Context, r *materializer.Runner) { + t.Helper() + for { + n, err := r.RunOnce(ctx) + require.NoError(t, err) + if n == 0 { + return + } + } +} + +// TestDuckLake_IncrementalRollup_MatchesRecompute drives the decode loop WITHOUT ever +// calling FlushRollup, so lake.signals_latest is populated only by the commit-time +// incremental fold, then asserts it equals a full RecomputeRollup after each scenario. +func TestDuckLake_IncrementalRollup_MatchesRecompute(t *testing.T) { + ctx := context.Background() + subject := fmt.Sprintf("did:erc721:137:%s:71", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -3).Truncate(time.Hour) + + t.Run("basic multi-batch increasing", func(t *testing.T) { + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "b1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) + seedRawStatus(t, db, "b2", subject, base.Add(2*time.Minute), speedAt(base.Add(2*time.Minute), 20)) + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "b3", subject, base.Add(3*time.Minute), speedAt(base.Add(3*time.Minute), 30)) + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[subject+"|speed"].count) + }) + + t.Run("redelivery same cloud_event_id no double count", func(t *testing.T) { + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "r1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "r1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) // redelivery + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subject+"|speed"].count) + }) + + t.Run("same-timestamp collision distinct count", func(t *testing.T) { + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + runner, mat := incrRunner(t, ctx, db) + ts := base.Add(5 * time.Minute) + seedRawStatus(t, db, "c1", subject, ts, speedAt(ts, 40)) + seedRawStatus(t, db, "c2", subject, ts, speedAt(ts, 41)) // same (s,n,ts), different ceid + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "a same-(subject,name,timestamp) collision is one distinct row") + }) + + t.Run("out-of-order older batch", func(t *testing.T) { + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "o2", subject, base.Add(10*time.Minute), speedAt(base.Add(10*time.Minute), 80)) + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "o1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 5)) // older, arrives later + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + got := dumpRollupMap(t, ctx, db)[subject+"|speed"] + assert.EqualValues(t, 2, got.count) + assert.EqualValues(t, 80, got.valueNumber.Float64, "latest stays the newer reading despite the later-arriving older one") + assert.True(t, got.firstSeen.Equal(base.Add(1*time.Minute)), "first_seen moves back to the older reading") + }) + + t.Run("multi-window fat span exact", func(t *testing.T) { + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(3) }) + seedRawStatusOneSnapshot(t, db, subject, base.Add(20*time.Minute), 10) // one snapshot, 10 rows, 3/window + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 10, dumpRollupMap(t, ctx, db)[subject+"|speed"].count) + }) + + t.Run("dormant re-report", func(t *testing.T) { + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "d1", subject, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) + drainNoFlush(t, ctx, runner) + // long gap, then report again + seedRawStatus(t, db, "d2", subject, base.Add(48*time.Hour), speedAt(base.Add(48*time.Hour), 55)) + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + got := dumpRollupMap(t, ctx, db)[subject+"|speed"] + assert.EqualValues(t, 2, got.count) + assert.EqualValues(t, 55, got.valueNumber.Float64) + }) +} + +// TestDuckLake_IncrementalRollup_CrashReplayExact proves the incremental fold is +// idempotent across a crash: a window commits its base rows AND its rollup delta, the pass +// then crashes before finishing the span, and on restart the replayed window must add 0 to +// count (rows already at rest → NOT-EXISTS delta 0, recency re-fold stable), so the final +// rollup still equals a full RecomputeRollup. +func TestDuckLake_IncrementalRollup_CrashReplayExact(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subject := fmt.Sprintf("did:erc721:137:%s:72", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + seedRawStatusOneSnapshot(t, db, subject, base, 9) // one snapshot, 9 rows + + // mat1 crashes right after the first intermediate window commits (base + rollup delta). + mat1, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + mat1.WithMaxRowsPerWindow(3) + mat1.WithWindowCommitHook(func(idx int) error { + if idx == 0 { + return fmt.Errorf("injected crash after first window") + } + return nil + }) + runner1 := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat1) + _, err = runner1.RunOnce(ctx) + require.Error(t, err) + assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "first window's rollup delta is durable") + + // Restart and drain: the replayed window must not double-count. + runner2, mat2 := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(3) }) + _ = runner2 + drainNoFlush(t, ctx, runner2) + assert.EqualValues(t, 9, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "replayed window added 0 — exactly nine") + assertMatchesRecompute(t, ctx, db, mat2) +} + +// TestDuckLake_IncrementalRollup_Randomized fuzzes many batches (redeliveries, collisions, +// out-of-order, varied window sizes, multiple subjects/names, location fixes) through the +// incremental path and asserts the result equals RecomputeRollup every time. Deterministic +// seed for reproducibility. +func TestDuckLake_IncrementalRollup_Randomized(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + rng := rand.New(rand.NewSource(20260707)) + subjects := []string{ + fmt.Sprintf("did:erc721:137:%s:81", vehicleNFT.Hex()), + fmt.Sprintf("did:erc721:137:%s:82", vehicleNFT.Hex()), + fmt.Sprintf("did:erc721:137:%s:83", vehicleNFT.Hex()), + } + names := []func(ts time.Time, v float64) map[string]any{speedAt, odoAt} + base := time.Now().UTC().AddDate(0, 0, -5).Truncate(time.Hour) + + runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(2 + rng.Intn(5)) }) + seen := map[string]bool{} // ids emitted, to force redeliveries + seq := 0 + for round := 0; round < 25; round++ { + n := 1 + rng.Intn(4) + for i := 0; i < n; i++ { + subj := subjects[rng.Intn(len(subjects))] + nameFn := names[rng.Intn(len(names))] + // timestamps wander forward and sometimes backward (out-of-order) + off := time.Duration(seq*7+rng.Intn(400)-100) * time.Second + ts := base.Add(off) + seq++ + id := fmt.Sprintf("rnd-%d", seq) + if len(seen) > 0 && rng.Intn(5) == 0 { // redelivery of a prior id + for k := range seen { + id = k + break + } + } else if rng.Intn(6) == 0 && seq > 1 { // collision: new id, reuse a recent timestamp + ts = base.Add(time.Duration((seq-1)*7) * time.Second) + } + seen[id] = true + seedRawStatus(t, db, id, subj, ts, nameFn(ts, float64(rng.Intn(120)))) + } + drainNoFlush(t, ctx, runner) + if round%5 == 4 { + assertMatchesRecompute(t, ctx, db, mat) + } + } + assertMatchesRecompute(t, ctx, db, mat) +} + +func odoAt(ts time.Time, v float64) map[string]any { + return map[string]any{"name": "powertrainTransmissionTravelledDistance", "timestamp": ts.Format(time.RFC3339Nano), "value": v} +} + +// TestDuckLake_IncrementalRollup_AncientRedelivery pins the >30d-redelivery count fix: a +// reading older than the 30d dedup probe floor, redelivered (same cloud_event_id) through +// the live path, must NOT inflate signals_latest.count — it stays equal to RecomputeRollup. +func TestDuckLake_IncrementalRollup_AncientRedelivery(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:73", vehicleNFT.Hex()) + ancient := time.Now().UTC().Add(-40 * 24 * time.Hour).Truncate(time.Hour) // > dedupProbeFloor (30d) + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "anc1", subj, ancient, speedAt(ancient, 22)) + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "anc1", subj, ancient, speedAt(ancient, 22)) // redelivery of the SAME event, >30d old + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 1, dumpRollupMap(t, ctx, db)[subj+"|speed"].count, "an ancient redelivery must not inflate count") +} diff --git a/tests/ducklake_pagination_test.go b/tests/ducklake_pagination_test.go new file mode 100644 index 0000000..d46ad64 --- /dev/null +++ b/tests/ducklake_pagination_test.go @@ -0,0 +1,148 @@ +// ducklake_pagination_test.go proves finding #1c: a single oversized snapshot is +// decoded and written in byte/row-bounded WINDOWS (so its resident working set can't +// OOM the single writer), while preserving exactly-once — including a crash between an +// intermediate window and the final cursor-advancing commit, which the old monolithic +// "read whole delta, one commit" path could never produce. +package tests + +import ( + "context" + "database/sql" + "errors" + "fmt" + "testing" + "time" + + "github.com/DIMO-Network/dq/internal/materializer" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// seedRawStatusOneSnapshot inserts n status rows for subject in a SINGLE transaction, +// so DuckLake commits them as ONE snapshot. The span bound (maxSnapshotSpan) cannot +// split a single snapshot, so a fat snapshot must be paged by row/byte budget or it +// materializes whole. +func seedRawStatusOneSnapshot(t *testing.T, db *sql.DB, subject string, base time.Time, n int) { + t.Helper() + tx, err := db.Begin() + require.NoError(t, err) + for i := 0; i < n; i++ { + at := base.Add(time.Duration(i) * time.Second) + ev := deviceStatus(fmt.Sprintf("snap-%s-%d", subject, i), subject, at, speedAt(at, float64(i))) + _, err := tx.Exec( + `INSERT INTO lake.raw_events (subject, "time", type, id, source, producer, data_content_type, data_version, extras, data) + VALUES (?, ?, ?, ?, ?, ?, '', ?, '{}', ?)`, + ev.Subject, ev.Time.UTC(), ev.Type, ev.ID, ev.Source, ev.Producer, ev.DataVersion, string(ev.Data)) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) +} + +func countSpeedRows(t *testing.T, ctx context.Context, db *sql.DB, subject string) int { + t.Helper() + var n int + require.NoError(t, db.QueryRowContext(ctx, + "SELECT count(*) FROM lake.signals WHERE subject = ? AND name = 'speed'", subject).Scan(&n)) + return n +} + +func TestDuckLake_PaginatesFatSnapshotExactlyOnce(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subject := fmt.Sprintf("did:erc721:137:%s:91", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -1).Truncate(time.Hour) + + seedRawStatusOneSnapshot(t, db, subject, base, 6) // ONE snapshot, six rows + + mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + mat.WithMaxRowsPerWindow(2) // force pagination: 6 rows over 2-row windows + + var intermediateWindows int + mat.WithWindowCommitHook(func(idx int) error { intermediateWindows++; return nil }) + + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). + WithDuckLake(mat) + + n, err := runner.RunOnce(ctx) + require.NoError(t, err) + assert.Equal(t, 6, n, "all six rows of the fat snapshot decoded") + assert.Equal(t, 6, countSpeedRows(t, ctx, db, subject), "exactly six distinct rows landed (no dup, no loss)") + assert.Positive(t, intermediateWindows, "the fat snapshot was written in multiple bounded windows, not one") + assert.Positive(t, readCursor(t, ctx, db), "cursor advanced past the snapshot") + + // Idempotent: a re-run with nothing new must not double-insert. + _, err = runner.RunOnce(ctx) + require.NoError(t, err) + assert.Equal(t, 6, countSpeedRows(t, ctx, db, subject), "re-run does not double-insert") +} + +func TestDuckLake_PaginationCrashMidSpanRecoversExactlyOnce(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subject := fmt.Sprintf("did:erc721:137:%s:92", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -1).Truncate(time.Hour) + seedRawStatusOneSnapshot(t, db, subject, base, 6) + + // mat1 crashes right after the first intermediate window commits, before the final + // cursor-advancing commit — the mid-snapshot crash the monolithic path (one commit) + // could never produce. + mat1, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + mat1.WithMaxRowsPerWindow(2) + crashErr := errors.New("injected crash after first window") + mat1.WithWindowCommitHook(func(idx int) error { + if idx == 0 { + return crashErr + } + return nil + }) + runner1 := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat1) + _, err = runner1.RunOnce(ctx) + require.Error(t, err, "the pass fails on the injected mid-span crash") + + // Partial durability: the first window's rows are at rest, but the cursor did NOT + // advance (the span isn't finished) — so restart re-reads the whole snapshot. + assert.Equal(t, 2, countSpeedRows(t, ctx, db, subject), "first window's rows are durable") + assert.EqualValues(t, 0, readCursor(t, ctx, db), "cursor not advanced on a partial span") + + // Restart: a fresh materializer re-reads the whole snapshot from the un-advanced + // cursor; the anti-join collapses the two already-written rows and the pass finishes + // exactly-once. + mat2, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + mat2.WithMaxRowsPerWindow(2) + runner2 := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat2) + drainRunner(t, ctx, runner2) + assert.Equal(t, 6, countSpeedRows(t, ctx, db, subject), "restart converges to exactly six rows (no dup from the replayed window)") + assert.Positive(t, readCursor(t, ctx, db), "cursor advanced after the span completed") +} + +// TestDuckLake_SmallSnapshotSingleWindow proves the common case is unchanged: a +// snapshot that fits in one window is written in a single commit (no intermediate +// windows), behaviorally identical to the pre-#1c monolithic path. +func TestDuckLake_SmallSnapshotSingleWindow(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subject := fmt.Sprintf("did:erc721:137:%s:93", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -1).Truncate(time.Hour) + seedRawStatusOneSnapshot(t, db, subject, base, 3) + + mat, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + mat.WithMaxRowsPerWindow(100) // window larger than the snapshot + + var intermediateWindows int + mat.WithWindowCommitHook(func(idx int) error { intermediateWindows++; return nil }) + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) + + n, err := runner.RunOnce(ctx) + require.NoError(t, err) + assert.Equal(t, 3, n) + assert.Equal(t, 3, countSpeedRows(t, ctx, db, subject)) + assert.Zero(t, intermediateWindows, "a snapshot within one window commits once — no intermediate windows") +} diff --git a/tests/ducklake_pg_test.go b/tests/ducklake_pg_test.go index 68288a8..afa586b 100644 --- a/tests/ducklake_pg_test.go +++ b/tests/ducklake_pg_test.go @@ -116,3 +116,95 @@ func TestDuckLakePostgres_ConcurrentMaterializers(t *testing.T) { GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) assert.Zero(t, dupes, "no duplicate decoded rows") } + +// TestDuckLakePostgres_ConcurrentPaginatedFatSnapshot proves finding #1c under a real +// catalog: a SINGLE oversized snapshot (many rows, span can't be count-split) is drained +// by two independent materializers using tiny per-window bounds, so most rows flow +// through intermediate (non-cursor-advancing) windows while both writers race the final +// cursor CAS. Exactly-once must still hold — the intermediate-window idempotency + the +// cursor-coupled final commit are the #1c invariant, exercised across connections that a +// single-process file catalog cannot reproduce. +func TestDuckLakePostgres_ConcurrentPaginatedFatSnapshot(t *testing.T) { + dsn := pgCatalogDSN(t) + ctx := context.Background() + dataPath := os.Getenv("PG_DATA_PATH") + if dataPath == "" { + dataPath = t.TempDir() + } + subject := fmt.Sprintf("did:erc721:137:%s:56", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + + seed := newPGLakeService(t, dsn, dataPath) + db := seed.DB() + for _, q := range []string{ + "DROP TABLE IF EXISTS lake.signals", "DROP TABLE IF EXISTS lake.events", + "DROP TABLE IF EXISTS lake.signals_latest", "DROP TABLE IF EXISTS lake.events_latest", + "DROP TABLE IF EXISTS lake.ingest_progress", "DROP TABLE IF EXISTS lake.raw_events", + `CREATE TABLE lake.raw_events (subject VARCHAR, "time" TIMESTAMP WITH TIME ZONE, type VARCHAR, + id VARCHAR, source VARCHAR, producer VARCHAR, data_content_type VARCHAR, data_version VARCHAR, + extras VARCHAR, data VARCHAR, data_base64 BLOB, data_index_key VARCHAR, voids_id VARCHAR)`, + } { + _, err := db.ExecContext(ctx, q) + require.NoError(t, err, q) + } + + // ONE snapshot, 50 rows: a single transaction commits them as one snapshot, so the + // span bound cannot split it — only intra-snapshot pagination can bound the pass. + const events = 50 + tx, err := db.Begin() + require.NoError(t, err) + for i := range events { + at := base.Add(time.Duration(i) * time.Second) + ev := deviceStatus(fmt.Sprintf("pgpag-%d", i), subject, at, speedAt(at, float64(i))) + _, err := tx.ExecContext(ctx, + `INSERT INTO lake.raw_events (subject, "time", type, id, source, producer, data_content_type, data_version, extras, data) + VALUES (?, ?, ?, ?, ?, ?, '', ?, '{}', ?)`, + ev.Subject, ev.Time.UTC(), ev.Type, ev.ID, ev.Source, ev.Producer, ev.DataVersion, string(ev.Data)) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) + + run := func() { + svc := newPGLakeService(t, dsn, dataPath) + mat, err := materializer.NewDuckLakeMaterializer(ctx, svc.DB(), zerolog.Nop()) + require.NoError(t, err) + mat.WithMaxRowsPerWindow(7) // force multi-window pagination of the fat snapshot + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()). + WithDuckLake(mat) + for { + n, err := runner.RunOnce(ctx) + require.NoError(t, err) + if n == 0 { + return + } + } + } + var wg sync.WaitGroup + for range 2 { + wg.Add(1) + go func() { defer wg.Done(); run() }() + } + wg.Wait() + + var rows int + require.NoError(t, db.QueryRowContext(ctx, + "SELECT count(*) FROM lake.signals WHERE subject = ? AND name = 'speed'", subject).Scan(&rows)) + assert.Equal(t, events, rows, "paginated fat snapshot decoded exactly once under concurrent materializers") + + var dupes int + require.NoError(t, db.QueryRowContext(ctx, `SELECT count(*) FROM ( + SELECT cloud_event_id, name, timestamp FROM lake.signals + GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) + assert.Zero(t, dupes, "no duplicate decoded rows across paginated windows") + + // The commit-time incremental rollup (#5b) must be exact even though it ran across + // intermediate windows under two racing writers: it equals a full RecomputeRollup. + incremental := dumpRollupMap(t, ctx, db) + assert.EqualValues(t, events, incremental[subject+"|speed"].count, "incremental rollup count is exact under concurrent paginated writers") + mat, err := materializer.NewDuckLakeMaterializer(ctx, newPGLakeService(t, dsn, dataPath).DB(), zerolog.Nop()) + require.NoError(t, err) + require.NoError(t, mat.RecomputeRollup(ctx)) + recomputed := dumpRollupMap(t, ctx, db) + assert.Equal(t, recomputed[subject+"|speed"].count, incremental[subject+"|speed"].count, "incremental rollup == recompute (PG)") + assert.True(t, recomputed[subject+"|speed"].timestamp.Equal(incremental[subject+"|speed"].timestamp), "incremental recency == recompute (PG)") +} diff --git a/tests/ducklake_verify_pg_test.go b/tests/ducklake_verify_pg_test.go new file mode 100644 index 0000000..27e3d08 --- /dev/null +++ b/tests/ducklake_verify_pg_test.go @@ -0,0 +1,194 @@ +// ducklake_verify_pg_test.go — verification loops 9-10: the #1c pagination + #5b +// incremental rollup under real-Postgres concurrency with MORE than two writers and a +// mid-span crash of one writer while the others drain. Gated on PG_CATALOG_DSN. +package tests + +import ( + "context" + "fmt" + "os" + "sync" + "testing" + "time" + + "github.com/DIMO-Network/dq/internal/materializer" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// V9 — three independent materializers drain the same paginated fat snapshot; exactly-once +// base rows AND an exact incremental rollup. +func TestVerify09_PG_ThreeConcurrentPaginated(t *testing.T) { + dsn := pgCatalogDSN(t) + ctx := context.Background() + dataPath := os.Getenv("PG_DATA_PATH") + if dataPath == "" { + dataPath = t.TempDir() + } + subject := fmt.Sprintf("did:erc721:137:%s:57", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + + seed := newPGLakeService(t, dsn, dataPath) + db := seed.DB() + for _, q := range []string{ + "DROP TABLE IF EXISTS lake.signals", "DROP TABLE IF EXISTS lake.events", + "DROP TABLE IF EXISTS lake.signals_latest", "DROP TABLE IF EXISTS lake.events_latest", + "DROP TABLE IF EXISTS lake.ingest_progress", "DROP TABLE IF EXISTS lake.raw_events", + `CREATE TABLE lake.raw_events (subject VARCHAR, "time" TIMESTAMP WITH TIME ZONE, type VARCHAR, + id VARCHAR, source VARCHAR, producer VARCHAR, data_content_type VARCHAR, data_version VARCHAR, + extras VARCHAR, data VARCHAR, data_base64 BLOB, data_index_key VARCHAR, voids_id VARCHAR)`, + } { + _, err := db.ExecContext(ctx, q) + require.NoError(t, err, q) + } + const events = 60 + tx, err := db.Begin() + require.NoError(t, err) + for i := range events { + at := base.Add(time.Duration(i) * time.Second) + ev := deviceStatus(fmt.Sprintf("v9-%d", i), subject, at, speedAt(at, float64(i))) + _, err := tx.ExecContext(ctx, + `INSERT INTO lake.raw_events (subject, "time", type, id, source, producer, data_content_type, data_version, extras, data) + VALUES (?, ?, ?, ?, ?, ?, '', ?, '{}', ?)`, + ev.Subject, ev.Time.UTC(), ev.Type, ev.ID, ev.Source, ev.Producer, ev.DataVersion, string(ev.Data)) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) + + run := func() { + svc := newPGLakeService(t, dsn, dataPath) + mat, err := materializer.NewDuckLakeMaterializer(ctx, svc.DB(), zerolog.Nop()) + require.NoError(t, err) + mat.WithMaxRowsPerWindow(5) + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) + for { + n, err := runner.RunOnce(ctx) + require.NoError(t, err) + if n == 0 { + return + } + } + } + var wg sync.WaitGroup + for range 3 { + wg.Add(1) + go func() { defer wg.Done(); run() }() + } + wg.Wait() + + var rows, dupes int + require.NoError(t, db.QueryRowContext(ctx, + "SELECT count(*) FROM lake.signals WHERE subject = ? AND name = 'speed'", subject).Scan(&rows)) + assert.Equal(t, events, rows, "three racing writers decode the paginated snapshot exactly once") + require.NoError(t, db.QueryRowContext(ctx, `SELECT count(*) FROM ( + SELECT cloud_event_id, name, timestamp FROM lake.signals GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) + assert.Zero(t, dupes) + assert.EqualValues(t, events, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "incremental rollup exact under three writers") +} + +// V10 — one writer crashes mid-span (its intermediate window errors) while two others drain; +// the idempotent windows + cursor-coupled final commit still yield exactly-once + an exact +// rollup. +func TestVerify10_PG_CrashOneWriterMidSpan(t *testing.T) { + dsn := pgCatalogDSN(t) + ctx := context.Background() + dataPath := os.Getenv("PG_DATA_PATH") + if dataPath == "" { + dataPath = t.TempDir() + } + subject := fmt.Sprintf("did:erc721:137:%s:58", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + + seed := newPGLakeService(t, dsn, dataPath) + db := seed.DB() + for _, q := range []string{ + "DROP TABLE IF EXISTS lake.signals", "DROP TABLE IF EXISTS lake.events", + "DROP TABLE IF EXISTS lake.signals_latest", "DROP TABLE IF EXISTS lake.events_latest", + "DROP TABLE IF EXISTS lake.ingest_progress", "DROP TABLE IF EXISTS lake.raw_events", + `CREATE TABLE lake.raw_events (subject VARCHAR, "time" TIMESTAMP WITH TIME ZONE, type VARCHAR, + id VARCHAR, source VARCHAR, producer VARCHAR, data_content_type VARCHAR, data_version VARCHAR, + extras VARCHAR, data VARCHAR, data_base64 BLOB, data_index_key VARCHAR, voids_id VARCHAR)`, + } { + _, err := db.ExecContext(ctx, q) + require.NoError(t, err, q) + } + const events = 60 + tx, err := db.Begin() + require.NoError(t, err) + for i := range events { + at := base.Add(time.Duration(i) * time.Second) + ev := deviceStatus(fmt.Sprintf("v10-%d", i), subject, at, speedAt(at, float64(i))) + _, err := tx.ExecContext(ctx, + `INSERT INTO lake.raw_events (subject, "time", type, id, source, producer, data_content_type, data_version, extras, data) + VALUES (?, ?, ?, ?, ?, ?, '', ?, '{}', ?)`, + ev.Subject, ev.Time.UTC(), ev.Type, ev.ID, ev.Source, ev.Producer, ev.DataVersion, string(ev.Data)) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) + + // The flaky writer crashes on its first intermediate window, then is restarted (a + // real supervisor loop); the two healthy writers race it the whole time. + flaky := func() { + crashed := false + for attempt := 0; attempt < 6; attempt++ { + svc := newPGLakeService(t, dsn, dataPath) + mat, err := materializer.NewDuckLakeMaterializer(ctx, svc.DB(), zerolog.Nop()) + require.NoError(t, err) + mat.WithMaxRowsPerWindow(5) + if !crashed { + mat.WithWindowCommitHook(func(idx int) error { + if idx == 0 { + crashed = true + return fmt.Errorf("injected mid-span crash") + } + return nil + }) + } + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) + done := false + for { + n, err := runner.RunOnce(ctx) + if err != nil { + break // crashed; loop restarts a fresh materializer + } + if n == 0 { + done = true + break + } + } + if done { + return + } + } + } + healthy := func() { + svc := newPGLakeService(t, dsn, dataPath) + mat, err := materializer.NewDuckLakeMaterializer(ctx, svc.DB(), zerolog.Nop()) + require.NoError(t, err) + mat.WithMaxRowsPerWindow(5) + runner := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat) + for { + n, err := runner.RunOnce(ctx) + require.NoError(t, err) + if n == 0 { + return + } + } + } + var wg sync.WaitGroup + wg.Add(3) + go func() { defer wg.Done(); flaky() }() + go func() { defer wg.Done(); healthy() }() + go func() { defer wg.Done(); healthy() }() + wg.Wait() + + var rows, dupes int + require.NoError(t, db.QueryRowContext(ctx, + "SELECT count(*) FROM lake.signals WHERE subject = ? AND name = 'speed'", subject).Scan(&rows)) + assert.Equal(t, events, rows, "exactly-once despite a writer crashing mid-span") + require.NoError(t, db.QueryRowContext(ctx, `SELECT count(*) FROM ( + SELECT cloud_event_id, name, timestamp FROM lake.signals GROUP BY cloud_event_id, name, timestamp HAVING count(*) > 1)`).Scan(&dupes)) + assert.Zero(t, dupes) + assert.EqualValues(t, events, dumpRollupMap(t, ctx, db)[subject+"|speed"].count, "incremental rollup exact after a mid-span crash under concurrency") +} diff --git a/tests/ducklake_verify_test.go b/tests/ducklake_verify_test.go new file mode 100644 index 0000000..61f3a02 --- /dev/null +++ b/tests/ducklake_verify_test.go @@ -0,0 +1,209 @@ +// ducklake_verify_test.go — adversarial verification campaign for the #1c pagination + +// #5b incremental-rollup exactly-once path. Each subtest is a distinct attack vector; the +// contract is always the same: the incrementally-maintained lake.signals_latest equals a +// full RecomputeRollup, and base rows are exactly-once. These are throwaway-hardening +// tests that also stay as regression guards. +package tests + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/DIMO-Network/dq/internal/materializer" + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func locAt(ts time.Time, lat, lon, hdop float64) map[string]any { + return map[string]any{"name": "currentLocationCoordinates", "timestamp": ts.Format(time.RFC3339Nano), + "value": map[string]any{"latitude": lat, "longitude": lon, "hdop": hdop}} +} + +// V1 — a location signal has value_number NULL and loc_* set; the fold must keep +// value_number NULL and carry loc columns exactly like the recompute. +func TestVerify01_LocationSignalNullValueNumber(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:1", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "v1a", subj, base.Add(1*time.Minute), locAt(base.Add(1*time.Minute), 42.1, -83.0, 1.2)) + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "v1b", subj, base.Add(2*time.Minute), locAt(base.Add(2*time.Minute), 42.3, -83.1, 0.9)) + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) // the real contract: value_number (whatever it is) matches the recompute + got := dumpRollupMap(t, ctx, db)[subj+"|currentLocationCoordinates"] + assert.InDelta(t, 42.3, got.locLat, 1e-9, "loc_lat folds to the newest fix") + assert.True(t, got.locTS.Equal(base.Add(2*time.Minute)), "loc_ts is the newest fix") +} + +// V2 — location is intermittent: a fix, then non-loc batches, then an OLDER loc; loc_ts +// must stay the newest fix (not regress, not be overwritten by the older one). +func TestVerify02_IntermittentLocation(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:2", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "v2loc", subj, base.Add(5*time.Minute), locAt(base.Add(5*time.Minute), 40.0, -80.0, 1.0)) + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "v2sp", subj, base.Add(9*time.Minute), speedAt(base.Add(9*time.Minute), 33)) // newer, no loc + drainNoFlush(t, ctx, runner) + seedRawStatus(t, db, "v2old", subj, base.Add(1*time.Minute), locAt(base.Add(1*time.Minute), 1.0, 2.0, 9.0)) // older loc + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + loc := dumpRollupMap(t, ctx, db)[subj+"|currentLocationCoordinates"] + assert.True(t, loc.locTS.Equal(base.Add(5*time.Minute)), "loc_ts stays the newest fix despite a later-arriving older one") + assert.InDelta(t, 40.0, loc.locLat, 1e-9) +} + +// V3 — a fat single snapshot with many (subject,name) pairs folds exactly in one window. +func TestVerify03_ManyKeysOneSnapshot(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + runner, mat := incrRunner(t, ctx, db) + tx, err := db.Begin() + require.NoError(t, err) + for s := 0; s < 6; s++ { + subj := fmt.Sprintf("did:erc721:137:%s:%d", vehicleNFT.Hex(), 30+s) + for i := 0; i < 4; i++ { + ts := base.Add(time.Duration(s*10+i) * time.Minute) + for _, sig := range []map[string]any{speedAt(ts, float64(i*5)), odoAt(ts, float64(1000+i))} { + ev := deviceStatus(fmt.Sprintf("v3-%d-%d-%s", s, i, sig["name"]), subj, ts, sig) + _, err := tx.Exec(`INSERT INTO lake.raw_events (subject,"time",type,id,source,producer,data_content_type,data_version,extras,data) VALUES (?,?,?,?,?,?,'',?, '{}',?)`, + ev.Subject, ev.Time.UTC(), ev.Type, ev.ID, ev.Source, ev.Producer, ev.DataVersion, string(ev.Data)) + require.NoError(t, err) + } + } + } + require.NoError(t, tx.Commit()) + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.Len(t, dumpRollupMap(t, ctx, db), 12, "6 subjects x 2 names = 12 rollup rows") +} + +// V4 — pagination driven by the BYTE budget (row cap disabled) still exact. +func TestVerify04_ByteBudgetPagination(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:4", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + var intermediate int + runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { + m.WithMaxRowsPerWindow(0) // disable the row cap + m.WithWindowByteBudget(100) // tiny byte budget -> flush after each read chunk + m.WithWindowCommitHook(func(int) error { intermediate++; return nil }) + }) + seedRawStatusOneSnapshot(t, db, subj, base, 1100) // > windowReadChunk so the byte path paginates + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 1100, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) + assert.Positive(t, intermediate, "the byte-budget path split the span into multiple windows") +} + +// V5 — crash at the LAST intermediate window (most windows durable, cursor not advanced); +// restart converges exactly for BOTH base rows and the rollup. +func TestVerify05_CrashLastIntermediateWindow(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:5", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + seedRawStatusOneSnapshot(t, db, subj, base, 9) // windows of 2 -> intermediate idx 0..3, final = 1 row + mat1, err := materializer.NewDuckLakeMaterializer(ctx, db, zerolog.Nop()) + require.NoError(t, err) + mat1.WithMaxRowsPerWindow(2) + mat1.WithWindowCommitHook(func(idx int) error { + if idx == 3 { + return fmt.Errorf("crash at last intermediate window") + } + return nil + }) + r1 := materializer.New(materializer.Config{ChainID: 137, VehicleNFTAddress: vehicleNFT}, zerolog.Nop()).WithDuckLake(mat1) + _, err = r1.RunOnce(ctx) + require.Error(t, err) + assert.EqualValues(t, 0, readCursor(t, ctx, db), "cursor not advanced on a partial span") + runner2, mat2 := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(2) }) + drainNoFlush(t, ctx, runner2) + assert.EqualValues(t, 9, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) + assertMatchesRecompute(t, ctx, db, mat2) +} + +// V6 — a paginated span mixes decodable rows with rows that decode to NOTHING (wrong-chain +// subject). The row-key cursor must advance past the non-decodable rows (no hang), and only +// real signals are counted. +func TestVerify06_MixedNonDecodableRows(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + good := fmt.Sprintf("did:erc721:137:%s:6", vehicleNFT.Hex()) + bad := fmt.Sprintf("did:erc721:1:%s:6", vehicleNFT.Hex()) // wrong chain -> not a vehicle -> 0 signals + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + tx, err := db.Begin() + require.NoError(t, err) + for i := 0; i < 12; i++ { + subj := good + if i%2 == 1 { + subj = bad + } + ts := base.Add(time.Duration(i) * time.Minute) + ev := deviceStatus(fmt.Sprintf("v6-%d", i), subj, ts, speedAt(ts, float64(i))) + _, err := tx.Exec(`INSERT INTO lake.raw_events (subject,"time",type,id,source,producer,data_content_type,data_version,extras,data) VALUES (?,?,?,?,?,?,'',?, '{}',?)`, + ev.Subject, ev.Time.UTC(), ev.Type, ev.ID, ev.Source, ev.Producer, ev.DataVersion, string(ev.Data)) + require.NoError(t, err) + } + require.NoError(t, tx.Commit()) + runner, mat := incrRunner(t, ctx, db, func(m *materializer.DuckLakeMaterializer) { m.WithMaxRowsPerWindow(3) }) + drainNoFlush(t, ctx, runner) + assert.Positive(t, readCursor(t, ctx, db), "cursor advanced past non-decodable rows") + assert.EqualValues(t, 6, dumpRollupMap(t, ctx, db)[good+"|speed"].count, "only the 6 vehicle rows counted") + assertMatchesRecompute(t, ctx, db, mat) +} + +// V7 — an out-of-order batch 100 days older than the existing latest: recency unchanged, +// count increments, first_seen moves back, exactly matching the recompute. +func TestVerify07_DeepOutOfOrder(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:7", vehicleNFT.Hex()) + now := time.Now().UTC().Truncate(time.Hour) + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "v7new", subj, now.Add(-1*time.Hour), speedAt(now.Add(-1*time.Hour), 70)) + drainNoFlush(t, ctx, runner) + old := now.AddDate(0, 0, -100) + seedRawStatus(t, db, "v7old", subj, old, speedAt(old, 12)) + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + got := dumpRollupMap(t, ctx, db)[subj+"|speed"] + assert.EqualValues(t, 2, got.count) + assert.EqualValues(t, 70, got.valueNumber.Float64, "recency unchanged by the 100-day-old arrival") + assert.True(t, got.firstSeen.Equal(old), "first_seen moved back 100 days") +} + +// V8 — interleave a full RecomputeRollup with incremental folds: the incremental path must +// stay exact when it continues on top of a freshly recomputed rollup (self-healing). +func TestVerify08_RecomputeInterleave(t *testing.T) { + ctx := context.Background() + svc := newLakeService(t, t.TempDir()) + db := svc.DB() + subj := fmt.Sprintf("did:erc721:137:%s:8", vehicleNFT.Hex()) + base := time.Now().UTC().AddDate(0, 0, -2).Truncate(time.Hour) + runner, mat := incrRunner(t, ctx, db) + seedRawStatus(t, db, "v8a", subj, base.Add(1*time.Minute), speedAt(base.Add(1*time.Minute), 10)) + seedRawStatus(t, db, "v8b", subj, base.Add(2*time.Minute), speedAt(base.Add(2*time.Minute), 20)) + drainNoFlush(t, ctx, runner) + require.NoError(t, mat.RecomputeRollup(ctx)) // rebuild from base mid-stream + seedRawStatus(t, db, "v8c", subj, base.Add(3*time.Minute), speedAt(base.Add(3*time.Minute), 30)) + drainNoFlush(t, ctx, runner) + assertMatchesRecompute(t, ctx, db, mat) + assert.EqualValues(t, 3, dumpRollupMap(t, ctx, db)[subj+"|speed"].count) +} diff --git a/tests/local_e2e_186612_fetch_test.go b/tests/local_e2e_186612_fetch_test.go new file mode 100644 index 0000000..4800536 --- /dev/null +++ b/tests/local_e2e_186612_fetch_test.go @@ -0,0 +1,157 @@ +//go:build e2e186612fetch + +// local_e2e_186612_fetch_test.go exercises dq's fetch-api (cloudEvents) surface +// in-process against a din-backfilled DuckLake catalog for one vehicle, with the +// auth directives bypassed. It confirms the three fetch queries serve and that +// availableCloudEventTypes matches the raw_events ground truth (the prod +// cloudevent store the dump was taken from). Env mirrors the signals harness: +// +// E2E_CATALOG, E2E_DATAPATH, E2E_SUBJECT, E2E_OUT +package tests + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "sort" + "testing" + + "github.com/99designs/gqlgen/client" + gqlhandler "github.com/99designs/gqlgen/graphql/handler" + "github.com/99designs/gqlgen/graphql/handler/transport" + "github.com/DIMO-Network/dq/internal/graph" + "github.com/DIMO-Network/dq/internal/service/duck" + "github.com/DIMO-Network/token-exchange-api/pkg/tokenclaims" + "github.com/stretchr/testify/require" +) + +func fetchEnv(t *testing.T, k string) string { + t.Helper() + v := os.Getenv(k) + require.NotEmptyf(t, v, "env %s is required", k) + return v +} + +func TestLocalE2E_186612_Fetch(t *testing.T) { + ctx := context.Background() + catalog := fetchEnv(t, "E2E_CATALOG") + dataPath := fetchEnv(t, "E2E_DATAPATH") + subject := fetchEnv(t, "E2E_SUBJECT") + outPath := os.Getenv("E2E_OUT") + + svc, err := duck.NewService(duck.Config{ + DuckLakeEnabled: true, + CatalogDSN: catalog, + DataPath: dataPath, + }) + require.NoError(t, err) + defer svc.Close() //nolint:errcheck + db := svc.DB() + + // Ground truth straight from the backfilled catalog: per-type count + span. + type typeRow struct { + Type string `json:"type"` + Count int64 `json:"count"` + FirstSeen string `json:"firstSeen"` + LastSeen string `json:"lastSeen"` + } + groundRows, err := db.QueryContext(ctx, `SELECT type, count(*), + CAST(min("time") AS VARCHAR), CAST(max("time") AS VARCHAR) + FROM lake.raw_events WHERE subject = ? GROUP BY type ORDER BY type`, subject) + require.NoError(t, err) + ground := map[string]typeRow{} + for groundRows.Next() { + var r typeRow + require.NoError(t, groundRows.Scan(&r.Type, &r.Count, &r.FirstSeen, &r.LastSeen)) + ground[r.Type] = r + } + require.NoError(t, groundRows.Err()) + require.NotEmpty(t, ground, "no raw events for subject") + t.Logf("ground-truth raw_events types: %d", len(ground)) + + // Fetch event service over the same lake (no S3: 186612 has no externalized blobs). + es := duck.NewLakeEventService(svc, nil, nil, "") + cfg := graph.Config{Resolvers: &graph.Resolver{EventService: es}} + cfg.Directives.RequiresVehicleToken = passDirective + cfg.Directives.IsSignal = passDirective + cfg.Directives.HasAggregation = passDirective + cfg.Directives.McpHide = passDirective + cfg.Directives.RequiresAllOfPrivileges = passPrivilegeDirective + cfg.Directives.RequiresOneOfPrivilege = passPrivilegeDirective + srv := gqlhandler.New(graph.NewExecutableSchema(cfg)) + srv.AddTransport(transport.POST{}) + + // The cloudEvent resolvers re-check claims in-resolver (requireSubjectOptsByDID), + // independent of the bypassed directive — inject a raw-data token whose Asset is + // the requested subject so the DID-link check short-circuits without identity. + tok := &tokenclaims.Token{CustomClaims: tokenclaims.CustomClaims{ + Asset: subject, + Permissions: []string{tokenclaims.PermissionGetRawData}, + }} + withClaims := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + srv.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), graph.ClaimsContextKey{}, tok))) + }) + gql := client.New(withClaims) + + // 1) availableCloudEventTypes — the fetch summary surface. + var typesResp struct { + AvailableCloudEventTypes []typeRow `json:"availableCloudEventTypes"` + } + require.NoError(t, gql.Post(fmt.Sprintf( + `{ availableCloudEventTypes(subject: %q) { type count firstSeen lastSeen } }`, subject), + &typesResp)) + + // Cross-check the fetch summary against the raw ground truth. + require.Len(t, typesResp.AvailableCloudEventTypes, len(ground), + "fetch type count differs from raw_events") + gotTypes := map[string]typeRow{} + for _, r := range typesResp.AvailableCloudEventTypes { + gotTypes[r.Type] = r + g, ok := ground[r.Type] + require.Truef(t, ok, "fetch returned unknown type %q", r.Type) + require.Equalf(t, g.Count, r.Count, "count mismatch for %s", r.Type) + } + + // 2) latestCloudEvent (dimo.status) — header + data. + var latestResp map[string]any + require.NoError(t, gql.Post(fmt.Sprintf( + `{ latestCloudEvent(subject: %q, filter: {type: "dimo.status"}) { + header { type source id time producer } data } }`, subject), + &latestResp)) + + // 3) cloudEvents list (dimo.events, limit 5) — headers. + var listResp map[string]any + require.NoError(t, gql.Post(fmt.Sprintf( + `{ cloudEvents(subject: %q, limit: 5, filter: {type: "dimo.events"}) { + header { type source id time producer } } }`, subject), + &listResp)) + + // Stable, readable summary. + keys := make([]string, 0, len(gotTypes)) + for k := range gotTypes { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + g := ground[k] + r := gotTypes[k] + t.Logf(" %-36s fetch.count=%-6d raw.count=%-6d %s .. %s", k, r.Count, g.Count, r.FirstSeen, r.LastSeen) + } + + out := map[string]any{ + "system": "dq-fetch", + "subject": subject, + "availableCloudEventTypes": typesResp.AvailableCloudEventTypes, + "groundTruthTypes": ground, + "latestCloudEvent": latestResp["latestCloudEvent"], + "cloudEvents": listResp["cloudEvents"], + } + if outPath != "" { + b, err := json.MarshalIndent(out, "", " ") + require.NoError(t, err) + require.NoError(t, os.WriteFile(outPath, b, 0o644)) + t.Logf("wrote dq fetch results -> %s (%d bytes)", outPath, len(b)) + } +}