diff --git a/TODO.md b/TODO.md index edd9f11..e42194b 100644 --- a/TODO.md +++ b/TODO.md @@ -47,11 +47,22 @@ `walkPhase` always drains its events to close, but it has the same unbounded-send shape and #5 will give it an early return, so it gets the same treatment plus a `ctx.Err()` guard after the walk: a - cancelled walk yields a partial size census, and the update phase - would read every unreached file as vanished and delete its record. - New tests drive `run(scan)` against a database whose insert trigger - aborts, and assert both that the scan fails instead of hanging and - that `runtime.NumGoroutine()` polls back to its pre-scan baseline + cancelled walk yields a partial size census, and every file it never + reached looks vanished to the update phase. That phase's own + `BeginTx` fails on the same cancelled context before deleting + anything, so the guard is defence in depth rather than the only + barrier — but it is the one that survives #5 deciding an interrupted + scan may commit what it has. Tests drive `run(scan)` against a + database whose insert trigger aborts, and assert both that the scan + fails instead of hanging and that `runtime.NumGoroutine()` polls + back to its pre-scan baseline; a second set cancels a scan part-way + through the walk — deterministically, by counting the scan's own + consultations of `ctx.Done()` rather than racing a timer — and + asserts that it stops at the guard holding a partial census and a + still-populated record index, with every record intact. The + remaining cancellation branches of both pools are covered by direct + tests of `sendEvent`, the walk workers, `dispatchDirs`, + `feedHashJobs`, `hashWorker` and `hashPhase` - guarantee the database is closed on every fatal exit path (2026-08-09, branch `db-close-on-fatal`, closes #4): `fatalf` and its `os.Exit(1)` are gone, so the deferred `db.Close()` — and with diff --git a/cancel_test.go b/cancel_test.go new file mode 100644 index 0000000..b894228 --- /dev/null +++ b/cancel_test.go @@ -0,0 +1,489 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "os" + "path/filepath" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" +) + +// poolUnwind bounds how long a goroutine is given to leave a pool +// after its context is cancelled. Only a failing run ever waits this +// long: a pool that ignored its cancellation parks forever, and this +// is what turns that into a failed assertion instead of a suite that +// hangs until the test binary's own timeout. +const poolUnwind = 2 * time.Second + +// walkClock is a context whose cancellation is driven by the scan's +// own progress rather than by the wall clock: it cancels itself the +// moment its Done method has been consulted n times. That is what +// makes "cancel in the middle of the walk" reproducible instead of a +// race against a timer. +// +// The accounting behind the n chosen by each test: every blocking +// channel operation in the walk selects on Done, so the walk spends +// one consultation per file event plus a couple per directory, while +// the index load that runs ahead of it spends a small fixed number +// (three) whatever the record count. +type walkClock struct { + n int64 + seen atomic.Int64 + once sync.Once + done chan struct{} +} + +// newWalkClock returns a context that cancels itself on the nth +// consultation of its Done method. +func newWalkClock(n int64) *walkClock { + return &walkClock{n: n, done: make(chan struct{})} +} + +// Done returns the cancellation channel, cancelling the context on the +// nth call and on every call after it. The same channel is returned +// throughout, so a caller that took it before the cancellation still +// observes the close. +func (c *walkClock) Done() <-chan struct{} { + if c.seen.Add(1) >= c.n { + c.once.Do(func() { close(c.done) }) + } + + return c.done +} + +// Err reports the cancellation without consuming a consultation, which +// is what lets the post-walk guard read it without disturbing the +// count. +func (c *walkClock) Err() error { + select { + case <-c.done: + return context.Canceled + default: + return nil + } +} + +// Deadline reports no deadline: this context is cancelled by progress, +// never by time. +func (c *walkClock) Deadline() (time.Time, bool) { + return time.Time{}, false +} + +// Value carries nothing. +func (c *walkClock) Value(_ any) any { + return nil +} + +// walkCancelDirs and walkCancelFilesPerDir shape the fixture for the +// mid-walk cancellation test. Spreading the files over directories is +// load-bearing: it is what bounds how much of the tree can still be +// walked after the cancellation, since the workers drop every +// directory still queued and only the handful already in flight can +// emit anything more. +const ( + walkCancelDirs = 100 + walkCancelFilesPerDir = 20 + walkCancelFiles = walkCancelDirs * walkCancelFilesPerDir + walkCancelWorkers = 4 + walkCancelInFlightDirs = walkCancelWorkers * walkCancelFilesPerDir +) + +// walkCancelAtDone is the consultation on which the fixture's context +// cancels itself. A quarter of the file count is far past the index +// load's fixed handful and far short of the walk's total, so the +// cancellation lands deep inside the walk and nowhere near either end +// of it. +const walkCancelAtDone = walkCancelFiles / 4 + +// buildWalkCancelTree writes walkCancelFiles empty files spread over +// walkCancelDirs subdirectories. Zero-length files are never opened by +// the hasher, so the fixture costs directory entries and no read I/O +// while still giving the walk thousands of events to emit. +func buildWalkCancelTree(t *testing.T) string { + t.Helper() + + dir := t.TempDir() + + for i := range walkCancelDirs { + sub := filepath.Join(dir, "d"+strconv.Itoa(i)) + + err := os.Mkdir(sub, 0o750) + if err != nil { + t.Fatal(err) + } + + writeEmptyFiles(t, sub, walkCancelFilesPerDir) + } + + return dir +} + +// assertRecordsIntact fails when the database no longer holds exactly +// the records it held before, reporting the first difference rather +// than dumping thousands of paths. +func assertRecordsIntact(t *testing.T, db *sql.DB, before []string) { + t.Helper() + + got := recordPaths(dbRecords(t, db)) + if len(got) != len(before) { + t.Fatalf("%d records after the cancelled scan, want %d", + len(got), len(before)) + } + + for i := range got { + if got[i] != before[i] { + t.Fatalf("record %d = %q after the cancelled scan, want %q", + i, got[i], before[i]) + } + } +} + +// TestSyncScanCancelledMidWalkKeepsRecords is the regression net under +// the post-walk guard. The scan is cancelled part-way through the +// walk, so it reaches the guard holding a genuinely partial size +// census and a still-populated index of records the walk never got to. +// Every one of those records would look vanished to the update phase. +// The guard is what stops the scan there, and this test is what +// notices if it stops doing so: deleting the guard, or making it +// unreachable, makes the scan carry its truncated view into a later +// phase and fail there instead, with a wrapped error rather than the +// bare cancellation. +// +//nolint:paralleltest // counts goroutines: must not run beside others +func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) { + dir := buildWalkCancelTree(t) + db := openTestDB(t) + + st := syncTree(t, db, dir) + if st.added != walkCancelFiles { + t.Fatalf("setup scan added %d records, want %d", + st.added, walkCancelFiles) + } + + before := recordPaths(dbRecords(t, db)) + base := baselineGoroutines(t) + + st, err := syncScan(newWalkClock(walkCancelAtDone), db, + []string{dir}, walkCancelWorkers, false) + + assertWalkGuardAborted(t, st, err) + assertRecordsIntact(t, db, before) + + if got := settledGoroutines(t, base); got > base { + t.Errorf("goroutines = %d after the cancelled scan, want %d back", + got, base) + } +} + +// assertWalkGuardAborted checks that the scan stopped at the post-walk +// guard: with a census that is neither empty (the walk really ran) +// nor complete (it really was cut short), and with the guard's own +// bare cancellation as the error. A wrapped error means the partial +// census was carried past the guard into the hash or update phase, +// which is the failure this test exists to catch. +func assertWalkGuardAborted(t *testing.T, st scanStats, err error) { + t.Helper() + + if !errors.Is(err, context.Canceled) { + t.Fatalf("syncScan cancelled mid-walk = %v, want %v", + err, context.Canceled) + } + + if errors.Unwrap(err) != nil { + t.Errorf("syncScan reported %q, want the guard's bare "+ + "cancellation: a wrapped error means the truncated census "+ + "reached a later phase", err) + } + + if st.unchanged == 0 { + t.Fatalf("stats = %+v: the census is empty, so the walk never "+ + "ran and the guard was reached for the wrong reason", st) + } + + if st.unchanged >= walkCancelFiles { + t.Fatalf("stats = %+v: the census covers the whole tree, so the "+ + "walk was not cut short", st) + } + + // The workers drop every directory still queued once the scan is + // cancelled, so only the directories already in flight can add to + // the census after the fact. A census beyond that bound would mean + // the cancellation was not observed where it should have been. + limit := walkCancelAtDone + walkCancelInFlightDirs + if st.unchanged > limit { + t.Errorf("census covers %d files, want at most %d: the walk kept "+ + "taking directories off the queue after cancellation", + st.unchanged, limit) + } + + if st.removed != 0 { + t.Errorf("stats = %+v: the scan counted records for removal from "+ + "a partial census", st) + } +} + +// TestSyncScanCancelledBeforeLoadIndex covers the trivial end of the +// cancellation path: a scan handed a context that is already cancelled +// fails in the index load, before the walk pool is ever started. It +// says nothing about the post-walk guard — nothing downstream of +// loadIndex runs at all — only that the failure surfaces as a +// cancellation and that no record is touched on the way out. +func TestSyncScanCancelledBeforeLoadIndex(t *testing.T) { + t.Parallel() + + dir := buildSmokeTree(t) + db := openTestDB(t) + + syncTree(t, db, dir) + + before := recordPaths(dbRecords(t, db)) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + st, err := syncScan(ctx, db, []string{dir}, walkCancelWorkers, false) + if !errors.Is(err, context.Canceled) { + t.Fatalf("syncScan on a cancelled context = %v, want %v", + err, context.Canceled) + } + + if st != (scanStats{}) { + t.Errorf("stats = %+v, want none: the scan gave up in the index "+ + "load, before any phase ran", st) + } + + assertRecordsIntact(t, db, before) +} + +// drainClosed counts the values received from ch until it closes, +// failing the test if it does not close within poolUnwind. A pool that +// ignored its cancellation leaves its channel open with its goroutines +// parked, and this is what reports that as an assertion. +func drainClosed[T any](t *testing.T, ch <-chan T, what string) int { + t.Helper() + + counted := make(chan int, 1) + + go func() { + n := 0 + for range ch { + n++ + } + + counted <- n + }() + + select { + case n := <-counted: + return n + case <-time.After(poolUnwind): + t.Fatalf("%s stayed open after cancellation", what) + + return 0 + } +} + +// awaitReturn fails the test if done is not closed within poolUnwind. +func awaitReturn(t *testing.T, done <-chan struct{}, what string) { + t.Helper() + + select { + case <-done: + case <-time.After(poolUnwind): + t.Fatalf("%s did not return after cancellation", what) + } +} + +// cancelledContext returns a context that is already cancelled. +func cancelledContext(t *testing.T) context.Context { + t.Helper() + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + return ctx +} + +// TestSendEventAbandonsBlockedSend checks that a walk goroutine with an +// event to deliver and nobody to deliver it to leaves on cancellation +// instead of holding the pool open. The channel here is unbuffered and +// unread, so the send can never complete. +func TestSendEventAbandonsBlockedSend(t *testing.T) { + t.Parallel() + + done := make(chan struct{}) + events := make(chan walkEvent) + + go func() { + defer close(done) + + sendEvent(cancelledContext(t), events, walkEvent{}) + }() + + awaitReturn(t, done, "sendEvent") +} + +// TestWalkWorkersDropQueuedDirs checks that cancelled walk workers keep +// reading jobs and drop the directories rather than stopping their +// read: the range over jobs has to run out for the pool to tear down +// and close its event stream. +func TestWalkWorkersDropQueuedDirs(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + writeEmptyFiles(t, dir, walkCancelFilesPerDir) + + jobs, _, events := startWalkWorkers(cancelledContext(t), 2, false) + + for range 4 { + jobs <- dirJob{path: dir} + } + + close(jobs) + + if n := drainClosed(t, events, "the walk event stream"); n != 0 { + t.Errorf("cancelled walk workers emitted %d events, want none", n) + } +} + +// TestWalkWorkerAbandonsSubdirHandoff checks the other blocking send a +// walk worker makes: handing discovered subdirectories back to the +// dispatcher. Once the dispatcher has left, nothing drains that +// channel, and a worker parked on it would hold the pool open forever. +func TestWalkWorkerAbandonsSubdirHandoff(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + writeEmptyFiles(t, dir, 1) + + err := os.Mkdir(filepath.Join(dir, "sub"), 0o750) + if err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + jobs, subdirs, events := startWalkWorkers(ctx, 1, false) + + // Fill the hand-back channel to its capacity — one slot per worker + // — so the worker's own hand-back is certain to block. + subdirs <- nil + + jobs <- dirJob{path: dir} + + // The file event proves the worker has read the directory and has + // nothing left to do but the blocked hand-back. + ev := <-events + if ev.fail { + t.Fatalf("walk event = %+v, want the fixture file", ev) + } + + cancel() + close(jobs) + drainClosed(t, events, "the walk event stream") +} + +// TestDispatchDirsClosesJobsWhenCancelled checks that a dispatcher +// leaving on cancellation closes the job channel on its way out. The +// workers range over that channel; a dispatcher that returned without +// closing it would strand every one of them. +func TestDispatchDirsClosesJobsWhenCancelled(t *testing.T) { + t.Parallel() + + // Unbuffered and unread: with no worker pool behind it, the + // dispatcher can only leave through its cancellation case. + jobs := make(chan dirJob) + subdirs := make(chan []dirJob) + initial := []dirJob{{path: "/a"}, {path: "/b"}} + + dispatchDirs(cancelledContext(t), initial, jobs, subdirs) + + if n := drainClosed(t, jobs, "the walk job queue"); n > len(initial) { + t.Errorf("dispatcher queued %d jobs, want at most %d", + n, len(initial)) + } +} + +// TestFeedHashJobsClosesJobsWhenCancelled checks that the hash feeder +// abandons the runs it has not queued yet and still closes the job +// channel, which is what lets the workers' range terminate. +func TestFeedHashJobsClosesJobsWhenCancelled(t *testing.T) { + t.Parallel() + + done := make(chan struct{}) + // Unbuffered and unread until the feeder has returned, so the only + // way out of the feeder is its cancellation case. + jobs := make(chan []fileRec) + runs := [][]fileRec{{{path: "a"}}, {{path: "b"}}} + + go func() { + defer close(done) + + feedHashJobs(cancelledContext(t), runs, jobs) + }() + + awaitReturn(t, done, "feedHashJobs") + + if _, ok := <-jobs; ok { + t.Error("the hash job channel was left open after cancellation") + } +} + +// TestHashWorkerDropsQueuedRuns checks that a cancelled hash worker +// keeps reading jobs and drops the runs rather than reading files +// nobody wants the hashes of — while still letting the range run out +// so the pool tears down. The queued run names a file that does not +// exist, so a worker that hashed it anyway would produce a result. +func TestHashWorkerDropsQueuedRuns(t *testing.T) { + t.Parallel() + + done := make(chan struct{}) + jobs := make(chan []fileRec, 1) + results := make(chan hashResult, 1) + + run := []fileRec{{path: filepath.Join(t.TempDir(), "missing"), size: 1}} + + jobs <- run + + close(jobs) + + go func() { + defer close(done) + + hashWorker(cancelledContext(t), jobs, results) + }() + + awaitReturn(t, done, "hashWorker") + + select { + case r := <-results: + t.Errorf("cancelled hash worker produced %+v, want the run dropped", + r) + default: + } +} + +// TestHashPhaseCancelledReturnsContextError checks the result loop's +// own exit: with the pool cancelled, no result will ever arrive, and +// the loop must leave through the cancellation rather than wait for a +// receive that cannot happen. +func TestHashPhaseCancelledReturnsContextError(t *testing.T) { + t.Parallel() + + s := &scanState{ + db: openTestDB(t), + toHash: []fileRec{{path: "a", size: 1, dev: 1, ino: 1}}, + } + + err := s.hashPhase(cancelledContext(t), 2) + if !errors.Is(err, context.Canceled) { + t.Fatalf("hashPhase on a cancelled context = %v, want %v", + err, context.Canceled) + } +} diff --git a/scan.go b/scan.go index 0c978bc..a6d825e 100644 --- a/scan.go +++ b/scan.go @@ -188,9 +188,13 @@ func syncScan(ctx context.Context, db *sql.DB, roots []string, changed, unhashed := s.walkPhase(startWalk(ctx, roots, oneFS, workers)) // A cancelled walk stops early, so its size census covers only part - // of the roots. Every file it never reached would look vanished to - // the update phase, which would then delete a perfectly good record - // for it: abort instead of writing that. + // of the roots, and every file it never reached looks vanished to + // the update phase. Defence in depth rather than the only barrier: + // that phase would today fail on its first BeginTx with the same + // cancelled context before deleting anything. But it is the barrier + // that survives a later decision to let an interrupted scan commit + // what it has, and it turns a confusing failure deep in the update + // phase into a clean abort at the phase boundary. err = ctx.Err() if err != nil { return s.st, err diff --git a/scan_test.go b/scan_test.go index 2c0a2f2..cc1fbf1 100644 --- a/scan_test.go +++ b/scan_test.go @@ -6,7 +6,6 @@ import ( "crypto/sha256" "database/sql" "encoding/hex" - "errors" "fmt" "os" "path/filepath" @@ -827,10 +826,14 @@ const injectedWriteFailure = "injected write failure" // hashLeakFiles is the size of the fixture for the hash-phase failure // test. The batch commit inside the hash phase is what fails, so the // tree must hold more than updateBatchSize files for the failure to -// happen at all; the surplus over that is what is still queued when it -// does, and it exceeds the depth of both pool channels so that the -// workers have nowhere left to put their results. An abandoned pool -// therefore parks forever, which is exactly what this test detects. +// happen at all, and the surplus over that is what is still queued +// when it does. That surplus is 2*workQueueDepth, which jobs, results +// and the workers in flight between them absorb exactly, so the feeder +// itself drains and exits; what an abandoned pool leaves parked is +// every worker, each holding a result nobody will ever receive, plus +// the goroutine waiting on them. That is what this test detects, and +// its margin over detecting nothing at all is the worker count — +// worth knowing before changing hashLeakWorkers or workQueueDepth. const hashLeakFiles = updateBatchSize + 2*workQueueDepth // hashLeakWorkers is the worker count for that scan: a fixed, modest @@ -966,42 +969,6 @@ func TestScanHashWriteFailureUnwindsPool(t *testing.T) { } } -// TestSyncScanCancelledWalkKeepsRecords checks the other half of the -// cancellation path: a scan whose context is already cancelled stops -// with that error instead of treating its truncated walk as the truth -// and deleting the record of every file it never reached. -// -//nolint:paralleltest // counts goroutines: must not run beside others -func TestSyncScanCancelledWalkKeepsRecords(t *testing.T) { - dir := buildSmokeTree(t) - db := openTestDB(t) - - syncTree(t, db, dir) - - before := recordPaths(dbRecords(t, db)) - - ctx, cancel := context.WithCancel(t.Context()) - cancel() - - base := baselineGoroutines(t) - - _, err := syncScan(ctx, db, []string{dir}, 4, false) - if !errors.Is(err, context.Canceled) { - t.Fatalf("syncScan on a cancelled context = %v, want %v", - err, context.Canceled) - } - - if got := recordPaths(dbRecords(t, db)); !slices.Equal(got, before) { - t.Errorf("records = %q after a cancelled scan, want %q", - got, before) - } - - if got := settledGoroutines(t, base); got > base { - t.Errorf("goroutines = %d after the cancelled scan, want %d back", - got, base) - } -} - func TestHashRuns(t *testing.T) { t.Parallel()