TODO.md

@@ -29,6 +29,40 @@
 
 # Completed Steps
 
+- unwind the hash worker pool on the error path (2026-08-09, branch
+  `hash-pool-cleanup`, closes #6): `hashPhase` used to return the
+  moment `recordRun` failed and abandon the pool — the feeder parked
+  forever on a full `jobs` channel and every worker on a full
+  `results` channel. That only stopped being invisible when #4 landed
+  and `runScan` began unwinding instead of calling `os.Exit`. The
+  pool is now an owned, context-aware `hashPool`: every blocking send
+  in the feeder and the workers selects on `ctx.Done()`, `jobs` is
+  closed on every path out, and `hashPhase` defers `pool.stop()`,
+  which cancels and then drains `results` until the last goroutine
+  has exited — draining is what frees a worker already parked on a
+  send. `ctx` is threaded from `cmd.Context()` through `runScan`,
+  `syncScan`, both worker pools and the whole database layer (it is
+  the first parameter everywhere), so #5 can hand this path a signal
+  and needs to add nothing else. The walk pool never leaked, because
+  `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 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)
+	}
+}

db.go

@@ -96,7 +96,7 @@ func openDB(path string) (*sql.DB, error) {
 
 // openScanDatabase opens the database for the scan subcommand, creating
 // the file, its parent directory, and the schema as needed.
-func openScanDatabase(path string) (*sql.DB, error) {
+func openScanDatabase(ctx context.Context, path string) (*sql.DB, error) {
 	err := os.MkdirAll(filepath.Dir(path), dbDirPerm)
 	if err != nil {
 		return nil, fmt.Errorf("create database directory: %w", err)
@@ -107,7 +107,7 @@ func openScanDatabase(path string) (*sql.DB, error) {
 		return nil, err
 	}
 
-	err = initSchema(db)
+	err = initSchema(ctx, db)
 	if err != nil {
 		_ = db.Close()
 
@@ -120,7 +120,9 @@ func openScanDatabase(path string) (*sql.DB, error) {
 // openReportDatabase opens an existing database for the report and
 // trees subcommands. A missing database file is an error directing the
 // user to run scan first; the schema version must match exactly.
-func openReportDatabase(path string) (*sql.DB, error) {
+func openReportDatabase(ctx context.Context,
+	path string,
+) (*sql.DB, error) {
 	_, err := os.Stat(path)
 	if errors.Is(err, fs.ErrNotExist) {
 		return nil, fmt.Errorf("%s: %w", path, errNoDatabase)
@@ -135,7 +137,7 @@ func openReportDatabase(path string) (*sql.DB, error) {
 		return nil, err
 	}
 
-	v, err := userVersion(db)
+	v, err := userVersion(ctx, db)
 	if err != nil {
 		_ = db.Close()
 
@@ -154,15 +156,15 @@ func openReportDatabase(path string) (*sql.DB, error) {
 
 // initSchema creates the schema on a fresh database and verifies the
 // schema version on an existing one.
-func initSchema(db *sql.DB) error {
-	v, err := userVersion(db)
+func initSchema(ctx context.Context, db *sql.DB) error {
+	v, err := userVersion(ctx, db)
 	if err != nil {
 		return err
 	}
 
 	switch v {
 	case 0:
-		return createSchema(db)
+		return createSchema(ctx, db)
 	case schemaVersion:
 		return nil
 	default:
@@ -173,9 +175,7 @@ func initSchema(db *sql.DB) error {
 
 // createSchema applies the schema to a fresh database and stamps the
 // schema version.
-func createSchema(db *sql.DB) error {
-	ctx := context.Background()
-
+func createSchema(ctx context.Context, db *sql.DB) error {
 	_, err := db.ExecContext(ctx, createTableSQL)
 	if err != nil {
 		return fmt.Errorf("create schema: %w", err)
@@ -191,11 +191,10 @@ func createSchema(db *sql.DB) error {
 }
 
 // userVersion reads the database's PRAGMA user_version.
-func userVersion(db *sql.DB) (int, error) {
+func userVersion(ctx context.Context, db *sql.DB) (int, error) {
 	var v int
 
-	err := db.QueryRowContext(context.Background(),
-		"PRAGMA user_version").Scan(&v)
+	err := db.QueryRowContext(ctx, "PRAGMA user_version").Scan(&v)
 	if err != nil {
 		return 0, fmt.Errorf("read schema version: %w", err)
 	}
@@ -204,8 +203,8 @@ func userVersion(db *sql.DB) (int, error) {
 }
 
 // loadFileRows reads every record from the files table.
-func loadFileRows(db *sql.DB) ([]scanRec, error) {
-	rows, err := db.QueryContext(context.Background(),
+func loadFileRows(ctx context.Context, db *sql.DB) ([]scanRec, error) {
+	rows, err := db.QueryContext(ctx,
 		"SELECT path, size, mtime, head, tail FROM files")
 	if err != nil {
 		return nil, fmt.Errorf("read records: %w", err)
@@ -242,10 +241,10 @@ func loadFileRows(db *sql.DB) ([]scanRec, error) {
 // it carries hashes to fn. Scan change detection needs no hash
 // values, and skipping the hash columns keeps the scan's in-memory
 // index small on multi-million-file databases.
-func loadFileMeta(db *sql.DB,
+func loadFileMeta(ctx context.Context, db *sql.DB,
 	fn func(path string, size, mtime int64, hashed bool),
 ) error {
-	rows, err := db.QueryContext(context.Background(),
+	rows, err := db.QueryContext(ctx,
 		"SELECT path, size, mtime, head <> '' FROM files")
 	if err != nil {
 		return fmt.Errorf("read records: %w", err)
@@ -286,18 +285,18 @@ const updateBatchSize = 10000
 // applyChanges writes one scan's database changes — upserts for new and
 // changed files, deletes for vanished ones — in batched transactions.
 // Progress is rendered on prog (one increment per change).
-func applyChanges(db *sql.DB, upserts []scanRec, deletes []string,
-	prog *progress,
+func applyChanges(ctx context.Context, db *sql.DB, upserts []scanRec,
+	deletes []string, prog *progress,
 ) error {
 	for batch := range slices.Chunk(upserts, updateBatchSize) {
-		err := applyBatch(db, batch, nil, prog)
+		err := applyBatch(ctx, db, batch, nil, prog)
 		if err != nil {
 			return err
 		}
 	}
 
 	for batch := range slices.Chunk(deletes, updateBatchSize) {
-		err := applyBatch(db, nil, batch, prog)
+		err := applyBatch(ctx, db, nil, batch, prog)
 		if err != nil {
 			return err
 		}
@@ -308,11 +307,9 @@ func applyChanges(db *sql.DB, upserts []scanRec, deletes []string,
 
 // applyBatch commits one batch of upserts and deletes in a single
 // transaction.
-func applyBatch(db *sql.DB, upserts []scanRec, deletes []string,
-	prog *progress,
+func applyBatch(ctx context.Context, db *sql.DB, upserts []scanRec,
+	deletes []string, prog *progress,
 ) error {
-	ctx := context.Background()
-
 	tx, err := db.BeginTx(ctx, nil)
 	if err != nil {
 		return fmt.Errorf("begin transaction: %w", err)

db_test.go

@@ -22,7 +22,7 @@ func testDBPath(t *testing.T) string {
 func openTestDB(t *testing.T) *sql.DB {
 	t.Helper()
 
-	db, err := openScanDatabase(testDBPath(t))
+	db, err := openScanDatabase(t.Context(), testDBPath(t))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -52,12 +52,12 @@ func TestOpenScanDatabaseCreates(t *testing.T) {
 	// The parent directory does not exist yet; scan must create it.
 	path := filepath.Join(t.TempDir(), "nested", "dir", "db.sqlite")
 
-	db, err := openScanDatabase(path)
+	db, err := openScanDatabase(t.Context(), path)
 	if err != nil {
 		t.Fatalf("openScanDatabase: %v", err)
 	}
 
-	v, err := userVersion(db)
+	v, err := userVersion(t.Context(), db)
 	if err != nil || v != schemaVersion {
 		t.Fatalf("userVersion = %d, %v; want %d, nil", v, err, schemaVersion)
 	}
@@ -65,14 +65,14 @@ func TestOpenScanDatabaseCreates(t *testing.T) {
 	_ = db.Close()
 
 	// Reopening an existing database must succeed and find the schema.
-	db, err = openScanDatabase(path)
+	db, err = openScanDatabase(t.Context(), path)
 	if err != nil {
 		t.Fatalf("reopen: %v", err)
 	}
 
 	defer func() { _ = db.Close() }()
 
-	recs, err := loadFileRows(db)
+	recs, err := loadFileRows(t.Context(), db)
 	if err != nil || len(recs) != 0 {
 		t.Fatalf("loadFileRows = %v, %v; want empty, nil", recs, err)
 	}
@@ -81,7 +81,7 @@ func TestOpenScanDatabaseCreates(t *testing.T) {
 func TestOpenReportDatabaseMissing(t *testing.T) {
 	t.Parallel()
 
-	_, err := openReportDatabase(testDBPath(t))
+	_, err := openReportDatabase(t.Context(), testDBPath(t))
 	if !errors.Is(err, errNoDatabase) {
 		t.Fatalf("err = %v, want errNoDatabase", err)
 	}
@@ -92,7 +92,7 @@ func TestOpenReportDatabaseVersionMismatch(t *testing.T) {
 
 	path := testDBPath(t)
 
-	db, err := openScanDatabase(path)
+	db, err := openScanDatabase(t.Context(), path)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -104,7 +104,7 @@ func TestOpenReportDatabaseVersionMismatch(t *testing.T) {
 
 	_ = db.Close()
 
-	_, err = openReportDatabase(path)
+	_, err = openReportDatabase(t.Context(), path)
 	if !errors.Is(err, errSchemaVersion) {
 		t.Fatalf("err = %v, want errSchemaVersion", err)
 	}
@@ -115,14 +115,14 @@ func TestOpenReportDatabaseOK(t *testing.T) {
 
 	path := testDBPath(t)
 
-	db, err := openScanDatabase(path)
+	db, err := openScanDatabase(t.Context(), path)
 	if err != nil {
 		t.Fatal(err)
 	}
 
 	_ = db.Close()
 
-	db, err = openReportDatabase(path)
+	db, err = openReportDatabase(t.Context(), path)
 	if err != nil {
 		t.Fatalf("openReportDatabase: %v", err)
 	}
@@ -142,12 +142,13 @@ func TestApplyChangesRoundTrip(t *testing.T) {
 		{size: 1, mtime: 10, head: "h1", tail: "t1", path: "/a/x"},
 	}
 
-	err := applyChanges(db, recs, nil, newProgress("update", 2))
+	err := applyChanges(t.Context(), db, recs, nil,
+		newProgress("update", 2))
 	if err != nil {
 		t.Fatalf("applyChanges: %v", err)
 	}
 
-	got, err := loadFileRows(db)
+	got, err := loadFileRows(t.Context(), db)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -164,13 +165,13 @@ func TestApplyChangesRoundTrip(t *testing.T) {
 	// removes exactly its path.
 	upd := scanRec{size: 3, mtime: 30, head: "h3", tail: "t3", path: "/a/x"}
 
-	err = applyChanges(db, []scanRec{upd},
+	err = applyChanges(t.Context(), db, []scanRec{upd},
 		[]string{"/a/tab\tnew\nline"}, newProgress("update", 2))
 	if err != nil {
 		t.Fatalf("applyChanges: %v", err)
 	}
 
-	got, err = loadFileRows(db)
+	got, err = loadFileRows(t.Context(), db)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -197,12 +198,13 @@ func TestApplyChangesBatching(t *testing.T) {
 		})
 	}
 
-	err := applyChanges(db, recs, nil, newProgress("update", int64(n)))
+	err := applyChanges(t.Context(), db, recs, nil,
+		newProgress("update", int64(n)))
 	if err != nil {
 		t.Fatalf("applyChanges: %v", err)
 	}
 
-	got, err := loadFileRows(db)
+	got, err := loadFileRows(t.Context(), db)
 	if err != nil || len(got) != n {
 		t.Fatalf("loadFileRows = %d rows, %v; want %d", len(got), err, n)
 	}
@@ -212,12 +214,13 @@ func TestApplyChangesBatching(t *testing.T) {
 		deletes = append(deletes, r.path)
 	}
 
-	err = applyChanges(db, nil, deletes, newProgress("update", int64(n)))
+	err = applyChanges(t.Context(), db, nil, deletes,
+		newProgress("update", int64(n)))
 	if err != nil {
 		t.Fatalf("applyChanges deletes: %v", err)
 	}
 
-	got, err = loadFileRows(db)
+	got, err = loadFileRows(t.Context(), db)
 	if err != nil || len(got) != 0 {
 		t.Fatalf("loadFileRows = %d rows, %v; want 0", len(got), err)
 	}

main.go

@@ -16,6 +16,7 @@
 package main
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -121,8 +122,8 @@ func newRootCommand(stderr io.Writer) *cobra.Command {
 		Use:   cmdScan + " [--workers N] [-x] PATH...",
 		Short: "Walk trees and synchronize the scan database",
 		Args:  cobra.MinimumNArgs(1),
-		RunE: runE(func(args []string) error {
-			return runScan(args, scanWorkers, scanOneFS)
+		RunE: runE(func(ctx context.Context, args []string) error {
+			return runScan(ctx, args, scanWorkers, scanOneFS)
 		}),
 	}
 	scanCmd.Flags().IntVar(&scanWorkers, "workers", runtime.NumCPU(),
@@ -134,8 +135,8 @@ func newRootCommand(stderr io.Writer) *cobra.Command {
 		Use:   cmdReport,
 		Short: "Read the scan database and print the file-level duplicates report",
 		Args:  cobra.NoArgs,
-		RunE: runE(func(_ []string) error {
-			return runReport()
+		RunE: runE(func(ctx context.Context, _ []string) error {
+			return runReport(ctx)
 		}),
 	}
 
@@ -143,8 +144,8 @@ func newRootCommand(stderr io.Writer) *cobra.Command {
 		Use:   cmdTrees,
 		Short: "Read the scan database and print the duplicate-tree report",
 		Args:  cobra.NoArgs,
-		RunE: runE(func(_ []string) error {
-			return runTrees()
+		RunE: runE(func(ctx context.Context, _ []string) error {
+			return runTrees(ctx)
 		}),
 	}
 
@@ -157,13 +158,17 @@ func newRootCommand(stderr io.Writer) *cobra.Command {
 // prints the error and the command's usage text for every error RunE
 // returns, but a subcommand that ran and failed has no usage problem
 // to report: both are silenced here, and the error is marked fatal so
-// that run reports it on stderr and exits 1 rather than 2.
-func runE(fn func(args []string) error) func(*cobra.Command, []string) error {
+// that run reports it on stderr and exits 1 rather than 2. The command's
+// context is handed to the implementation: cancelling it unwinds the
+// scan's worker pools.
+func runE(
+	fn func(ctx context.Context, args []string) error,
+) func(*cobra.Command, []string) error {
 	return func(cmd *cobra.Command, args []string) error {
 		cmd.SilenceUsage = true
 		cmd.SilenceErrors = true
 
-		err := fn(args)
+		err := fn(cmd.Context(), args)
 		if err != nil {
 			return fatalError{err: err}
 		}

main_test.go

@@ -124,7 +124,7 @@ func TestOpenDatabaseKeepsWALWhileOpen(t *testing.T) {
 	// database was closed and its WAL checkpointed.
 	path := testDBPath(t)
 
-	db, err := openScanDatabase(path)
+	db, err := openScanDatabase(t.Context(), path)
 	if err != nil {
 		t.Fatal(err)
 	}

report.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"context"
 	"fmt"
 	"os"
 	"slices"
@@ -32,17 +33,17 @@ type scanRec struct {
 // exiting, so that the deferred close — which checkpoints the SQLite
 // WAL — always runs; the database is closed before the caller formats
 // its output, so it stays closed even if that output fails.
-func loadRecords() ([]scanRec, error) {
+func loadRecords(ctx context.Context) ([]scanRec, error) {
 	dbPath := databasePath()
 
-	db, err := openReportDatabase(dbPath)
+	db, err := openReportDatabase(ctx, dbPath)
 	if err != nil {
 		return nil, err
 	}
 
 	defer func() { _ = db.Close() }()
 
-	recs, err := loadFileRows(db)
+	recs, err := loadFileRows(ctx, db)
 	if err != nil {
 		return nil, fmt.Errorf("database %s: %w", dbPath, err)
 	}
@@ -62,8 +63,8 @@ type dupeGroup struct {
 // from the database and prints the file-level duplicates report as TSV
 // on stdout. It never touches the scanned filesystem; its only I/O is
 // the database, stdout, and stderr.
-func runReport() error {
-	recs, err := loadRecords()
+func runReport(ctx context.Context) error {
+	recs, err := loadRecords(ctx)
 	if err != nil {
 		return err
 	}

scan.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"cmp"
+	"context"
 	"crypto/sha256"
 	"database/sql"
 	"encoding/hex"
@@ -51,7 +52,11 @@ type fileMeta struct {
 // duplicate. Flag parsing and the at-least-one-operand check are done
 // by cobra. Errors are returned rather than exiting, so that the
 // deferred close — which checkpoints the SQLite WAL — always runs.
-func runScan(roots []string, workers int, oneFS bool) error {
+// Cancelling ctx unwinds the worker pools and aborts the scan with the
+// context's error.
+func runScan(ctx context.Context, roots []string, workers int,
+	oneFS bool,
+) error {
 	if workers < 1 {
 		workers = 1
 	}
@@ -63,14 +68,14 @@ func runScan(roots []string, workers int, oneFS bool) error {
 
 	dbPath := databasePath()
 
-	db, err := openScanDatabase(dbPath)
+	db, err := openScanDatabase(ctx, dbPath)
 	if err != nil {
 		return err
 	}
 
 	defer func() { _ = db.Close() }()
 
-	st, err := syncScan(db, roots, workers, oneFS)
+	st, err := syncScan(ctx, db, roots, workers, oneFS)
 	if err != nil {
 		return fmt.Errorf("update database %s: %w", dbPath, err)
 	}
@@ -168,28 +173,41 @@ type scanState struct {
 // and update (record the size-unique files without reading them, and
 // delete the records the scan no longer verifies). Records outside
 // the roots are never touched.
-func syncScan(db *sql.DB, roots []string, workers int,
-	oneFS bool,
+func syncScan(ctx context.Context, db *sql.DB, roots []string,
+	workers int, oneFS bool,
 ) (scanStats, error) {
 	roots = pruneRoots(roots)
 
 	s := &scanState{db: db}
 
-	err := s.loadIndex(roots)
+	err := s.loadIndex(ctx, roots)
 	if err != nil {
 		return s.st, err
 	}
 
-	changed, unhashed := s.walkPhase(startWalk(roots, oneFS, workers))
+	changed, unhashed := s.walkPhase(startWalk(ctx, roots, oneFS, workers))
+
+	// A cancelled walk stops early, so its size census covers only part
+	// 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
+	}
 
 	s.partition(changed, unhashed)
 
-	err = s.hashPhase(workers)
+	err = s.hashPhase(ctx, workers)
 	if err != nil {
 		return s.st, err
 	}
 
-	return s.st, s.updatePhase()
+	return s.st, s.updatePhase(ctx)
 }
 
 // loadIndex indexes the database records under the scan roots for
@@ -197,7 +215,7 @@ func syncScan(db *sql.DB, roots []string, workers int,
 // them: out-of-scope records join the size census so a scanned file
 // can be recognized as a possible duplicate of a tree scanned
 // separately into the same database.
-func (s *scanState) loadIndex(roots []string) error {
+func (s *scanState) loadIndex(ctx context.Context, roots []string) error {
 	// Indexing tens of millions of records takes real time; without a
 	// display the scan looks hung before the walk begins.
 	prog := newProgress("load", -1)
@@ -205,7 +223,7 @@ func (s *scanState) loadIndex(roots []string) error {
 
 	s.existing = make(map[string]fileMeta)
 
-	return loadFileMeta(s.db,
+	return loadFileMeta(ctx, s.db,
 		func(path string, size, mtime int64, hashed bool) {
 			prog.increment()
 
@@ -370,28 +388,29 @@ func sameInode(a, b fileRec) bool {
 // reads, so the bar shows a real ETA. A run that fails to hash is
 // warned about and skipped; stale records for its paths, if any, are
 // deleted by the update phase.
-func (s *scanState) hashPhase(workers int) error {
+//
+// Returning early — a failed database write, or a cancelled scan — must
+// not strand the pool: the feeder would park forever on a full jobs
+// channel and every worker on a full results channel. The deferred stop
+// is what prevents that.
+func (s *scanState) hashPhase(ctx context.Context, workers int) error {
 	runs := hashRuns(s.toHash)
 	s.toHash = nil
 
-	jobs := make(chan []fileRec, workQueueDepth)
-	results := make(chan hashResult, workQueueDepth)
-
-	startHashWorkers(jobs, results, workers)
-
-	go func() {
-		for _, run := range runs {
-			jobs <- run
-		}
-
-		close(jobs)
-	}()
+	pool := startHashPool(ctx, runs, workers)
+	defer pool.stop()
 
 	prog := newProgress("hash", int64(len(runs)))
 	defer prog.finish()
 
 	for range runs {
-		r := <-results
+		var r hashResult
+
+		select {
+		case r = <-pool.results:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
 
 		prog.increment()
 
@@ -403,7 +422,7 @@ func (s *scanState) hashPhase(workers int) error {
 			continue
 		}
 
-		err := s.recordRun(r)
+		err := s.recordRun(ctx, r)
 		if err != nil {
 			return err
 		}
@@ -415,7 +434,7 @@ func (s *scanState) hashPhase(workers int) error {
 // recordRun folds one hash result into the running batch: every path
 // in the run (one file, or several hard links to it) gets a record
 // with the shared hashes.
-func (s *scanState) recordRun(r hashResult) error {
+func (s *scanState) recordRun(ctx context.Context, r hashResult) error {
 	for _, rec := range r.run {
 		s.resolve(rec.path)
 
@@ -432,7 +451,7 @@ func (s *scanState) recordRun(r hashResult) error {
 		return nil
 	}
 
-	err := applyBatch(s.db, s.batch, nil, nil)
+	err := applyBatch(ctx, s.db, s.batch, nil, nil)
 	s.batch = s.batch[:0]
 
 	return err
@@ -443,7 +462,7 @@ func (s *scanState) recordRun(r hashResult) error {
 // size-unique new or changed file, and deletions for every record the
 // scan did not verify (vanished files, plus paths that failed to stat
 // or hash).
-func (s *scanState) updatePhase() error {
+func (s *scanState) updatePhase(ctx context.Context) error {
 	deletes := make([]string, 0, len(s.existing))
 	for path := range s.existing {
 		deletes = append(deletes, path)
@@ -459,7 +478,7 @@ func (s *scanState) updatePhase() error {
 
 	defer prog.finish()
 
-	err := applyChanges(s.db, s.batch, nil, prog)
+	err := applyChanges(ctx, s.db, s.batch, nil, prog)
 	if err != nil {
 		return err
 	}
@@ -476,13 +495,13 @@ func (s *scanState) updatePhase() error {
 			})
 		}
 
-		err = applyBatch(s.db, recs, nil, prog)
+		err = applyBatch(ctx, s.db, recs, nil, prog)
 		if err != nil {
 			return err
 		}
 	}
 
-	return applyChanges(s.db, nil, deletes, prog)
+	return applyChanges(ctx, s.db, nil, deletes, prog)
 }
 
 // underAnyRoot reports whether path is any of the roots or lies under
@@ -532,34 +551,53 @@ type walkEvent struct {
 // startWalk seeds every root into the shared walk worker pool and
 // returns the event stream: one record per regular file, one warning
 // event per per-path error. The channel is closed when the walk
-// completes.
-func startWalk(roots []string, oneFS bool, workers int) <-chan walkEvent {
-	jobs, subdirs, events := startWalkWorkers(workers, oneFS)
+// completes, and also when ctx is cancelled — every goroutine in the
+// pool abandons its blocking send in that case, so the consumer sees a
+// truncated but properly terminated stream instead of a stalled one.
+func startWalk(ctx context.Context, roots []string, oneFS bool,
+	workers int,
+) <-chan walkEvent {
+	jobs, subdirs, events := startWalkWorkers(ctx, workers, oneFS)
 
 	go func() {
 		initial := make([]dirJob, 0, len(roots))
 		for _, root := range roots {
-			initial = append(initial, seedRoot(root, events)...)
+			initial = append(initial, seedRoot(ctx, root, events)...)
 		}
 
-		dispatchDirs(initial, jobs, subdirs)
+		dispatchDirs(ctx, initial, jobs, subdirs)
 	}()
 
 	return events
 }
 
+// sendEvent delivers one walk event, abandoning the send when the scan
+// is cancelled. Every walk goroutine reaches the consumer through this
+// one channel, so this is where a cancelled walk unwinds rather than
+// parking on a buffer nobody is draining.
+func sendEvent(ctx context.Context, events chan<- walkEvent,
+	ev walkEvent,
+) {
+	select {
+	case events <- ev:
+	case <-ctx.Done():
+	}
+}
+
 // seedRoot turns one PATH operand into the walk's starting state: a
 // regular-file operand is statted and emitted directly, a directory
 // operand becomes an initial job, and a symlink or other non-regular
 // operand yields nothing (symlinks are never followed, including as
 // operands).
-func seedRoot(root string, events chan<- walkEvent) []dirJob {
+func seedRoot(ctx context.Context, root string,
+	events chan<- walkEvent,
+) []dirJob {
 	fi, err := os.Lstat(root)
 	if err != nil {
-		events <- walkEvent{
+		sendEvent(ctx, events, walkEvent{
 			warn: fmt.Sprintf("walk %s: %v", root, err),
 			fail: true,
-		}
+		})
 
 		return nil
 	}
@@ -576,13 +614,13 @@ func seedRoot(root string, events chan<- walkEvent) []dirJob {
 	case fi.Mode().IsRegular():
 		dev, ino := inodeOfInfo(fi)
 
-		events <- walkEvent{rec: fileRec{
+		sendEvent(ctx, events, walkEvent{rec: fileRec{
 			path:  root,
 			size:  fi.Size(),
 			mtime: fi.ModTime().Unix(),
 			dev:   dev,
 			ino:   ino,
-		}}
+		}})
 
 		return nil
 	default:
@@ -593,8 +631,10 @@ func seedRoot(root string, events chan<- walkEvent) []dirJob {
 // startWalkWorkers starts the walk worker pool. Each worker processes
 // one directory at a time, emitting an event per regular file and
 // handing discovered subdirectories back to the dispatcher; events is
-// closed once every worker has finished.
-func startWalkWorkers(workers int,
+// closed once every worker has finished. A cancelled scan makes the
+// workers drop the directories still queued rather than stop reading
+// jobs, so the range always runs out and the pool always tears down.
+func startWalkWorkers(ctx context.Context, workers int,
 	oneFS bool,
 ) (chan dirJob, chan []dirJob, chan walkEvent) {
 	jobs := make(chan dirJob, workQueueDepth)
@@ -606,7 +646,14 @@ func startWalkWorkers(workers int,
 	for range workers {
 		wg.Go(func() {
 			for job := range jobs {
-				subdirs <- walkOneDir(job, oneFS, events)
+				if ctx.Err() != nil {
+					continue
+				}
+
+				select {
+				case subdirs <- walkOneDir(ctx, job, oneFS, events):
+				case <-ctx.Done():
+				}
 			}
 		})
 	}
@@ -622,11 +669,15 @@ func startWalkWorkers(workers int,
 // dispatchDirs feeds directory jobs to the walk workers, queueing
 // newly discovered subdirectories (newest first, which keeps the
 // frontier small) until every directory has been processed, then
-// closes jobs.
-func dispatchDirs(initial []dirJob, jobs chan<- dirJob,
-	subdirs <-chan []dirJob,
+// closes jobs. jobs is closed on every path out, cancellation
+// included: the workers range over it, and a dispatcher that returned
+// without closing would strand all of them.
+func dispatchDirs(ctx context.Context, initial []dirJob,
+	jobs chan<- dirJob, subdirs <-chan []dirJob,
 ) {
 	go func() {
+		defer close(jobs)
+
 		queue := slices.Clone(initial)
 		pending := len(queue)
 
@@ -647,23 +698,25 @@ func dispatchDirs(initial []dirJob, jobs chan<- dirJob,
 			case subs := <-subdirs:
 				pending += len(subs) - 1
 				queue = append(queue, subs...)
+			case <-ctx.Done():
+				return
 			}
 		}
-
-		close(jobs)
 	}()
 }
 
 // walkOneDir reads one directory, emitting an event per regular-file
 // entry and a warning event per unreadable one, and returns the
 // subdirectories to descend into.
-func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
+func walkOneDir(ctx context.Context, job dirJob, oneFS bool,
+	events chan<- walkEvent,
+) []dirJob {
 	entries, err := os.ReadDir(job.path)
 	if err != nil {
-		events <- walkEvent{
+		sendEvent(ctx, events, walkEvent{
 			warn: fmt.Sprintf("walk %s: %v", job.path, err),
 			fail: true,
-		}
+		})
 
 		return nil
 	}
@@ -674,7 +727,7 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
 		p := filepath.Join(job.path, e.Name())
 
 		if e.IsDir() {
-			if sub, ok := subdirJob(p, e, job, oneFS, events); ok {
+			if sub, ok := subdirJob(ctx, p, e, job, oneFS, events); ok {
 				subs = append(subs, sub)
 			}
 
@@ -686,7 +739,7 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
 			continue
 		}
 
-		emitFile(p, e, events)
+		emitFile(ctx, p, e, events)
 	}
 
 	return subs
@@ -697,13 +750,15 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
 // metadata is still hot; a path that fails to stat (or stops being a
 // regular file) between the directory read and the lstat is warned
 // about and skipped.
-func emitFile(p string, e fs.DirEntry, events chan<- walkEvent) {
+func emitFile(ctx context.Context, p string, e fs.DirEntry,
+	events chan<- walkEvent,
+) {
 	info, err := e.Info()
 	if err != nil {
-		events <- walkEvent{
+		sendEvent(ctx, events, walkEvent{
 			warn: fmt.Sprintf("stat %s: %v", p, err),
 			fail: true,
-		}
+		})
 
 		return
 	}
@@ -714,21 +769,21 @@ func emitFile(p string, e fs.DirEntry, events chan<- walkEvent) {
 
 	dev, ino := inodeOfInfo(info)
 
-	events <- walkEvent{rec: fileRec{
+	sendEvent(ctx, events, walkEvent{rec: fileRec{
 		path:  p,
 		size:  info.Size(),
 		mtime: info.ModTime().Unix(),
 		dev:   dev,
 		ino:   ino,
-	}}
+	}})
 }
 
 // subdirJob applies the descent rules to directory p: never enter
 // .zfs (ZFS snapshot pseudo-dirs would list every file once per
 // snapshot), and with -x never enter a directory on a different
 // filesystem than its operand.
-func subdirJob(p string, e fs.DirEntry, parent dirJob, oneFS bool,
-	events chan<- walkEvent,
+func subdirJob(ctx context.Context, p string, e fs.DirEntry,
+	parent dirJob, oneFS bool, events chan<- walkEvent,
 ) (dirJob, bool) {
 	if e.Name() == ".zfs" {
 		return dirJob{}, false
@@ -741,10 +796,10 @@ func subdirJob(p string, e fs.DirEntry, parent dirJob, oneFS bool,
 
 	info, err := e.Info()
 	if err != nil {
-		events <- walkEvent{
+		sendEvent(ctx, events, walkEvent{
 			warn: fmt.Sprintf("walk %s: %v", p, err),
 			fail: true,
-		}
+		})
 
 		return dirJob{}, false
 	}
@@ -788,22 +843,102 @@ type hashResult struct {
 	err  error
 }
 
-// startHashWorkers starts the hash worker pool: workers read inode
-// runs, hash each run's first path (all paths in a run are hard links
-// to the same inode), write one result per run, and exit when jobs is
-// closed.
-func startHashWorkers(jobs <-chan []fileRec, results chan<- hashResult,
+// hashPool owns every goroutine of the hash worker pool: the feeder
+// that queues the inode runs and the workers that read them. Both block
+// on channel sends, so both are cancellable — the pool's context is
+// derived from the scan's, and stop cancels it and waits the goroutines
+// out. The consumer must call stop on every path out of the phase, not
+// just the happy one.
+type hashPool struct {
+	results <-chan hashResult
+	cancel  context.CancelFunc
+	done    <-chan struct{}
+}
+
+// startHashPool starts the feeder and the workers over runs. Workers
+// hash each run's first path (all paths in a run are hard links to the
+// same inode) and write one result per run.
+func startHashPool(ctx context.Context, runs [][]fileRec,
 	workers int,
-) {
+) *hashPool {
+	ctx, cancel := context.WithCancel(ctx)
+
+	jobs := make(chan []fileRec, workQueueDepth)
+	results := make(chan hashResult, workQueueDepth)
+
+	var wg sync.WaitGroup
+
+	wg.Go(func() { feedHashJobs(ctx, runs, jobs) })
+
 	for range workers {
-		go func() {
-			for run := range jobs {
-				head, tail, err := hashHeadTail(run[0].path, run[0].size)
-				results <- hashResult{
-					run: run, head: head, tail: tail, err: err,
-				}
-			}
-		}()
+		wg.Go(func() { hashWorker(ctx, jobs, results) })
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	return &hashPool{results: results, cancel: cancel, done: done}
+}
+
+// stop cancels the pool and blocks until every one of its goroutines
+// has exited, draining results while it waits: a worker already parked
+// on a send observes the cancellation only once a receiver frees it.
+// Calling stop more than once is safe.
+func (p *hashPool) stop() {
+	p.cancel()
+
+	for {
+		select {
+		case <-p.results:
+		case <-p.done:
+			return
+		}
+	}
+}
+
+// feedHashJobs queues every run for the workers, closing jobs on the
+// way out — including when the scan is cancelled mid-queue, so that the
+// workers' range over jobs always terminates.
+func feedHashJobs(ctx context.Context, runs [][]fileRec,
+	jobs chan<- []fileRec,
+) {
+	defer close(jobs)
+
+	for _, run := range runs {
+		select {
+		case jobs <- run:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// hashWorker hashes one inode run at a time until jobs is closed or the
+// scan is cancelled. A cancelled worker drops the runs still queued
+// instead of stopping its reads of jobs: the range must run out for the
+// pool to tear down, and reading a file nobody wants the hash of only
+// delays that.
+func hashWorker(ctx context.Context, jobs <-chan []fileRec,
+	results chan<- hashResult,
+) {
+	for run := range jobs {
+		if ctx.Err() != nil {
+			continue
+		}
+
+		head, tail, err := hashHeadTail(run[0].path, run[0].size)
+
+		select {
+		case results <- hashResult{
+			run: run, head: head, tail: tail, err: err,
+		}:
+		case <-ctx.Done():
+			return
+		}
 	}
 }
 

scan_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"database/sql"
@@ -8,7 +9,9 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
 	"slices"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -130,7 +133,7 @@ func collectWalk(t *testing.T, roots []string, oneFS bool,
 		errs int
 	)
 
-	for ev := range startWalk(roots, oneFS, workers) {
+	for ev := range startWalk(t.Context(), roots, oneFS, workers) {
 		if ev.fail {
 			errs++
 
@@ -359,7 +362,7 @@ const smokeTreeFiles = 15
 func syncTree(t *testing.T, db *sql.DB, roots ...string) scanStats {
 	t.Helper()
 
-	st, err := syncScan(db, roots, 4, false)
+	st, err := syncScan(t.Context(), db, roots, 4, false)
 	if err != nil {
 		t.Fatalf("syncScan: %v", err)
 	}
@@ -371,7 +374,7 @@ func syncTree(t *testing.T, db *sql.DB, roots ...string) scanStats {
 func dbRecords(t *testing.T, db *sql.DB) []scanRec {
 	t.Helper()
 
-	recs, err := loadFileRows(db)
+	recs, err := loadFileRows(t.Context(), db)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -815,6 +818,157 @@ func TestScanHardlinkRunFailsTogether(t *testing.T) {
 	}
 }
 
+// injectedWriteFailure is the message the injected database trigger
+// aborts with, so the test can recognize its own failure in the error
+// the scan reports.
+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, 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
+// number keeps the leak deterministic on any machine.
+const hashLeakWorkers = 4
+
+// goroutineSettle bounds how long a goroutine count is given to come
+// back down to its target. Only a failing run ever waits this long.
+const goroutineSettle = 5 * time.Second
+
+// goroutinePoll is the interval between goroutine-count samples.
+const goroutinePoll = 10 * time.Millisecond
+
+// writeEmptyFiles creates n empty files directly in dir. Zero-length
+// files are never opened by the hasher — their hashes are constant —
+// so a fixture this size costs directory entries and no read I/O,
+// while still queueing n runs through the hash pool.
+func writeEmptyFiles(t *testing.T, dir string, n int) {
+	t.Helper()
+
+	for i := range n {
+		err := os.WriteFile(
+			filepath.Join(dir, strconv.Itoa(i)), nil, 0o600)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+// injectWriteFailure creates a scan database at path carrying the real
+// schema plus a trigger that aborts every insert. Reads are untouched,
+// so a scan loads its index and walks normally and then fails on the
+// first record it tries to commit — a genuine database write failure
+// partway through the hash phase.
+func injectWriteFailure(t *testing.T, path string) {
+	t.Helper()
+
+	db, err := openScanDatabase(t.Context(), path)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = db.ExecContext(t.Context(),
+		"CREATE TRIGGER refuse_insert BEFORE INSERT ON files "+
+			"BEGIN SELECT RAISE(ABORT, '"+injectedWriteFailure+"'); END")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = db.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+}
+
+// baselineGoroutines waits for the goroutine count to stop moving and
+// returns it. Handles closed by earlier tests take a moment to reap
+// their driver goroutines, so a single sample would make the baseline
+// itself flaky.
+func baselineGoroutines(t *testing.T) int {
+	t.Helper()
+
+	deadline := time.Now().Add(goroutineSettle)
+	last := runtime.NumGoroutine()
+
+	for time.Now().Before(deadline) {
+		time.Sleep(goroutinePoll)
+
+		n := runtime.NumGoroutine()
+		if n == last {
+			return n
+		}
+
+		last = n
+	}
+
+	return last
+}
+
+// settledGoroutines polls runtime.NumGoroutine until it is back at or
+// below want and returns the last count seen. Polling, rather than one
+// sample after a fixed sleep, is what keeps this from being a race
+// between the assertion and goroutines that are already exiting.
+func settledGoroutines(t *testing.T, want int) int {
+	t.Helper()
+
+	deadline := time.Now().Add(goroutineSettle)
+
+	for {
+		n := runtime.NumGoroutine()
+		if n <= want || time.Now().After(deadline) {
+			return n
+		}
+
+		time.Sleep(goroutinePoll)
+	}
+}
+
+// TestScanHashWriteFailureUnwindsPool drives the scan entry point
+// against a database that refuses every write. The hash phase gives up
+// partway through with thousands of runs still queued, which used to
+// leave the feeder parked on a full job channel and every worker parked
+// on a full result channel for the life of the process.
+func TestScanHashWriteFailureUnwindsPool(t *testing.T) {
+	path := testDBPath(t)
+	t.Setenv(databaseEnv, path)
+
+	dir := t.TempDir()
+
+	writeEmptyFiles(t, dir, hashLeakFiles)
+	injectWriteFailure(t, path)
+
+	base := baselineGoroutines(t)
+
+	var stderr bytes.Buffer
+
+	code := run([]string{
+		cmdScan, "--workers", strconv.Itoa(hashLeakWorkers), dir,
+	}, &stderr)
+	if code != exitFatal {
+		t.Fatalf("run(scan) = %d, want %d; stderr: %s",
+			code, exitFatal, stderr.String())
+	}
+
+	if !strings.Contains(stderr.String(), injectedWriteFailure) {
+		t.Errorf("stderr = %q, want the injected write failure",
+			stderr.String())
+	}
+
+	if got := settledGoroutines(t, base); got > base {
+		t.Errorf("goroutines = %d after the failed scan, want %d back",
+			got, base)
+	}
+}
+
 func TestHashRuns(t *testing.T) {
 	t.Parallel()
 

trees.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bufio"
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"os"
@@ -34,8 +35,8 @@ type treeNode struct {
 // maximal duplicate-tree groups as TSV on stdout. It never touches the
 // scanned filesystem; its only I/O is the database, stdout, and
 // stderr.
-func runTrees() error {
-	recs, err := loadRecords()
+func runTrees(ctx context.Context) error {
+	recs, err := loadRecords(ctx)
 	if err != nil {
 		return err
 	}