Run VACUUM after a large DeleteJob purge to fix B-tree fragmentation

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
 // ---------------------------------------------------------------------------

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.

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