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