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