From a45e54db774f9ef0a239c5778da0d39ea00b6cd4 Mon Sep 17 00:00:00 2001 From: Steven Rhoods Date: Mon, 17 Aug 2026 23:28:25 +0100 Subject: [PATCH] Run VACUUM after a large DeleteJob purge to fix B-tree fragmentation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Purging a large job (millions of shards/splits/etc. rows deleted in one transaction) left the shards B-tree fragmented badly enough to visibly slow every later query against it. The existing auto_vacuum=INCREMENTAL pump (500 pages/30s) is sized for the Shard Reaper's steady trickle and doesn't catch up before the next large purge lands. DeleteJob now triggers a full VACUUM whenever the purge removes at least vacuumShardThreshold (1M) shard rows — below that, incremental vacuuming already handles it. Runs on its own goroutine (checkpoint TRUNCATE, then VACUUM) so the purge API call isn't held open for it, and coalesces concurrent/rapid DeleteJob calls onto a single pending run via a CAS-guarded flag, since a bulk purge can call DeleteJob once per matched job and VACUUM compacts the whole file regardless of which job triggered it — running it once per job in that loop would be pure waste. This is expensive by design, not despite the cost: VACUUM needs an exclusive lock for as long as the full-file rewrite takes (the same class of cost WALCheckpoint's TRUNCATE mode already documents at up to 144s for a much cheaper operation) and — since db is the sole writer connection — has to run there, so it holds mu and blocks every other write (and every rdb reader once the rewrite starts) for its complete duration. Moving it off the request path doesn't remove that cost, only defers it to a moment the caller isn't blocked on it. Tests cover the threshold gate, that VACUUM actually reclaims space (freelist_count -> 0), and that concurrent large purges coalesce into exactly one run — each verified by deliberately breaking the logic first and confirming the test fails for the expected reason. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01PsdNZLfmAFrMX2VUtkLtmm --- coordinator/internal/store/store.go | 81 +++++++++++- coordinator/internal/store/store_test.go | 149 +++++++++++++++++++++++ docs/DESIGN-coordinator.md | 24 ++++ 3 files changed, 253 insertions(+), 1 deletion(-) diff --git a/coordinator/internal/store/store.go b/coordinator/internal/store/store.go index 67def5f..e273aa8 100644 --- a/coordinator/internal/store/store.go +++ b/coordinator/internal/store/store.go @@ -15,6 +15,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "time" _ "modernc.org/sqlite" @@ -64,6 +65,20 @@ type Store struct { db *sql.DB // single writer connection; all mutations serialize on mu rdb *sql.DB // read-only connection pool; pure reads run here, lock-free mu sync.Mutex // single-writer discipline for db + + // vacuumPending coalesces DeleteJob's post-purge VACUUM trigger (see + // vacuumAfterLargeDelete): a bulk purge calls DeleteJob once per job, and + // without this a purge of N large terminal jobs in one request would queue + // N full-database VACUUMs back to back — wasted work, since a VACUUM + // compacts the whole file regardless of which job's rows triggered it, so + // only the first of any back-to-back run does anything useful. CAS-guarded + // so concurrent DeleteJob calls only ever start one pending run. + vacuumPending atomic.Bool + + // vacuumRuns counts completed VACUUM executions. Test-only instrument (no + // production reader) for asserting on coalescing directly rather than via + // side effects — see TestDeleteJobCoalescesConcurrentVacuums. + vacuumRuns atomic.Int64 } // lockWaitWarnThreshold: diagnostic for the lease-requeue investigation @@ -826,17 +841,81 @@ func (s *Store) DeleteJob(name string) (int64, error) { `DELETE FROM jobs WHERE id = ?`, } args := []any{id, id, id, id, id, id, id, id, id, id} + const shardsStmt = 3 // index of the `DELETE FROM shards` statement above + var shardsDeleted int64 for i, q := range stmts { - if _, err := tx.Exec(q, args[i]); err != nil { + res, err := tx.Exec(q, args[i]) + if err != nil { return 0, err } + if i == shardsStmt { + shardsDeleted, _ = res.RowsAffected() + } } if err := tx.Commit(); err != nil { return 0, err } + if shardsDeleted >= vacuumShardThreshold { + s.vacuumAfterLargeDelete() + } return id, nil } +// vacuumShardThreshold is how many shard rows a single DeleteJob purge must +// remove before it queues a full VACUUM (see vacuumAfterLargeDelete) — a +// purge below this is exactly what RunIncrementalVacuum's steady 500-page/30s +// trickle already handles; VACUUM's full-file rewrite is reserved for the +// scale where B-tree fragmentation from millions of deleted rows was +// observed live to fall off a performance cliff that incremental vacuuming +// alone did not recover from before the next large purge landed. +// var, not const, so a test can shorten it rather than seed a million rows. +var vacuumShardThreshold int64 = 1_000_000 + +// vacuumAfterLargeDelete queues a full VACUUM after a DeleteJob purge large +// enough to fragment the shards B-tree (vacuumShardThreshold). Runs +// asynchronously so the DeleteJob caller (the purge API handler) is not held +// open for it, but coalesces via vacuumPending: a bulk purge (purgeJobs) +// calls DeleteJob once per matched job, and VACUUM compacts the whole +// database file regardless of which job's rows triggered it, so any +// back-to-back purges within the same run should share one VACUUM rather +// than repeat it once per job. +// +// This is expensive on purpose, not despite the cost: VACUUM must hold an +// exclusive lock on the database for as long as the rewrite takes — the same +// class of cost WALCheckpoint's TRUNCATE mode documents (measured there at up +// to 144s at a 10-agent fleet for a far cheaper operation), except VACUUM +// rewrites the *entire* file, not just replays the WAL, so its hold is +// larger still and scales with total database size, not with what changed. +// Because s.db is the sole writer connection (MaxOpenConns(1)), VACUUM has +// to run there — a second connection would only contend with s.db and s.rdb +// for the same file lock, not avoid it — so this still takes s.mu for the +// full duration and stalls every other write (and, once VACUUM actually +// starts rewriting, every s.rdb reader too) fleet-wide until it completes. +// Moving this off the request path (the goroutine) does not remove that +// cost, only defers it to a moment the caller isn't blocked waiting on. +func (s *Store) vacuumAfterLargeDelete() { + if !s.vacuumPending.CompareAndSwap(false, true) { + return // a VACUUM is already queued or running; it will cover this purge too + } + go func() { + defer s.vacuumPending.Store(false) + if err := s.walCheckpoint("TRUNCATE", "VacuumAfterDelete-checkpoint"); err != nil { + slog.Warn("vacuum after delete: checkpoint failed", "err", err) + return + } + start := time.Now() + release := s.lockTimed("VacuumAfterDelete") + _, err := s.db.Exec(`VACUUM`) + release() + if err != nil { + slog.Warn("vacuum after delete: failed", "err", err, "elapsed_ms", time.Since(start).Milliseconds()) + return + } + s.vacuumRuns.Add(1) + slog.Info("vacuum after delete: complete", "elapsed_ms", time.Since(start).Milliseconds()) + }() +} + // --------------------------------------------------------------------------- // Passes // --------------------------------------------------------------------------- diff --git a/coordinator/internal/store/store_test.go b/coordinator/internal/store/store_test.go index fc6646f..9d1ab09 100644 --- a/coordinator/internal/store/store_test.go +++ b/coordinator/internal/store/store_test.go @@ -261,6 +261,155 @@ func TestDeleteJobRemovesJournalTypeCounts(t *testing.T) { } } +// waitVacuumIdle blocks until no VACUUM is pending/running, or fails the test +// after a generous timeout — vacuumAfterLargeDelete runs on its own +// goroutine, so a test asserting on its effects has to synchronize on +// vacuumPending rather than the DeleteJob call returning. +func waitVacuumIdle(t *testing.T, s *Store) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for s.vacuumPending.Load() { + if time.Now().After(deadline) { + t.Fatal("timed out waiting for pending VACUUM to finish") + } + time.Sleep(time.Millisecond) + } +} + +func freelistCount(t *testing.T, s *Store) int { + t.Helper() + var n int + if err := s.db.QueryRow(`PRAGMA freelist_count`).Scan(&n); err != nil { + t.Fatal(err) + } + return n +} + +// TestDeleteJobSkipsVacuumBelowThreshold: a purge of a small job must not +// queue a VACUUM at all — that scale is exactly what RunIncrementalVacuum's +// steady trickle already handles, and this guards against the threshold +// gate being dropped or inverted. +func TestDeleteJobSkipsVacuumBelowThreshold(t *testing.T) { + s := openTest(t) + old := vacuumShardThreshold + vacuumShardThreshold = 1_000_000 // seed's job deletes far fewer shards than this + defer func() { vacuumShardThreshold = old }() + + jobID, _, _ := seed(t, s) + if err := s.SetJobState(jobID, model.JobCompleted); err != nil { + t.Fatal(err) + } + if _, err := s.DeleteJob("t1"); err != nil { + t.Fatal(err) + } + if s.vacuumPending.Load() { + t.Error("DeleteJob queued a VACUUM for a purge below vacuumShardThreshold") + } +} + +// TestDeleteJobVacuumsAboveThreshold: a purge whose shard-row count meets +// vacuumShardThreshold must queue and complete a VACUUM — verified by +// checking freelist_count actually drops to zero, not just that the pending +// flag was set, so this fails if vacuumAfterLargeDelete's SQL is ever wrong +// (e.g. VACUUM silently erroring) as well as if the trigger itself is wrong. +func TestDeleteJobVacuumsAboveThreshold(t *testing.T) { + s := openTest(t) + old := vacuumShardThreshold + vacuumShardThreshold = 50 + defer func() { vacuumShardThreshold = old }() + + jobID, passID, _ := seed(t, s) // seed's own root shard, plus: + batch := make([]NewShard, 100) + for i := range batch { + batch[i] = NewShard{Kind: model.KindDir, RelPath: fmt.Sprintf("d%03d", i)} + } + if _, err := s.InsertShards(passID, 0, batch); err != nil { + t.Fatal(err) + } + if err := s.SetJobState(jobID, model.JobCompleted); err != nil { + t.Fatal(err) + } + if before := freelistCount(t, s); before != 0 { + t.Fatalf("freelist_count = %d before purge, want 0 (nothing deleted yet)", before) + } + + if _, err := s.DeleteJob("t1"); err != nil { + t.Fatal(err) + } + waitVacuumIdle(t, s) + + if after := freelistCount(t, s); after != 0 { + t.Errorf("freelist_count = %d after VACUUM, want 0 (VACUUM reclaims all free pages)", after) + } +} + +// TestDeleteJobCoalescesConcurrentVacuums: purging several large jobs back to +// back (as purgeJobs' bulk-purge loop does) must not queue one VACUUM per +// purge — VACUUM compacts the whole database regardless of which job +// triggered it, so overlapping DeleteJob calls should share a single run. +// Asserts on vacuumRuns directly (not just that the purge itself succeeded): +// without the CAS gate in vacuumAfterLargeDelete, N concurrent large purges +// still all complete correctly (proven by deliberately removing the gate and +// re-running this suite: purges still succeed, just with N redundant +// VACUUM/checkpoint calls racing each other) — coalescing is a waste-avoidance +// property, not a correctness one, so it needs its own direct assertion or a +// regression here would pass silently. +func TestDeleteJobCoalescesConcurrentVacuums(t *testing.T) { + s := openTest(t) + old := vacuumShardThreshold + vacuumShardThreshold = 50 + defer func() { vacuumShardThreshold = old }() + + const nJobs = 5 + for i := 0; i < nJobs; i++ { + job, err := s.CreateJob(fmt.Sprintf("bulk%d", i), []byte(specYAML), false, "") + if err != nil { + t.Fatal(err) + } + if err := s.SetJobState(job.ID, model.JobRunning); err != nil { + t.Fatal(err) + } + pass, err := s.CreatePass(job.ID, 1, model.PassScanning) + if err != nil { + t.Fatal(err) + } + batch := make([]NewShard, 60) + for j := range batch { + batch[j] = NewShard{Kind: model.KindDir, RelPath: fmt.Sprintf("d%03d", j)} + } + if _, err := s.InsertShards(pass.ID, 0, batch); err != nil { + t.Fatal(err) + } + if err := s.SetJobState(job.ID, model.JobCompleted); err != nil { + t.Fatal(err) + } + } + + var wg sync.WaitGroup + for i := 0; i < nJobs; i++ { + wg.Add(1) + go func(name string) { + defer wg.Done() + if _, err := s.DeleteJob(name); err != nil { + t.Error(err) + } + }(fmt.Sprintf("bulk%d", i)) + } + wg.Wait() + waitVacuumIdle(t, s) + + jobs, err := s.ListJobs() + if err != nil { + t.Fatal(err) + } + if len(jobs) != 0 { + t.Errorf("%d jobs survived concurrent purge, want 0", len(jobs)) + } + if runs := s.vacuumRuns.Load(); runs != 1 { + t.Errorf("vacuumRuns = %d, want exactly 1 — %d concurrent large purges should coalesce into a single VACUUM", runs, nJobs) + } +} + // A zero count for a type must not be written as a row (so it doesn't show up // as a spurious zero-count line in a caller that lists map keys) — mirrors the // omission behavior the WebUI/email rendering already relies on. diff --git a/docs/DESIGN-coordinator.md b/docs/DESIGN-coordinator.md index 9c16e7d..8edadc7 100644 --- a/docs/DESIGN-coordinator.md +++ b/docs/DESIGN-coordinator.md @@ -408,6 +408,30 @@ journal_cursors (pass_id, agent_id, acked_seq) -- JournalBatch flow control `auto_vacuum=INCREMENTAL` and a periodic `PRAGMA incremental_vacuum` pump reclaims the pages the reaper frees, so deleting rows actually shrinks the file instead of leaving freed-but-unreturned pages in it forever. +- **`DeleteJob` purges and B-tree fragmentation:** the steady incremental-vacuum + pump above is sized for the Shard Reaper's continuous trickle, not for `DeleteJob` + removing a whole large job's `shards`/`splits`/etc. rows in one transaction — + found live: a purge of a job with millions of shard rows left the `shards` B-tree + fragmented badly enough to visibly slow every later query against it, and the + 500-page/30s incremental pump did not catch up before the next large purge + landed. `DeleteJob` now runs a full `VACUUM` (`store.vacuumAfterLargeDelete`) + whenever the purge removed at least `vacuumShardThreshold` (1M) shard rows — + below that, incremental vacuuming already handles it and a full rewrite would be + needless cost. Deliberately expensive: `VACUUM` rewrites the *entire* database + file, not just the freed pages, and needs the same exclusive lock class + `WALCheckpoint`'s TRUNCATE mode does (§ above that section documents measuring + that alone at up to 144s of write-lock hold on a 10-agent fleet) — for the whole + file, this runs longer still, and since `db` is the sole writer connection + (`MaxOpenConns(1)`), it has to run there, so it holds `mu` for its complete + duration and blocks every other write (and, once the rewrite itself starts, + every `rdb` reader too) fleet-wide. Two things keep this from being worse than + it has to be: it runs on its own goroutine off the purge request's own path (the + operator's `DELETE /api/v1/jobs/{name}` call returns as soon as the row deletes + commit, not after the vacuum), and concurrent/rapid `DeleteJob` calls coalesce + onto a single pending run (`vacuumPending`, CAS-guarded) — a bulk purge + (`POST /api/v1/jobs/purge`) can call `DeleteJob` once per matched job, and since + `VACUUM` compacts the whole file regardless of which job's rows triggered it, + running it once per job in that loop would be pure waste, not extra safety. - **Eager SCANNING reap:** the transition-time reap above still leaves SCANNING's own DONE probe/dir/entrylist/chunk rows sitting in `shards` (and their `splits` rows) for the entire scan, since that phase alone can run for most of a large job's wall