From 139924995796a417cf6c16a6fe671958e27fa105 Mon Sep 17 00:00:00 2001
From: sneak <sneak@sneak.berlin>
Date: Sun, 9 Aug 2026 03:00:01 +0000
Subject: [PATCH 1/2] 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      |  23 +++++
 db.go        |  47 ++++-----
 db_test.go   |  39 +++----
 main.go      |  23 +++--
 main_test.go |   2 +-
 report.go    |  11 +-
 scan.go      | 287 +++++++++++++++++++++++++++++++++++++--------------
 scan_test.go | 193 +++++++++++++++++++++++++++++++++-
 trees.go     |   5 +-
 9 files changed, 489 insertions(+), 141 deletions(-)

diff --git a/TODO.md b/TODO.md
index 1d91eeb..edd9f11 100644
--- a/TODO.md
+++ b/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
diff --git a/db.go b/db.go
index 1965e76..573d944 100644
--- a/db.go
+++ b/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)
diff --git a/db_test.go b/db_test.go
index 5a3de46..41ab6f2 100644
--- a/db_test.go
+++ b/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)
 	}
diff --git a/main.go b/main.go
index 873bf7d..e9012db 100644
--- a/main.go
+++ b/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}
 		}
diff --git a/main_test.go b/main_test.go
index 9dbff77..b307a82 100644
--- a/main_test.go
+++ b/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)
 	}
diff --git a/report.go b/report.go
index f326fb6..7f7d0f9 100644
--- a/report.go
+++ b/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
 	}
diff --git a/scan.go b/scan.go
index e33318b..0c978bc 100644
--- a/scan.go
+++ b/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
+		}
 	}
 }
 
diff --git a/scan_test.go b/scan_test.go
index 9e563cd..2c0a2f2 100644
--- a/scan_test.go
+++ b/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()
 
diff --git a/trees.go b/trees.go
index 7dba881..fd05c0a 100644
--- a/trees.go
+++ b/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
 	}

From 1a38570301884e641e893f6b64b7a92e6dbb217c Mon Sep 17 00:00:00 2001
From: clawbot <clawbot@eeqj.de>
Date: Sun, 9 Aug 2026 05:12:43 +0000
Subject: [PATCH 2/2] Cover the post-walk cancellation guard with a test that
 reaches it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

TestSyncScanCancelledWalkKeepsRecords handed syncScan a context that
was already cancelled. loadIndex is the first thing syncScan does, and
its QueryContext fails on that context, so the scan returned before
startWalk was ever called: no walk ran, no pool started, no write path
was reachable, and all three of the test's assertions held for the
wrong reason. The post-walk ctx.Err() guard, which is the highest-stakes
line in the change, had no coverage at all — a panic in its body, or
deleting it outright, left the suite green.

Replace it with TestSyncScanCancelledMidWalkKeepsRecords, which cancels
during the walk and so reaches the guard holding a genuinely partial
census and a still-populated record index. The cancellation is driven
by the scan's own progress rather than by a timer: walkClock is a
context that cancels itself once its Done method has been consulted a
set number of times, and since every blocking channel operation in the
walk selects on Done — one consultation per event, a couple per
directory, against the index load's fixed three — a threshold set to a
quarter of the fixture's file count lands the cancellation deep inside
the walk on every run. The census settles at around 380 of 2000 files,
leaving some 1600 records that a complete-looking census would have
handed to the update phase as deletions.

The already-cancelled case is kept, renamed to what it actually tests
and with its goroutine assertion dropped, since nothing that could leak
is ever started.

Direct tests cover the remaining cancellation branches of both pools:
sendEvent abandoning a blocked send, walk workers dropping queued
directories, a walk worker abandoning its subdirectory hand-off,
dispatchDirs closing jobs on its way out, feedHashJobs doing the same,
hashWorker dropping queued runs, and hashPhase leaving its result loop.
Each is deterministic — the channels involved are unbuffered, unread or
pre-filled, so the cancellation case is the only one that can be ready.

Also correct two overstated claims. The hashLeakFiles comment described
a mechanism that does not occur: the surplus is absorbed exactly by the
two pool channels plus the workers in flight, so the feeder drains and
exits, and what an abandoned pool leaves parked is the workers and the
goroutine waiting on them. And the guard is defence in depth, not the
sole barrier against data loss: the update phase's BeginTx fails on the
same cancelled context before deleting anything today. The guard is
what keeps that true once an interrupted scan is allowed to commit what
it has.
---
 TODO.md        |  21 ++-
 cancel_test.go | 489 +++++++++++++++++++++++++++++++++++++++++++++++++
 scan.go        |  10 +-
 scan_test.go   |  49 +----
 4 files changed, 520 insertions(+), 49 deletions(-)
 create mode 100644 cancel_test.go

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