Merge branch 'scan-operand-progress': announce operands on stderr

With per-operand walk/hash/update cycles, a multi-operand invocation
(e.g. scan /srv/*) showed pass totals that looked like the whole
run's: an operator watching operand 3 of 14 hash 300k files concluded
the other 20M files were being skipped. Print the operand path and
its position before each cycle.

README.md

@@ -131,9 +131,11 @@ All three subcommands operate on a single SQLite database file:
     `scan` first.
 -   The database uses WAL journal mode and a busy timeout, so running a
     report while a cron `scan` is in progress is safe; the reports see
-    the last committed state. Each scan commits its changes in a single
-    transaction, so a report never observes a half-finished scan and a
-    scan that dies partway leaves the previous state intact.
+    the last committed state. Each `PATH` operand's changes are
+    committed in a single transaction when that operand completes, so
+    a report never observes a half-finished operand, and a scan that
+    dies partway keeps every operand completed so far (the interrupted
+    operand's changes are lost, not corrupted).
 -   Schema (`PRAGMA user_version` is the schema version, currently 1; a
     database with any other version is a fatal error):
 
@@ -160,9 +162,11 @@ or a regular file; an operand that does not exist is a fatal error
 (exit 1). Because database records persist between runs and are keyed
 by absolute path, each operand is resolved to an absolute, lexically
 cleaned path (symlinks are not resolved) before walking, so results do
-not depend on the working directory. Operands are walked in the order
-given; overlapping operands (one containing another) are harmless — a
-file reached via multiple operands produces one database record.
+not depend on the working directory. Operands are processed in the
+order given, each one walked and committed independently; overlapping
+operands (one containing another) are harmless — a file reached via
+multiple operands produces one database record (later operands see the
+records committed by earlier ones and reuse them unchanged).
 
 `scan` synchronizes the database with the filesystem state under the
 scanned operands:
@@ -184,20 +188,35 @@ scanned operands:
     disjoint trees can be scanned on different schedules into the same
     database.
 
-`scan` runs **four sequential passes**, in this order, so that every
-expensive pass has an exact total for meaningful progress and ETA:
+`scan` runs **three sequential passes per operand**, committing each
+operand before starting the next:
 
-1.  **walk** — recursively enumerate the tree under each `PATH` in
-    turn, collecting the list of regular-file paths. Total unknown
-    while running: show a live count, not a percentage.
-2.  **stat** — `lstat` every collected path, recording size and mtime.
-3.  **hash** — for each new or changed file (per the rules above), read
+1.  **walk** — enumerate the tree with the worker pool: each worker
+    reads one directory at a time, records size and mtime for every
+    regular-file entry (`lstat` while the directory is fresh in
+    cache), and hands discovered subdirectories back to the shared
+    queue. Sequential directory enumeration is metadata-latency-bound
+    and takes hours at tens of millions of files; per-directory
+    parallelism is what makes the walk tractable on large or busy
+    pools. Total unknown while running: show a live count, not a
+    percentage.
+2.  **hash** — for each new or changed file (per the rules above), read
     the first `min(1024, size)` bytes and the last `min(1024, size)`
     bytes (the two reads overlap when `size < 2048`; for `size == 0`
     hash the empty input) and compute the SHA-256 of each. Unchanged
-    files are not read and do not appear in this pass's total.
-4.  **update** — apply all insertions, updates, and deletions to the
-    database in a single transaction.
+    files are not read and do not appear in this pass's total, which
+    is therefore exact for meaningful progress and ETA.
+3.  **update** — apply the operand's insertions, updates, and
+    deletions to the database in a single transaction.
+
+Because every operand runs its own three passes with its own totals,
+`scan` announces each operand on stderr before starting it, so a
+multi-operand invocation (e.g. a shell glob) is not mistaken for a
+nearly-finished run:
+
+```
+scan: /srv/media (operand 3 of 14)
+```
 
 Rules for the walk:
 
@@ -219,9 +238,11 @@ Rules for the walk:
     records (accepted: the database mirrors what the latest scan could
     actually verify).
 
-Concurrency: the stat and hash passes use a worker pool (`--workers`,
-default `runtime.NumCPU()`). The main goroutine owns database writes
-and progress rendering; progress display must never block the workers.
+Concurrency: the walk and hash passes use a worker pool (`--workers`,
+default `runtime.NumCPU()`); the walk parallelizes across directories,
+so raising `--workers` can speed up metadata-bound walks on busy
+pools. The main goroutine owns database writes and progress rendering;
+progress display must never block the workers.
 
 `scan` writes nothing to stdout. The summary line on stderr reports the
 files seen this run broken down by disposition, plus skips:
@@ -336,8 +357,8 @@ all dupe rows) in human units.
 Use the progress-bar library for all scan-pass progress; rendering in the
 style of `pv` is the model. All progress goes to stderr.
 
-Each scan pass gets its own bar. Required elements for the stat, hash,
-and update passes (known totals):
+Each scan pass gets its own bar, repeated per operand. Required
+elements for the hash and update passes (known totals):
 
 -   elapsed time
 -   estimated time remaining

TODO.md

@@ -19,6 +19,19 @@
 
 # Completed Steps
 
+- announce each operand on stderr before its passes (2026-07-24,
+  branch `scan-operand-progress`): with per-operand walk/hash/update
+  cycles, a multi-operand run (e.g. `scan /srv/*`) showed pass totals
+  that looked like the whole run's — an operator watching operand 3 of
+  14 hash 300k files concluded 20M files were being skipped
+- parallel walk (2026-07-24, branch `parallel-walk`): the walk pass
+  was a single goroutine and took hours at ~20M files on a busy pool
+  (observed: 22M files in 4h on a ZFS server); it is now a
+  per-directory worker-pool traversal that records size/mtime during
+  the walk (folding away the separate stat pass, halving metadata
+  I/O), and each `PATH` operand commits in its own transaction so an
+  interrupted scan keeps completed operands
+
 - persistent scan database (2026-07-24, branch `persistent-database`):
   `scan` now maintains a SQLite database (`modernc.org/sqlite`, pure
   Go, cgo stays disabled) keyed by absolute path that survives between

scan.go

@@ -11,6 +11,7 @@ import (
 	"path/filepath"
 	"slices"
 	"strings"
+	"sync"
 	"syscall"
 )
 
@@ -21,8 +22,8 @@ const chunk = 1024
 // and hash worker pools.
 const workQueueDepth = 1024
 
-// errNotRegular reports a path that stopped being a regular file
-// between the walk and stat passes.
+// errNotRegular reports a path whose type changed between the
+// directory read and its lstat.
 var errNotRegular = errors.New("no longer a regular file")
 
 // fileRec carries one file between the stat and hash passes.
@@ -98,27 +99,48 @@ type scanStats struct {
 	skipped   int
 }
 
-// syncScan synchronizes the database with the filesystem under roots:
-// walk and stat everything, hash only new or changed files, and apply
-// the resulting record insertions, updates, and deletions in a single
-// transaction. Records outside the roots are never touched.
+// syncScan synchronizes the database with the filesystem under roots.
+// Operands are processed in order, each one walked, hashed, and
+// committed independently, so an interrupted scan keeps every operand
+// completed so far. Records outside the roots are never touched.
 func syncScan(db *sql.DB, roots []string, workers int,
 	oneFS bool,
 ) (scanStats, error) {
 	var st scanStats
 
-	existing, err := loadScopedRows(db, roots)
+	for i, root := range roots {
+		// Each operand runs its own walk/hash/update sequence, so
+		// announce it: without this, the per-operand pass totals look
+		// like the whole run's.
+		fmt.Fprintf(os.Stderr, "scan: %s (operand %d of %d)\n",
+			root, i+1, len(roots))
+
+		err := syncRoot(db, root, workers, oneFS, &st)
 		if err != nil {
 			return st, err
 		}
+	}
 
-	paths, walkErrs := walkPass(roots, oneFS)
-	recs, statErrs := statPass(uniquePaths(paths), workers)
-	toHash, unchanged := partitionChanged(recs, existing)
+	return st, nil
+}
+
+// syncRoot walks one PATH operand, hashes its new and changed files,
+// and commits the operand's record insertions, updates, and deletions
+// in a single transaction.
+func syncRoot(db *sql.DB, root string, workers int, oneFS bool,
+	st *scanStats,
+) error {
+	existing, err := loadScopedRows(db, root)
+	if err != nil {
+		return err
+	}
+
+	walked, walkErrs := walkPass(root, oneFS, workers)
+	toHash, unchanged := partitionChanged(walked, existing)
 	hashed, hashErrs := hashPass(toHash, workers)
 
-	st.unchanged = len(unchanged)
-	st.skipped = walkErrs + statErrs + hashErrs
+	st.unchanged += len(unchanged)
+	st.skipped += walkErrs + hashErrs
 
 	for _, r := range hashed {
 		if _, ok := existing[r.path]; ok {
@@ -129,15 +151,15 @@ func syncScan(db *sql.DB, roots []string, workers int,
 	}
 
 	deletes := collectDeletes(existing, unchanged, hashed)
-	st.removed = len(deletes)
+	st.removed += len(deletes)
 
-	return st, applyPass(db, hashed, deletes)
+	return applyPass(db, hashed, deletes)
 }
 
-// loadScopedRows loads the database records whose paths lie under any
-// of the scan roots, keyed by path. Records outside the roots belong
-// to other trees and are left untouched by this scan.
-func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
+// loadScopedRows loads the database records whose paths lie under
+// root, keyed by path. Records outside the scanned operands belong to
+// other trees and are left untouched.
+func loadScopedRows(db *sql.DB, root string) (map[string]scanRec, error) {
 	all, err := loadFileRows(db)
 	if err != nil {
 		return nil, err
@@ -146,7 +168,7 @@ func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
 	scoped := make(map[string]scanRec)
 
 	for _, r := range all {
-		if underAnyRoot(r.path, roots) {
+		if underRoot(r.path, root) {
 			scoped[r.path] = r
 		}
 	}
@@ -154,18 +176,6 @@ func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
 	return scoped, nil
 }
 
-// underAnyRoot reports whether path is any of the roots or lies under
-// one of them.
-func underAnyRoot(path string, roots []string) bool {
-	for _, root := range roots {
-		if underRoot(path, root) {
-			return true
-		}
-	}
-
-	return false
-}
-
 // underRoot reports whether path is root itself or lies under it. Both
 // must be absolute and lexically clean.
 func underRoot(path, root string) bool {
@@ -181,26 +191,6 @@ func underRoot(path, root string) bool {
 	return strings.HasPrefix(path, prefix)
 }
 
-// uniquePaths deduplicates the walked paths, preserving order.
-// Overlapping operands can reach the same file more than once, but the
-// database keys records by path, so each file is processed once.
-func uniquePaths(paths []string) []string {
-	seen := make(map[string]bool, len(paths))
-	out := make([]string, 0, len(paths))
-
-	for _, p := range paths {
-		if seen[p] {
-			continue
-		}
-
-		seen[p] = true
-
-		out = append(out, p)
-	}
-
-	return out
-}
-
 // partitionChanged splits the stat results into files that must be
 // hashed (new, or changed) and files whose existing records are reused
 // without reading them: a file is unchanged when its statted size
@@ -270,105 +260,247 @@ func applyPass(db *sql.DB, upserts []scanRec, deletes []string) error {
 	return err
 }
 
-// treeWalker carries the walk-pass state shared by all PATH operands.
-type treeWalker struct {
-	prog  *progress
-	oneFS bool
-	paths []string
-	errs  int
+// dirJob is one directory awaiting traversal by the walk workers. It
+// carries its operand's filesystem device so -x can stop at
+// filesystem boundaries.
+type dirJob struct {
+	path      string
+	rootDev   uint64
+	rootDevOK bool
 }
 
-// walkPass enumerates every regular file under each root operand in
-// order. It never follows symlinks, never descends into directories
-// named .zfs, and warns and continues on any per-path error. With
-// oneFS set it never descends into a directory on a different
-// filesystem than its root operand.
-func walkPass(roots []string, oneFS bool) ([]string, int) {
-	w := &treeWalker{prog: newProgress("walk", -1), oneFS: oneFS}
-
-	for _, root := range roots {
-		w.walkRoot(root)
+// walkEvent is one walk result delivered to the main goroutine: a
+// regular file's stat record, or a warning when fail is set.
+type walkEvent struct {
+	rec  fileRec
+	warn string
+	fail bool
 }
 
-	w.prog.finish()
+// walkPass enumerates every regular file under root with a
+// per-directory worker pool, recording size and mtime from lstat
+// while each directory is fresh in cache. It never follows symlinks,
+// never descends into directories named .zfs, and warns and continues
+// on any per-path error. With oneFS set it never descends into a
+// directory on a different filesystem than root.
+func walkPass(root string, oneFS bool, workers int) ([]fileRec, int) {
+	prog := newProgress("walk", -1)
 
-	return w.paths, w.errs
+	recs, initial, errs := seedRoot(root, prog)
+	jobs, subdirs, events := startWalkWorkers(workers, oneFS)
+
+	dispatchDirs(initial, jobs, subdirs)
+
+	for ev := range events {
+		if ev.fail {
+			errs++
+
+			prog.warnf("%s", ev.warn)
+
+			continue
 		}
 
-// walkRoot walks a single PATH operand, appending regular-file paths.
-func (w *treeWalker) walkRoot(root string) {
-	rootDev, rootDevOK := deviceOf(root)
+		recs = append(recs, ev.rec)
 
-	walkErr := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
+		prog.increment()
+	}
+
+	prog.finish()
+
+	return recs, errs
+}
+
+// seedRoot turns the PATH operand into the walk's starting state: a
+// regular-file operand becomes a record directly, a directory operand
+// becomes the initial job, and a symlink or other non-regular operand
+// yields nothing (symlinks are never followed, including as
+// operands).
+func seedRoot(root string, prog *progress) ([]fileRec, []dirJob, int) {
+	fi, err := os.Lstat(root)
 	if err != nil {
-			w.errs++
+		prog.warnf("walk %s: %v", root, err)
 
-			w.prog.warnf("walk %s: %v", p, err)
+		return nil, nil, 1
+	}
 
-			if d != nil && d.IsDir() {
-				return filepath.SkipDir
+	switch {
+	case fi.IsDir():
+		if filepath.Base(root) == ".zfs" {
+			return nil, nil, 0
+		}
+
+		dev, ok := deviceOfInfo(fi)
+
+		return nil, []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}, 0
+	case fi.Mode().IsRegular():
+		rec := fileRec{path: root, size: fi.Size(), mtime: fi.ModTime().Unix()}
+
+		prog.increment()
+
+		return []fileRec{rec}, nil, 0
+	default:
+		return nil, nil, 0
+	}
+}
+
+// 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,
+	oneFS bool,
+) (chan dirJob, chan []dirJob, chan walkEvent) {
+	jobs := make(chan dirJob, workQueueDepth)
+	subdirs := make(chan []dirJob, workers)
+	events := make(chan walkEvent, workQueueDepth)
+
+	var wg sync.WaitGroup
+
+	for range workers {
+		wg.Go(func() {
+			for job := range jobs {
+				subdirs <- walkOneDir(job, oneFS, events)
+			}
+		})
+	}
+
+	go func() {
+		wg.Wait()
+		close(events)
+	}()
+
+	return jobs, subdirs, events
+}
+
+// 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,
+) {
+	go func() {
+		queue := slices.Clone(initial)
+		pending := len(queue)
+
+		for pending > 0 {
+			var (
+				out  chan<- dirJob
+				next dirJob
+			)
+
+			if len(queue) > 0 {
+				out = jobs
+				next = queue[len(queue)-1]
+			}
+
+			select {
+			case out <- next:
+				queue = queue[:len(queue)-1]
+			case subs := <-subdirs:
+				pending += len(subs) - 1
+				queue = append(queue, subs...)
+			}
+		}
+
+		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 {
+	entries, err := os.ReadDir(job.path)
+	if err != nil {
+		events <- walkEvent{
+			warn: fmt.Sprintf("walk %s: %v", job.path, err),
+			fail: true,
 		}
 
 		return nil
 	}
 
-		if d.IsDir() {
-			return w.dirAction(p, d, rootDev, rootDevOK)
+	var subs []dirJob
+
+	for _, e := range entries {
+		p := filepath.Join(job.path, e.Name())
+
+		if e.IsDir() {
+			if sub, ok := subdirJob(p, e, job, oneFS, events); ok {
+				subs = append(subs, sub)
+			}
+
+			continue
 		}
 		// Regular files only: skip symlinks, sockets, FIFOs, and
 		// device nodes.
-		if !d.Type().IsRegular() {
-			return nil
+		if !e.Type().IsRegular() {
+			continue
 		}
 
-		w.paths = append(w.paths, p)
+		events <- fileEvent(p, e)
+	}
 
-		w.prog.increment()
+	return subs
+}
 
-		return nil
-	})
-	if walkErr != nil {
-		fatalf("walk %s: %v", root, walkErr)
+// fileEvent lstats one regular-file directory entry into its walk
+// event.
+func fileEvent(p string, e fs.DirEntry) walkEvent {
+	fi, err := e.Info()
+
+	switch {
+	case err != nil:
+		return walkEvent{
+			warn: fmt.Sprintf("walk %s: %v", p, err),
+			fail: true,
+		}
+	case !fi.Mode().IsRegular():
+		return walkEvent{
+			warn: fmt.Sprintf("walk %s: %v", p, errNotRegular),
+			fail: true,
+		}
+	default:
+		return walkEvent{rec: fileRec{
+			path:  p,
+			size:  fi.Size(),
+			mtime: fi.ModTime().Unix(),
+		}}
 	}
 }
 
-// dirAction decides whether the walk descends into directory p.
-func (w *treeWalker) dirAction(p string, d fs.DirEntry, rootDev uint64, rootDevOK bool) error {
-	// ZFS snapshot pseudo-dirs would list every file once per
-	// snapshot; never descend.
-	if d.Name() == ".zfs" {
-		return filepath.SkipDir
+// 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,
+) (dirJob, bool) {
+	if e.Name() == ".zfs" {
+		return dirJob{}, false
 	}
 
-	if !w.oneFS || !rootDevOK {
-		return nil
+	job := dirJob{path: p, rootDev: parent.rootDev, rootDevOK: parent.rootDevOK}
+	if !oneFS || !parent.rootDevOK {
+		return job, true
 	}
 
-	info, err := d.Info()
+	info, err := e.Info()
 	if err != nil {
-		w.errs++
-
-		w.prog.warnf("walk %s: %v", p, err)
-
-		return filepath.SkipDir
+		events <- walkEvent{
+			warn: fmt.Sprintf("walk %s: %v", p, err),
+			fail: true,
 		}
 
-	if dev, ok := deviceOfInfo(info); ok && dev != rootDev {
-		return filepath.SkipDir
+		return dirJob{}, false
 	}
 
-	return nil
+	if dev, ok := deviceOfInfo(info); ok && dev != parent.rootDev {
+		return dirJob{}, false
 	}
 
-// deviceOf returns the filesystem device ID of path without following
-// symlinks.
-func deviceOf(path string) (uint64, bool) {
-	fi, err := os.Lstat(path)
-	if err != nil {
-		return 0, false
-	}
-
-	return deviceOfInfo(fi)
+	return job, true
 }
 
 // deviceOfInfo extracts the filesystem device ID from a FileInfo, when
@@ -382,72 +514,6 @@ func deviceOfInfo(fi fs.FileInfo) (uint64, bool) {
 	return statDev(st), true
 }
 
-// statPass lstats every collected path in a worker pool, recording size
-// and mtime. Paths that fail to stat (or are no longer regular files)
-// are warned about and dropped.
-func statPass(paths []string, workers int) ([]fileRec, int) {
-	type result struct {
-		rec fileRec
-		err error
-	}
-
-	jobs := make(chan string, workQueueDepth)
-	results := make(chan result, workQueueDepth)
-
-	for range workers {
-		go func() {
-			for p := range jobs {
-				fi, err := os.Lstat(p)
-				switch {
-				case err != nil:
-					results <- result{rec: fileRec{path: p}, err: err}
-				case !fi.Mode().IsRegular():
-					results <- result{
-						rec: fileRec{path: p},
-						err: errNotRegular,
-					}
-				default:
-					results <- result{rec: fileRec{
-						path:  p,
-						size:  fi.Size(),
-						mtime: fi.ModTime().Unix(),
-					}}
-				}
-			}
-		}()
-	}
-
-	go func() {
-		for _, p := range paths {
-			jobs <- p
-		}
-
-		close(jobs)
-	}()
-
-	prog := newProgress("stat", int64(len(paths)))
-
-	var errs int
-
-	recs := make([]fileRec, 0, len(paths))
-	for range paths {
-		r := <-results
-		if r.err != nil {
-			errs++
-
-			prog.warnf("stat %s: %v", r.rec.path, r.err)
-		} else {
-			recs = append(recs, r.rec)
-		}
-
-		prog.increment()
-	}
-
-	prog.finish()
-
-	return recs, errs
-}
-
 // hashResult carries one file's head/tail hashes (or the error that
 // prevented hashing it) from the hash workers to the main goroutine.
 type hashResult struct {

scan_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"database/sql"
 	"encoding/hex"
+	"fmt"
 	"os"
 	"path/filepath"
 	"slices"
@@ -109,6 +110,33 @@ func TestHashHeadTailErrors(t *testing.T) {
 	}
 }
 
+// walkedPaths returns the sorted paths of walked records.
+func walkedPaths(recs []fileRec) []string {
+	paths := make([]string, 0, len(recs))
+	for _, r := range recs {
+		paths = append(paths, r.path)
+	}
+
+	slices.Sort(paths)
+
+	return paths
+}
+
+// walkedByPath finds the walked record with the given path.
+func walkedByPath(t *testing.T, recs []fileRec, path string) fileRec {
+	t.Helper()
+
+	for _, r := range recs {
+		if r.path == path {
+			return r
+		}
+	}
+
+	t.Fatalf("no walked record for %q", path)
+
+	return fileRec{}
+}
+
 func TestWalkPass(t *testing.T) {
 	t.Parallel()
 
@@ -130,37 +158,46 @@ func TestWalkPass(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	paths, errs := walkPass([]string{dir}, false)
+	recs, errs := walkPass(dir, false, 4)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
 
-	slices.Sort(paths)
+	if got := walkedPaths(recs); !slices.Equal(got, want) {
+		t.Fatalf("paths = %q, want %q", got, want)
+	}
 
-	if !slices.Equal(paths, want) {
-		t.Fatalf("paths = %q, want %q", paths, want)
+	// The walk records lstat sizes and mtimes.
+	if r := walkedByPath(t, recs, want[0]); r.size != 1 || r.mtime <= 0 {
+		t.Fatalf("rec = %+v, want size 1 and a positive mtime", r)
 	}
 }
 
-func TestWalkPassMultipleRoots(t *testing.T) {
+func TestWalkPassDeepAndWide(t *testing.T) {
 	t.Parallel()
 
-	rootA := t.TempDir()
-	rootB := t.TempDir()
-	want := []string{
-		writeFile(t, rootA, "a1", []byte("1")),
-		writeFile(t, rootA, "sub/a2", []byte("2")),
-		writeFile(t, rootB, "b1", []byte("3")),
+	// Exercise the dispatcher with more directories than workers and
+	// with nesting deeper than the worker count.
+	dir := t.TempDir()
+	deep := "deep" + strings.Repeat("/d", 30)
+
+	want := make([]string, 0, 41)
+	want = append(want, writeFile(t, dir, deep+"/f", []byte("x")))
+
+	for i := range 40 {
+		want = append(want, writeFile(t, dir,
+			fmt.Sprintf("wide/%02d/f", i), []byte("y")))
 	}
 
-	paths, errs := walkPass([]string{rootA, rootB}, false)
+	slices.Sort(want)
+
+	recs, errs := walkPass(dir, false, 8)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
 
-	// Operands are walked in the order given.
-	if !slices.Equal(paths, want) {
-		t.Fatalf("paths = %q, want %q", paths, want)
+	if got := walkedPaths(recs); !slices.Equal(got, want) {
+		t.Fatalf("walked %d paths, want %d", len(got), len(want))
 	}
 }
 
@@ -178,15 +215,15 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
 	}
 
 	// A regular-file operand is emitted as itself.
-	paths, errs := walkPass([]string{f}, false)
-	if errs != 0 || !slices.Equal(paths, []string{f}) {
-		t.Fatalf("file operand: paths = %q, errs = %d", paths, errs)
+	recs, errs := walkPass(f, false, 2)
+	if errs != 0 || len(recs) != 1 || recs[0].path != f || recs[0].size != 4 {
+		t.Fatalf("file operand: recs = %+v, errs = %d", recs, errs)
 	}
 
 	// A symlink operand is not followed and yields nothing.
-	paths, errs = walkPass([]string{link}, false)
-	if errs != 0 || len(paths) != 0 {
-		t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs)
+	recs, errs = walkPass(link, false, 2)
+	if errs != 0 || len(recs) != 0 {
+		t.Fatalf("symlink operand: recs = %+v, errs = %d", recs, errs)
 	}
 }
 
@@ -200,58 +237,37 @@ func TestWalkPassOneFilesystemSameFS(t *testing.T) {
 		writeFile(t, dir, "sub/deep/b", []byte("b")),
 	}
 
-	paths, errs := walkPass([]string{dir}, true)
+	recs, errs := walkPass(dir, true, 4)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
 
-	slices.Sort(paths)
-
-	if !slices.Equal(paths, want) {
-		t.Fatalf("paths = %q, want %q", paths, want)
+	if got := walkedPaths(recs); !slices.Equal(got, want) {
+		t.Fatalf("paths = %q, want %q", got, want)
 	}
 }
 
-func TestDeviceOf(t *testing.T) {
+func TestDeviceOfInfo(t *testing.T) {
 	t.Parallel()
 
 	dir := t.TempDir()
 
-	dev1, ok1 := deviceOf(dir)
+	fi1, err := os.Lstat(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
 
-	dev2, ok2 := deviceOf(dir)
+	fi2, err := os.Lstat(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	dev1, ok1 := deviceOfInfo(fi1)
+
+	dev2, ok2 := deviceOfInfo(fi2)
 	if !ok1 || !ok2 || dev1 != dev2 {
-		t.Fatalf("deviceOf unstable: %d/%v vs %d/%v", dev1, ok1, dev2, ok2)
-	}
-
-	if _, ok := deviceOf(filepath.Join(dir, "missing")); ok {
-		t.Fatal("deviceOf reported ok for a missing path")
-	}
-}
-
-func TestStatPass(t *testing.T) {
-	t.Parallel()
-
-	dir := t.TempDir()
-	a := writeFile(t, dir, "a", pattern(1, 10))
-	b := writeFile(t, dir, "b", pattern(2, 20))
-	missing := filepath.Join(dir, "vanished")
-
-	recs, errs := statPass([]string{a, b, missing}, 2)
-	if errs != 1 {
-		t.Fatalf("errs = %d, want 1 for the vanished file", errs)
-	}
-
-	slices.SortFunc(recs, func(x, y fileRec) int {
-		return strings.Compare(x.path, y.path)
-	})
-
-	if len(recs) != 2 || recs[0].size != 10 || recs[1].size != 20 {
-		t.Fatalf("recs = %+v, want sizes 10 and 20", recs)
-	}
-
-	if recs[0].mtime <= 0 || recs[1].mtime <= 0 {
-		t.Fatalf("recs = %+v, want positive mtimes", recs)
+		t.Fatalf("deviceOfInfo unstable: %d/%v vs %d/%v",
+			dev1, ok1, dev2, ok2)
 	}
 }
 
@@ -607,10 +623,17 @@ func TestSyncScanOverlappingRoots(t *testing.T) {
 
 	writeFile(t, dir, "sub/f", pattern(1, 10))
 
-	// A file reachable via two overlapping operands yields one record.
+	// A file reachable via two overlapping operands yields one
+	// record: the second operand sees the rows the first operand
+	// committed and reuses them unchanged, which also proves each
+	// operand commits before the next starts.
 	st := syncTree(t, db, dir, filepath.Join(dir, "sub"))
-	if st.added != 1 {
-		t.Fatalf("stats = %+v, want 1 added", st)
+	if st != (scanStats{added: 1, unchanged: 1}) {
+		t.Fatalf("stats = %+v, want 1 added 1 unchanged", st)
+	}
+
+	if got := recordPaths(dbRecords(t, db)); len(got) != 1 {
+		t.Fatalf("records = %q, want exactly one", got)
 	}
 }
 
@@ -659,14 +682,3 @@ func TestUnderRoot(t *testing.T) {
 		}
 	}
 }
-
-func TestUniquePaths(t *testing.T) {
-	t.Parallel()
-
-	got := uniquePaths([]string{"/a", "/b", "/a", "/c", "/b"})
-
-	want := []string{"/a", "/b", "/c"}
-	if !slices.Equal(got, want) {
-		t.Fatalf("uniquePaths = %q, want %q", got, want)
-	}
-}