dandri/meshcore-analyzer
1
0
Fork 0
mirror of https://github.com/Kpa-clawbot/meshcore-analyzer.git synced 2026-09-27 07:09:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingestor): walk the prune batch off idx_transmissions_first_seen

Browse Source 
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<?)    empty 10us

Order by (first_seen, id) instead. The index carries the rowid as its
tiebreaker, so the plan is identical to plain first_seen while the LIMIT
stays deterministic on tied timestamps. Both DELETE statements now build
from one shared subquery constant, so they cannot drift apart.

End the loop on a short batch rather than an empty one: a batch that
returns fewer rows than its limit already proves nothing is left below
the cutoff. That saves the terminating round-trip in every run except
when the aged rows are an exact multiple of the batch size.

Document that hold time scales with observations per transmission, not
with the transmission bound.

Tests:
- TestPruneAgedTransmissionIDsUsesFirstSeenIndex asserts the plan of the
  subquery and of both statements: uses idx_transmissions_first_seen, no
  SCAN transmissions, no temp b-tree sort. It fails against ORDER BY id
  on all three statements; the transaction-count tests cannot see it.
- TestPruneOldPacketsExactMultipleTerminates covers the one case the
  short-batch exit cannot catch.
- TestPruneOldPacketsDeletesInBoundedBatches now expects 3 transactions
  (2 full + 1 partial) instead of 4.

go test -race ./... passes on origin/master plus this change
(492.7s, 0 data races).

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
liquidraver
2026-09-11 07:03:48 +02:00
co-authored by Claude Opus 5
parent 56fa658ac2
commit ecf0b371ba
2 changed files with 127 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/ingestor/maintenance.go
+35 -20
View File
@@ -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)
cmd/ingestor/prune_chunked_test.go
+92 -6
View File
@@ -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, &notused, &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)
}
}
}