diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index a5d54fa0..e5a1a05a 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -158,27 +158,29 @@ func OpenStoreWithInterval(dbPath string, sampleIntervalSec int) (*Store, error) } } - // #1690: backfill transmissions.last_seen from MAX(observations.timestamp) - // per transmission. The column is added inline by dbschema.Apply (cheap - // metadata-only ALTER); the populate query is potentially expensive - // (full obs scan + group) so we run it async. Subsequent observation - // inserts maintain the column inline (see InsertTransmission below). - // PREFLIGHT: async=true reason="full-table backfill JOIN (1.9M+ obs × 86k+ tx in prod) — must not block ingestor boot" + // #1690 (#1724 cold-load fix): backfill transmissions.last_seen from + // MAX(observations.timestamp) per transmission. The column is added + // inline by dbschema.Apply (cheap metadata-only ALTER); the populate + // query was originally a single correlated UPDATE that pinned the + // single writer for 10-15 min on operator-scale DBs (#1724). + // chunkedTxLastSeenBackfill runs the work in bounded chunks with a + // reader yield between batches so /api/healthz, packet reads, and + // ingest do not queue behind it. Math for ~71K transmissions: + // 71K / 5000 ≈ 15 batches × (~50ms exec + 100ms yield) ≈ ~2.5s wall + // with readers slotted in at most every 150ms. PR #1725 docs claimed + // "~300 batches × 150ms ≈ 45s" — that confused observations (1.5M) + // with transmissions (71K); the real number is ~20x smaller. + // Subsequent observation inserts maintain the column inline (see + // InsertTransmission + stmtBumpTxLastSeen below). + // PREFLIGHT: async=true reason="chunked backfill UPDATE; bounded reader yield every batch; safe at any scale" if err := s.RunAsyncMigration(context.Background(), "tx_last_seen_backfill_v1", func(ctx context.Context, d *sql.DB) error { - log.Println("[migration/async] Backfilling transmissions.last_seen from MAX(observations.timestamp)...") - res, err := d.ExecContext(ctx, ` - UPDATE transmissions - SET last_seen = COALESCE(( - SELECT MAX(timestamp) FROM observations WHERE transmission_id = transmissions.id - ), last_seen) - WHERE last_seen = 0 - `) + log.Println("[migration/async] Backfilling transmissions.last_seen (chunked, reader-yielding)...") + processed, total, err := chunkedTxLastSeenBackfill(ctx, d, 5000, 100*time.Millisecond, nil) if err != nil { return err } - n, _ := res.RowsAffected() - log.Printf("[migration/async] transmissions.last_seen backfill complete: %d rows updated", n) + log.Printf("[migration/async] transmissions.last_seen backfill complete: %d / %d rows", processed, total) return nil }); err != nil { log.Printf("[migration/async] scheduling tx_last_seen_backfill_v1 failed: %v", err) diff --git a/cmd/ingestor/tx_last_seen_backfill.go b/cmd/ingestor/tx_last_seen_backfill.go index 5d5a383e..16263ac9 100644 --- a/cmd/ingestor/tx_last_seen_backfill.go +++ b/cmd/ingestor/tx_last_seen_backfill.go @@ -1,19 +1,69 @@ // Package main: chunked tx_last_seen_v1 backfill (issue #1724). // -// Stub for the RED test commit — returns zero counts and nil error so the -// test compiles + runs to assertions (which then fail, proving the gate). -// The GREEN commit replaces this body with the real chunked implementation. +// Background: the original #1690 backfill was a single correlated UPDATE +// +// UPDATE transmissions +// SET last_seen = (SELECT MAX(timestamp) FROM observations WHERE transmission_id = transmissions.id) +// WHERE last_seen = 0 +// +// run under the ingestor's SetMaxOpenConns(1) writer. On real-size +// operator DBs (~71K transmissions / ~1.5M observations / ~2GB) it pinned +// the single writer connection for 10-15 minutes after +// backgroundLoadComplete=true, queueing every reader behind +// sqlite_busy_timeout. The result was an unresponsive system long after +// the warm-up banner cleared. +// +// Fix: chunk the work, yield the writer between chunks. Each chunk is one +// bounded UPDATE (size = batchSize transmissions). Between chunks we sleep +// yieldDelay so concurrent readers can slot in. The maxID snapshot taken +// at start ensures concurrent ingest (id > maxID) cannot trap us in an +// infinite loop. Orphan transmissions (no matching observation row) are +// filtered out via EXISTS so the loop always terminates. +// +// Math reality-check (for the operator-scale DB): +// - 71K transmissions, batchSize=5000 → ~15 chunks +// - per-chunk cost: bounded scan over (idx_tx_last_seen) + ~5K +// correlated MAX lookups on observations(transmission_id) → tens of ms +// - per-chunk yield: 100ms +// - wall time: ~15 * (~50ms + 100ms) ≈ 2-3s total, with readers slotted +// in every 150ms ceiling latency. +// - Previously: one ~10-15min UPDATE held the writer for the full duration. +// +// The original PR description claimed "300 batches × 150ms ≈ 45s" — that +// was wrong; it confused observations (1.5M) with transmissions (71K). +// The real number is ~20x smaller. See PR #1725 review history. package main import ( "context" "database/sql" + "fmt" "time" ) +// txBackfillProgressFn is invoked AFTER each non-empty batch with the running +// (processed, total) counts AND once more at the end with the final stable +// counts (terminal callback). It is NOT invoked on a stale n=0 batch. +// Callers must tolerate the terminal call being identical to the last +// in-flight call (this is a benign no-op for status surfaces). type txBackfillProgressFn func(processed, total int64) +// chunkedTxLastSeenBackfill is the orphan-safe, reader-yielding replacement +// for the single-statement #1690 backfill. +// +// Contract: +// - batchSize MUST be > 0 (rejected at the entrance — no <0 sentinel; explicit failure). +// - yieldDelay MUST be >= 0 (zero = no sleep, still releases the writer between Execs). +// - On ctx cancellation BETWEEN batches, the function returns +// context.Canceled (or ctx.Err()) with the partial-progress counts — +// it does NOT mark the migration done. +// - All errors propagate (snapshot maxID, count total, UPDATE, RowsAffected). +// - Progress callback fires per non-empty batch + once terminal with final counts. +// - Concurrent INSERTs (id > maxID) are intentionally skipped — they're +// handled inline by stmtBumpTxLastSeen on the writer fast-path. +// - Orphan transmissions (no observations) are skipped via EXISTS so the +// loop terminates deterministically. func chunkedTxLastSeenBackfill( ctx context.Context, db *sql.DB, @@ -21,11 +71,102 @@ func chunkedTxLastSeenBackfill( yieldDelay time.Duration, progress txBackfillProgressFn, ) (processed int64, total int64, err error) { - // Intentional no-op stub: real implementation lands in the GREEN commit. - _ = ctx - _ = db - _ = batchSize - _ = yieldDelay - _ = progress - return 0, 0, nil + if batchSize <= 0 { + return 0, 0, fmt.Errorf("chunkedTxLastSeenBackfill: batchSize must be > 0 (got %d)", batchSize) + } + if yieldDelay < 0 { + return 0, 0, fmt.Errorf("chunkedTxLastSeenBackfill: yieldDelay must be >= 0 (got %v)", yieldDelay) + } + + var maxID int64 + if err := db.QueryRowContext(ctx, + `SELECT COALESCE(MAX(id), 0) FROM transmissions`).Scan(&maxID); err != nil { + return 0, 0, fmt.Errorf("snapshot transmissions.max(id): %w", err) + } + + // Count only the rows we actually intend to touch (last_seen=0, has obs, + // id<=maxID). This gives operators an honest denominator. + if err := db.QueryRowContext(ctx, ` + SELECT COUNT(*) FROM transmissions t + WHERE t.last_seen = 0 + AND t.id <= ? + AND EXISTS (SELECT 1 FROM observations o WHERE o.transmission_id = t.id) + `, maxID).Scan(&total); err != nil { + return 0, 0, fmt.Errorf("count backfill total: %w", err) + } + + if total == 0 { + if progress != nil { + progress(0, 0) + } + return 0, 0, nil + } + + for { + if cerr := ctx.Err(); cerr != nil { + if progress != nil { + progress(processed, total) + } + return processed, total, cerr + } + + // One bounded chunk. The inner ORDER BY id + LIMIT gives deterministic + // forward progress; EXISTS skips orphans; id<=maxID keeps concurrent + // inserts out of scope so the loop terminates. + res, uerr := db.ExecContext(ctx, ` + UPDATE transmissions + SET last_seen = ( + SELECT MAX(timestamp) FROM observations + WHERE transmission_id = transmissions.id + ) + WHERE id IN ( + SELECT id FROM transmissions + WHERE last_seen = 0 + AND id <= ? + AND EXISTS ( + SELECT 1 FROM observations WHERE transmission_id = transmissions.id + ) + ORDER BY id + LIMIT ? + ) + `, maxID, batchSize) + if uerr != nil { + return processed, total, fmt.Errorf("chunk update: %w", uerr) + } + n, raErr := res.RowsAffected() + if raErr != nil { + return processed, total, fmt.Errorf("chunk RowsAffected: %w", raErr) + } + if n == 0 { + // All eligible rows processed (or none left). Do NOT fire a stale + // progress here; terminal fire happens once outside the loop. + break + } + processed += n + if progress != nil { + progress(processed, total) + } + + if yieldDelay > 0 { + t := time.NewTimer(yieldDelay) + select { + case <-ctx.Done(): + if !t.Stop() { + <-t.C + } + if progress != nil { + progress(processed, total) + } + return processed, total, ctx.Err() + case <-t.C: + // timer fired; loop continues. + } + } + } + + // Terminal callback: single fire with final stable counts. + if progress != nil { + progress(processed, total) + } + return processed, total, nil } diff --git a/cmd/ingestor/tx_last_seen_backfill_test.go b/cmd/ingestor/tx_last_seen_backfill_test.go index b0cf21b3..eeb68507 100644 --- a/cmd/ingestor/tx_last_seen_backfill_test.go +++ b/cmd/ingestor/tx_last_seen_backfill_test.go @@ -172,9 +172,11 @@ func TestChunkedBackfill_CtxCancelMidLoop(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) - // Cancel shortly after the first batch starts. + // Cancel after first batches have committed but before the loop + // is done. With batchSize=1000 and yieldDelay=80ms, expect several + // batches to complete before the 250ms cancel fires. go func() { - time.Sleep(30 * time.Millisecond) + time.Sleep(250 * time.Millisecond) cancel() }()