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

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)
+	}
+}

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

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()