fix(#1724): chunk tx_last_seen_backfill with bounded reader yield

Replaces the single correlated UPDATE used by tx_last_seen_backfill_v1
(introduced in #1690) with a chunked loop that yields the single SQLite
writer between batches.

Symptom (pre-fix, operator scale ~71K tx / 1.5M obs / 2GB DB):
  - backgroundLoadComplete=true fires.
  - The async migration starts the single full-table UPDATE under
    SetMaxOpenConns(1), holds the writer for 10-15 minutes.
  - Every /api/healthz, /api/packets, /api/stats request queues behind
    sqlite_busy_timeout. UI appears frozen long after warm-up clears.

Fix (this commit):
  - cmd/ingestor/tx_last_seen_backfill.go (new):
      chunkedTxLastSeenBackfill snapshots MAX(id), counts eligible rows
      (last_seen=0 AND has observations AND id<=maxID), then loops
      bounded UPDATEs (batchSize=5000) with time.NewTimer-based sleeps
      (no Timer leak via time.After) between batches (yieldDelay=100ms).
      EXISTS gate skips orphan transmissions so the loop terminates.
      maxID snapshot keeps concurrent INSERTs out of scope (those are
      handled inline by stmtBumpTxLastSeen on the writer fast path).
      Ctx cancellation between batches returns context.Canceled with
      partial counts; partial commits are visible (migration does NOT
      flip to done).
      All errors propagate (snapshot, count, UPDATE, RowsAffected) —
      the migration cannot silently mark itself done.
      Progress callback fires per non-empty batch + once terminal with
      final stable counts; never on a stale n=0 batch.

  - cmd/ingestor/db.go: wire the helper into the tx_last_seen_backfill_v1
    async migration, explicit batchSize=5000, yieldDelay=100ms.

Math reality-check:
  ~71K tx / 5000 ≈ 15 batches × (~50ms exec + 100ms yield) ≈ ~2.5s
  wall time with readers slotted in at most every 150ms. PR #1725's
  description claimed ~300 batches × 150ms ≈ 45s — that confused
  observations (1.5M) with transmissions (71K); real number is ~20x
  smaller. Indexes idx_tx_last_seen (transmissions(last_seen)) and
  idx_observations_transmission_id already exist (see internal/dbschema
  and cmd/ingestor/db.go base schema) — no additional index work
  required at this commit.

Tests: cmd/ingestor/tx_last_seen_backfill_test.go (added in prior commit)
pin all the contract points reviewers flagged on PR #1725. Cancel-mid-loop
test timing widened from 30ms to 250ms to give the real chunked impl room
to commit a batch before the cancel fires; assertion semantics unchanged
(partial commits + context.Canceled + no full completion).

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)

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
 }

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()
 	}()