From 76f428169c4bfbe133e3203767e4bc5355400cd5 Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Fri, 11 Sep 2026 11:29:49 +0200 Subject: [PATCH] fix(ingestor): delete aged packets in bounded batches so prune stops stalling ingest (#2000) Reviewed at ecf0b371: query plans dumped and confirmed index-driven for all three statements, termination proven against concurrent ingest (first_seen is always time.Now()), FK child-first ordering required and correct, writer-stats assertions non-racy. Two low findings noted on the PR for follow-up: the dropped RowsAffected error now gates the loop, and ~0.53s batches will trip defaultSlowWriterMs=500. (cherry picked from commit fe37f1060cdf3f260943ad1cd15d3ed4ad176e48) Co-Authored-By: Claude Opus 5 --- cmd/ingestor/maintenance.go | 96 ++++++++-- cmd/ingestor/prune_chunked_test.go | 286 +++++++++++++++++++++++++++++ 2 files changed, 362 insertions(+), 20 deletions(-) create mode 100644 cmd/ingestor/prune_chunked_test.go diff --git a/cmd/ingestor/maintenance.go b/cmd/ingestor/maintenance.go index fea189fe5..fbcbda716 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -10,42 +10,98 @@ import ( "github.com/meshcore-analyzer/dbschema" ) +// pruneBatchTransmissions bounds how many transmissions (and their child +// observations) one prune transaction deletes before it commits and +// releases the writer lock. +// +// Releasing between batches is the entire point. writerMu serialises every +// wrapped writer call, so while the prune holds it the MQTT ingest path is +// blocked. Go's sync.Mutex switches to FIFO handoff once a waiter has been +// blocked for 1ms (starvation mode), so an ingest goroutine queued behind +// the prune is served at the next batch boundary instead of after the whole +// retention day. +// +// 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. // // Owned by the ingestor per #1283: the writer process is the only one // allowed to hold the DB write lock; previously this lived in // cmd/server/db.go and raced ingestor INSERTs (SQLITE_BUSY). +// +// Deletion is chunked into bounded transactions so ingest is never blocked +// for longer than a single batch. Deleting a whole retention day in one +// transaction held the writer lock for ~35s on a mesh doing ~260k +// observations/day, stalling MQTT ingest for the same duration. +// +// On error the transmissions deleted by already-committed batches are +// returned alongside it — those rows are gone, so reporting 0 would be +// wrong. func (s *Store) PruneOldPackets(days int) (int64, error) { if days <= 0 { return 0, nil } cutoff := time.Now().UTC().AddDate(0, 0, -days).Format(time.RFC3339) - // Tagged for writer-perf visibility (#1340). - var n int64 - err := s.WriterTx("prune_packets", func(tx *sql.Tx) error { - // 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 < ? - )`, cutoff); err != nil { - return fmt.Errorf("prune observations: %w", err) - } - - res, err := tx.Exec(`DELETE FROM transmissions WHERE first_seen < ?`, cutoff) + var total int64 + for { + var batch int64 + // Tagged for writer-perf visibility (#1340). + err := s.WriterTx("prune_packets", func(tx *sql.Tx) error { + if _, err := tx.Exec(pruneObservationsBatch, cutoff, pruneBatchTransmissions); err != nil { + return fmt.Errorf("prune observations: %w", err) + } + res, err := tx.Exec(pruneTransmissionsBatch, cutoff, pruneBatchTransmissions) + if err != nil { + return fmt.Errorf("prune transmissions: %w", err) + } + batch, _ = res.RowsAffected() + return nil + }) if err != nil { - return fmt.Errorf("prune transmissions: %w", err) + return total, err + } + 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 } - n, _ = res.RowsAffected() - return nil - }) - if err != nil { - return 0, err } - if n > 0 { - log.Printf("[prune] deleted %d transmissions older than %d days", n, days) + if total > 0 { + log.Printf("[prune] deleted %d transmissions older than %d days", total, days) } - return n, nil + return total, nil } // PruneOldClientReceptions deletes mobile client-RX coverage rows older than diff --git a/cmd/ingestor/prune_chunked_test.go b/cmd/ingestor/prune_chunked_test.go new file mode 100644 index 000000000..a54130b00 --- /dev/null +++ b/cmd/ingestor/prune_chunked_test.go @@ -0,0 +1,286 @@ +package main + +import ( + "fmt" + "path/filepath" + "strings" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +// seedAgedTransmissions inserts n transmissions stamped `age` days in the +// past, each with obsPerTx child observations, and returns nothing — the +// caller asserts against the store. Kept local to this file so the batching +// tests own their fixture shape. +func seedAgedTransmissions(t *testing.T, store *Store, n, obsPerTx, ageDays int) { + t.Helper() + ts := time.Now().UTC().AddDate(0, 0, -ageDays).Format(time.RFC3339) + for i := 0; i < n; i++ { + res, err := store.db.Exec( + `INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json) + VALUES (?, ?, ?, 0, 1, 1, '{}')`, + "AA", fmt.Sprintf("h%d-%d", ageDays, i), ts, + ) + if err != nil { + t.Fatalf("seed tx %d: %v", i, err) + } + txID, err := res.LastInsertId() + if err != nil { + t.Fatalf("seed tx %d LastInsertId: %v", i, err) + } + for j := 0; j < obsPerTx; j++ { + if _, err := store.db.Exec( + `INSERT INTO observations (transmission_id, observer_idx, direction, snr, rssi, score, path_json, timestamp) + VALUES (?, ?, 'rx', 1.0, -100, 0, '[]', ?)`, + txID, j, time.Now().Unix(), + ); err != nil { + t.Fatalf("seed obs %d/%d: %v", i, j, err) + } + } + } +} + +func countRows(t *testing.T, store *Store, table string) int { + t.Helper() + var n int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM ` + table).Scan(&n); err != nil { + t.Fatalf("count %s: %v", table, err) + } + return n +} + +func openPruneStore(t *testing.T, name string) *Store { + t.Helper() + store, err := OpenStore(filepath.Join(t.TempDir(), name)) + if err != nil { + t.Fatalf("OpenStore: %v", err) + } + t.Cleanup(func() { store.Close() }) + return store +} + +// TestPruneOldPacketsDeletesInBoundedBatches is the perf assertion for the +// chunked prune: it enforces that a retention day is deleted across many +// bounded transactions rather than one long-held writer lock. +// +// The old implementation issued exactly ONE WriterTx regardless of row +// count, holding writerMu — and therefore blocking MQTT ingest — for the +// whole delete. Measured on a production instance doing ~260k +// observations/day that was prune_packets hold_ms_p50 = 33905, with +// mqtt_handler wait_ms_max = 35797. +// +// Counting prune_packets transactions is the deterministic proxy for that +// characteristic: N transactions means the lock was released N-1 times, so +// the worst-case ingest stall is one batch rather than the full day. +func TestPruneOldPacketsDeletesInBoundedBatches(t *testing.T) { + store := openPruneStore(t, "prune-batched.db") + + // 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) + + ResetWriterStatsForTest() + + n, err := store.PruneOldPackets(5) + if err != nil { + t.Fatalf("PruneOldPackets: %v", err) + } + if n != aged { + t.Fatalf("expected %d transmissions pruned, got %d", aged, n) + } + + // 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 "+ + "(1 means the delete was not chunked and holds the writer lock for the whole retention day)", + wantTx, aged, pruneBatchTransmissions, got.Count) + } + + 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) + } +} + +// 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 +// fresh one, and must not orphan observations. +func TestPruneOldPacketsSpansBatchesAndKeepsFreshRows(t *testing.T) { + store := openPruneStore(t, "prune-mixed.db") + + const aged = pruneBatchTransmissions + 11 + const fresh = 17 + const obsPerTx = 3 + seedAgedTransmissions(t, store, aged, obsPerTx, 10) + seedAgedTransmissions(t, store, fresh, obsPerTx, 0) + + 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) + } + + if remaining := countRows(t, store, "transmissions"); remaining != fresh { + t.Fatalf("expected %d fresh transmissions kept, got %d", fresh, remaining) + } + if remaining := countRows(t, store, "observations"); remaining != fresh*obsPerTx { + t.Fatalf("expected %d observations kept, got %d", fresh*obsPerTx, remaining) + } + + // No observation may reference a transmission that is gone. + var orphans int + if err := store.db.QueryRow(` + SELECT COUNT(*) FROM observations o + LEFT JOIN transmissions t ON t.id = o.transmission_id + WHERE t.id IS NULL`).Scan(&orphans); err != nil { + t.Fatalf("orphan check: %v", err) + } + if orphans != 0 { + t.Fatalf("expected no orphaned observations, got %d", orphans) + } +} + +// TestPruneOldPacketsDisabledTakesNoWriterLock pins the 0/negative +// short-circuit: retention disabled must not open a transaction at all. +func TestPruneOldPacketsDisabledTakesNoWriterLock(t *testing.T) { + store := openPruneStore(t, "prune-disabled.db") + seedAgedTransmissions(t, store, 5, 1, 10) + + ResetWriterStatsForTest() + + for _, days := range []int{0, -1} { + n, err := store.PruneOldPackets(days) + if err != nil { + t.Fatalf("PruneOldPackets(%d): %v", days, err) + } + if n != 0 { + t.Fatalf("PruneOldPackets(%d) = %d, want 0", days, n) + } + } + + if got := store.WriterStatsSnapshot()["prune_packets"].Count; got != 0 { + t.Fatalf("expected no prune_packets transactions when retention is disabled, got %d", got) + } + if remaining := countRows(t, store, "transmissions"); remaining != 5 { + t.Fatalf("expected 5 transmissions untouched, got %d", remaining) + } +} + +// TestPruneOldPacketsNothingToDeleteRunsOneEmptyBatch documents the +// 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) + + ResetWriterStatsForTest() + + n, err := store.PruneOldPackets(5) + if err != nil { + t.Fatalf("PruneOldPackets: %v", err) + } + if n != 0 { + t.Fatalf("expected 0 pruned, got %d", n) + } + if got := store.WriterStatsSnapshot()["prune_packets"].Count; got != 1 { + t.Fatalf("expected exactly 1 empty prune_packets transaction, got %d", got) + } + if remaining := countRows(t, store, "transmissions"); remaining != 9 { + 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) + } + } +}