From ecf0b371ba23aeb8eb4806c10ec7e99107444f1a Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Fri, 11 Sep 2026 07:03:48 +0200 Subject: [PATCH] fix(ingestor): walk the prune batch off idx_transmissions_first_seen Review follow-up for #2000, found in review by @efiten. Ordering the batch subquery by id made SQLite abandon idx_transmissions_first_seen for a rowid SCAN. While rows are being deleted that costs nothing - the oldest rows have the lowest rowids and match at once - but the batch that finds nothing walks the whole table under writerMu, once per statement. That empty batch is the steady state on any instance where nothing has aged out yet. Reproduced on 1M transmissions: ORDER BY id SCAN transmissions empty 73.22ms ORDER BY first_seen, id COVERING INDEX (first_seen --- cmd/ingestor/maintenance.go | 55 +++++++++++------ cmd/ingestor/prune_chunked_test.go | 98 ++++++++++++++++++++++++++++-- 2 files changed, 127 insertions(+), 26 deletions(-) diff --git a/cmd/ingestor/maintenance.go b/cmd/ingestor/maintenance.go index aa09a0eb..07d7e1ba 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -21,12 +21,36 @@ import ( // the prune is served at the next batch boundary instead of after the whole // retention day. // -// 250 keeps a batch near ~4k observation deletes at a typical ~16 -// observations per transmission — a few hundred milliseconds — while -// keeping the number of commits low (a 16k-transmission day is 64 -// transactions, not 16k). +// The bound is on transmissions, but hold time scales with the rows actually +// deleted, and each transmission carries an unbounded number of observations. +// At ~16 observations per transmission a batch is ~4k row deletes and a few +// hundred milliseconds; an instance with a denser observation ratio gets a +// proportionally longer hold from the same batch size. 250 also keeps the +// commit count low (a 16k-transmission day is 64 transactions, not 16k). const pruneBatchTransmissions = 250 +// pruneAgedTransmissionIDs selects the next batch of transmissions older than +// the cutoff. Both statements of a batch embed it, so they resolve the same +// set: nothing modifies `transmissions` between them inside the transaction. +// +// The ORDER BY must be satisfiable from idx_transmissions_first_seen. That +// index carries the rowid as its tiebreaker, so "first_seen, id" is walked +// straight off it and the LIMIT stays deterministic even when timestamps tie. +// Ordering by id alone looks equivalent but makes SQLite abandon the index for +// a rowid SCAN. That is harmless while rows are being deleted — the oldest +// rows have the lowest rowids and match at once — but the batch that finds +// nothing, which is the steady state whenever nothing has aged out, walks the +// whole table under writerMu. TestPruneAgedTransmissionIDsUsesFirstSeenIndex +// pins the plan. +const pruneAgedTransmissionIDs = `SELECT id FROM transmissions WHERE first_seen < ? ORDER BY first_seen, id LIMIT ?` + +// The two statements of one prune batch. Child observations go first (no +// CASCADE in SQLite). +const ( + pruneObservationsBatch = `DELETE FROM observations WHERE transmission_id IN (` + pruneAgedTransmissionIDs + `)` + pruneTransmissionsBatch = `DELETE FROM transmissions WHERE id IN (` + pruneAgedTransmissionIDs + `)` +) + // PruneOldPackets deletes transmissions (and their child observations) // older than `days`. Returns count of transmissions deleted. // @@ -53,20 +77,10 @@ func (s *Store) PruneOldPackets(days int) (int64, error) { var batch int64 // Tagged for writer-perf visibility (#1340). err := s.WriterTx("prune_packets", func(tx *sql.Tx) error { - // Both statements resolve the same bounded set: nothing modifies - // `transmissions` between them inside this transaction, and - // ORDER BY id makes the LIMIT deterministic. - // - // Delete child observations first (no CASCADE in SQLite). - if _, err := tx.Exec(`DELETE FROM observations WHERE transmission_id IN ( - SELECT id FROM transmissions WHERE first_seen < ? ORDER BY id LIMIT ? - )`, cutoff, pruneBatchTransmissions); err != nil { + if _, err := tx.Exec(pruneObservationsBatch, cutoff, pruneBatchTransmissions); err != nil { return fmt.Errorf("prune observations: %w", err) } - - res, err := tx.Exec(`DELETE FROM transmissions WHERE id IN ( - SELECT id FROM transmissions WHERE first_seen < ? ORDER BY id LIMIT ? - )`, cutoff, pruneBatchTransmissions) + res, err := tx.Exec(pruneTransmissionsBatch, cutoff, pruneBatchTransmissions) if err != nil { return fmt.Errorf("prune transmissions: %w", err) } @@ -76,12 +90,13 @@ func (s *Store) PruneOldPackets(days int) (int64, error) { if err != nil { return total, err } - // A batch that deleted nothing means no rows are left below the - // cutoff; every batch before it deleted exactly what it selected. - if batch == 0 { + total += batch + // A short batch proves nothing is left below the cutoff: the subquery + // found fewer rows than it was allowed to take. Only a batch that came + // back exactly full needs another pass. + if batch < pruneBatchTransmissions { break } - total += batch } if total > 0 { log.Printf("[prune] deleted %d transmissions older than %d days", total, days) diff --git a/cmd/ingestor/prune_chunked_test.go b/cmd/ingestor/prune_chunked_test.go index 52ebd1eb..a54130b0 100644 --- a/cmd/ingestor/prune_chunked_test.go +++ b/cmd/ingestor/prune_chunked_test.go @@ -3,6 +3,7 @@ package main import ( "fmt" "path/filepath" + "strings" "testing" "time" @@ -76,8 +77,8 @@ func openPruneStore(t *testing.T, name string) *Store { func TestPruneOldPacketsDeletesInBoundedBatches(t *testing.T) { store := openPruneStore(t, "prune-batched.db") - // Two full batches plus a partial one, so the loop's remainder path and - // its terminating empty batch are both exercised. + // Two full batches plus a partial one. The partial batch is what ends the + // loop, so no terminating empty batch should run. const aged = pruneBatchTransmissions*2 + 37 seedAgedTransmissions(t, store, aged, 2, 10) @@ -91,8 +92,8 @@ func TestPruneOldPacketsDeletesInBoundedBatches(t *testing.T) { t.Fatalf("expected %d transmissions pruned, got %d", aged, n) } - // 2 full batches + 1 partial + 1 empty batch that terminates the loop. - wantTx := int64(4) + // 2 full batches + 1 partial batch, which proves nothing is left. + wantTx := int64(3) got := store.WriterStatsSnapshot()["prune_packets"] if got.Count != wantTx { t.Fatalf("expected %d prune_packets transactions for %d rows at batch size %d, got %d "+ @@ -108,6 +109,38 @@ func TestPruneOldPacketsDeletesInBoundedBatches(t *testing.T) { } } +// TestPruneOldPacketsExactMultipleTerminates covers the one case the +// short-batch exit cannot catch on its own: when the aged rows are an exact +// multiple of the batch size, the last batch comes back full, so the loop +// must run one more — empty — batch to learn that nothing is left. +func TestPruneOldPacketsExactMultipleTerminates(t *testing.T) { + store := openPruneStore(t, "prune-exact.db") + + const aged = pruneBatchTransmissions * 2 + seedAgedTransmissions(t, store, aged, 1, 10) + + ResetWriterStatsForTest() + + n, err := store.PruneOldPackets(5) + if err != nil { + t.Fatalf("PruneOldPackets: %v", err) + } + if n != aged { + t.Fatalf("expected %d pruned, got %d", aged, n) + } + + // 2 full batches + 1 empty batch that finds nothing and ends the loop. + if got := store.WriterStatsSnapshot()["prune_packets"].Count; got != 3 { + t.Fatalf("expected 3 prune_packets transactions for an exact multiple of the batch size, got %d", got) + } + if remaining := countRows(t, store, "transmissions"); remaining != 0 { + t.Fatalf("expected all aged transmissions gone, %d remain", remaining) + } + if remaining := countRows(t, store, "observations"); remaining != 0 { + t.Fatalf("expected all child observations gone, %d remain", remaining) + } +} + // TestPruneOldPacketsSpansBatchesAndKeepsFreshRows covers the correctness // risk the batching introduces: with a LIMIT on both statements, a cutoff // that straddles several batches must still delete every aged row and no @@ -176,8 +209,10 @@ func TestPruneOldPacketsDisabledTakesNoWriterLock(t *testing.T) { } // TestPruneOldPacketsNothingToDeleteRunsOneEmptyBatch documents the -// steady-state cost when nothing has aged out yet: a single empty batch, -// held for microseconds, then the loop exits. +// steady-state cost when nothing has aged out yet: a single empty batch, then +// the loop exits. That batch is only cheap because the subquery is walked off +// idx_transmissions_first_seen — see +// TestPruneAgedTransmissionIDsUsesFirstSeenIndex. func TestPruneOldPacketsNothingToDeleteRunsOneEmptyBatch(t *testing.T) { store := openPruneStore(t, "prune-noop.db") seedAgedTransmissions(t, store, 9, 2, 0) @@ -198,3 +233,54 @@ func TestPruneOldPacketsNothingToDeleteRunsOneEmptyBatch(t *testing.T) { t.Fatalf("expected 9 transmissions kept, got %d", remaining) } } + +// TestPruneAgedTransmissionIDsUsesFirstSeenIndex pins the query plan of the +// batch subquery and of both statements that embed it. +// +// Ordering the batch by id instead of first_seen makes SQLite drop +// idx_transmissions_first_seen for a rowid SCAN. On a 1M-row table that took +// the terminating, nothing-left batch from ~10µs to ~73ms — once per statement, +// under writerMu, in the state an instance is in whenever nothing has aged out. +// Transaction counts cannot see that regression, so the plan is the assertion. +func TestPruneAgedTransmissionIDsUsesFirstSeenIndex(t *testing.T) { + store := openPruneStore(t, "prune-plan.db") + seedAgedTransmissions(t, store, 20, 2, 10) + + cutoff := time.Now().UTC().AddDate(0, 0, -5).Format(time.RFC3339) + for name, q := range map[string]string{ + "batch subquery": pruneAgedTransmissionIDs, + "observations delete": pruneObservationsBatch, + "transmissions delete": pruneTransmissionsBatch, + } { + rows, err := store.db.Query("EXPLAIN QUERY PLAN "+q, cutoff, pruneBatchTransmissions) + if err != nil { + t.Fatalf("%s: EXPLAIN QUERY PLAN: %v", name, err) + } + var steps []string + for rows.Next() { + var id, parent, notused int + var detail string + if err := rows.Scan(&id, &parent, ¬used, &detail); err != nil { + rows.Close() + t.Fatalf("%s: scan plan row: %v", name, err) + } + steps = append(steps, detail) + } + if err := rows.Err(); err != nil { + rows.Close() + t.Fatalf("%s: plan rows: %v", name, err) + } + rows.Close() + plan := strings.Join(steps, " | ") + + if !strings.Contains(plan, "idx_transmissions_first_seen") { + t.Errorf("%s: plan does not use idx_transmissions_first_seen: %s", name, plan) + } + if strings.Contains(plan, "SCAN transmissions") { + t.Errorf("%s: plan scans transmissions, so the empty terminating batch walks the whole table under writerMu: %s", name, plan) + } + if strings.Contains(plan, "TEMP B-TREE") { + t.Errorf("%s: plan sorts in a temp b-tree instead of walking the index in order: %s", name, plan) + } + } +}