Unwind the hash worker pool instead of abandoning it (closes #6)

hashPhase returned the moment recordRun failed and left the pool
running: the feeder parked forever on a full jobs channel and every
worker on a full results channel. Until #4 landed the process exited
before that mattered; now that runScan returns an error and unwinds,
the goroutines are a real leak.

The pool is now an owned hashPool. Its context is derived from the
scan's, every blocking send in the feeder and the workers selects on
ctx.Done(), the feeder closes jobs on every path out so the workers'
range always terminates, and hashPhase defers pool.stop(), which
cancels and then drains results until the last goroutine has exited.
Draining is the half that matters: a worker already parked on a send
cannot observe the cancellation until a receiver frees it.

ctx comes from cmd.Context() and is threaded through runScan,
syncScan, both worker pools and the database layer as the first
parameter throughout, so graceful interrupt handling has a path to
hook into rather than a pool to rewrite.

The walk pool never leaked, because walkPhase always drains its
events to close, but it has the same unbounded-send shape and gets
the same treatment, together with a ctx.Err() guard after the walk: a
cancelled walk leaves a partial size census, and the update phase
would read every file it never reached as vanished and delete its
record.

Tests drive the scan entry point against a database whose insert
trigger aborts, with a fixture large enough that the failure lands
partway through the hash phase with more runs queued than either pool
channel can hold, and assert that the scan fails instead of hanging
and that runtime.NumGoroutine polls back to its pre-scan baseline.

TODO.md

@@ -29,6 +29,29 @@
 
 # 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 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
 - 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

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,37 @@ 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. 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.
+	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 +211,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 +219,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 +384,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 +418,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 +430,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 +447,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 +458,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 +474,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 +491,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 +547,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 +610,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 +627,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 +642,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 +665,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 +694,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 +723,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 +735,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 +746,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 +765,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 +792,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 +839,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,14 +1,18 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"database/sql"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
+	"runtime"
 	"slices"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -130,7 +134,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 +363,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 +375,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 +819,189 @@ 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; 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.
+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)
+	}
+}
+
+// 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()
 

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
 	}