Merge branch 'parallel-phases': sequential phases, parallelism within each

README.md

@@ -188,25 +188,27 @@ scanned operands:
     disjoint trees can be scanned on different schedules into the same
     database.
 
-`scan` runs **three sequential passes per operand**, committing each
-operand before starting the next:
+`scan` runs **four sequential passes per operand**, committing each
+operand before starting the next. Parallelism lives strictly inside
+each pass; the passes themselves never overlap:
 
 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
+    reads one directory at a time, collecting regular-file paths and
+    handing 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
+2.  **stat** — `lstat` every collected path with the worker pool
+    (per-file parallelism), recording size and mtime.
+3.  **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, which
     is therefore exact for meaningful progress and ETA.
-3.  **update** — apply the operand's insertions, updates, and
+4.  **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,
@@ -238,11 +240,12 @@ Rules for the walk:
     records (accepted: the database mirrors what the latest scan could
     actually verify).
 
-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.
+Concurrency: the walk, stat, and hash passes each use a worker pool
+(`--workers`, default `runtime.NumCPU()`); the walk parallelizes
+across directories, stat and hash across files, so raising
+`--workers` can speed up metadata-bound passes 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:
@@ -358,7 +361,7 @@ 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, repeated per operand. Required
-elements for the hash and update passes (known totals):
+elements for the stat, hash, and update passes (known totals):
 
 -   elapsed time
 -   estimated time remaining

TODO.md

@@ -19,6 +19,12 @@
 
 # Completed Steps
 
+- split the stat pass back out of the walk (2026-07-24, branch
+  `parallel-phases`): phases are strictly sequential again — walk,
+  stat, hash, update per operand — with parallelism only inside each
+  phase; the walk enumerates paths with per-directory workers and the
+  stat pass lstats them with per-file workers, restoring the exact
+  total/ETA stat bar
 - 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

scan.go

@@ -135,12 +135,13 @@ func syncRoot(db *sql.DB, root string, workers int, oneFS bool,
 		return err
 	}
 
-	walked, walkErrs := walkPass(root, oneFS, workers)
-	toHash, unchanged := partitionChanged(walked, existing)
+	paths, walkErrs := walkPass(root, oneFS, workers)
+	recs, statErrs := statPass(paths, workers)
+	toHash, unchanged := partitionChanged(recs, existing)
 	hashed, hashErrs := hashPass(toHash, workers)
 
 	st.unchanged += len(unchanged)
-	st.skipped += walkErrs + hashErrs
+	st.skipped += walkErrs + statErrs + hashErrs
 
 	for _, r := range hashed {
 		if _, ok := existing[r.path]; ok {
@@ -270,23 +271,22 @@ type dirJob struct {
 }
 
 // walkEvent is one walk result delivered to the main goroutine: a
-// regular file's stat record, or a warning when fail is set.
+// regular-file path, or a warning when fail is set.
 type walkEvent struct {
-	rec  fileRec
+	path string
 	warn string
 	fail bool
 }
 
 // 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
+// per-directory worker pool. 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) {
+func walkPass(root string, oneFS bool, workers int) ([]string, int) {
 	prog := newProgress("walk", -1)
 
-	recs, initial, errs := seedRoot(root, prog)
+	paths, initial, errs := seedRoot(root, prog)
 	jobs, subdirs, events := startWalkWorkers(workers, oneFS)
 
 	dispatchDirs(initial, jobs, subdirs)
@@ -300,22 +300,22 @@ func walkPass(root string, oneFS bool, workers int) ([]fileRec, int) {
 			continue
 		}
 
-		recs = append(recs, ev.rec)
+		paths = append(paths, ev.path)
 
 		prog.increment()
 	}
 
 	prog.finish()
 
-	return recs, errs
+	return paths, errs
 }
 
 // seedRoot turns the PATH operand into the walk's starting state: a
-// regular-file operand becomes a record directly, a directory operand
+// regular-file operand becomes a path 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) {
+func seedRoot(root string, prog *progress) ([]string, []dirJob, int) {
 	fi, err := os.Lstat(root)
 	if err != nil {
 		prog.warnf("walk %s: %v", root, err)
@@ -333,11 +333,9 @@ func seedRoot(root string, prog *progress) ([]fileRec, []dirJob, int) {
 
 		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
+		return []string{root}, nil, 0
 	default:
 		return nil, nil, 0
 	}
@@ -439,37 +437,12 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
 			continue
 		}
 
-		events <- fileEvent(p, e)
+		events <- walkEvent{path: p}
 	}
 
 	return subs
 }
 
-// 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(),
-		}}
-	}
-}
-
 // 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
@@ -514,6 +487,72 @@ 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

@@ -110,33 +110,6 @@ 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()
 
@@ -158,18 +131,15 @@ func TestWalkPass(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	recs, errs := walkPass(dir, false, 4)
+	paths, errs := walkPass(dir, false, 4)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
 
-	if got := walkedPaths(recs); !slices.Equal(got, want) {
-		t.Fatalf("paths = %q, want %q", got, want)
-	}
+	slices.Sort(paths)
 
-	// 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)
+	if !slices.Equal(paths, want) {
+		t.Fatalf("paths = %q, want %q", paths, want)
 	}
 }
 
@@ -191,13 +161,15 @@ func TestWalkPassDeepAndWide(t *testing.T) {
 
 	slices.Sort(want)
 
-	recs, errs := walkPass(dir, false, 8)
+	paths, errs := walkPass(dir, false, 8)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
 
-	if got := walkedPaths(recs); !slices.Equal(got, want) {
-		t.Fatalf("walked %d paths, want %d", len(got), len(want))
+	slices.Sort(paths)
+
+	if !slices.Equal(paths, want) {
+		t.Fatalf("walked %d paths, want %d", len(paths), len(want))
 	}
 }
 
@@ -215,15 +187,15 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
 	}
 
 	// A regular-file operand is emitted as itself.
-	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)
+	paths, errs := walkPass(f, false, 2)
+	if errs != 0 || !slices.Equal(paths, []string{f}) {
+		t.Fatalf("file operand: paths = %q, errs = %d", paths, errs)
 	}
 
 	// A symlink operand is not followed and yields nothing.
-	recs, errs = walkPass(link, false, 2)
-	if errs != 0 || len(recs) != 0 {
-		t.Fatalf("symlink operand: recs = %+v, errs = %d", recs, errs)
+	paths, errs = walkPass(link, false, 2)
+	if errs != 0 || len(paths) != 0 {
+		t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs)
 	}
 }
 
@@ -237,13 +209,41 @@ func TestWalkPassOneFilesystemSameFS(t *testing.T) {
 		writeFile(t, dir, "sub/deep/b", []byte("b")),
 	}
 
-	recs, errs := walkPass(dir, true, 4)
+	paths, errs := walkPass(dir, true, 4)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
 
-	if got := walkedPaths(recs); !slices.Equal(got, want) {
-		t.Fatalf("paths = %q, want %q", got, want)
+	slices.Sort(paths)
+
+	if !slices.Equal(paths, want) {
+		t.Fatalf("paths = %q, want %q", paths, want)
+	}
+}
+
+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)
 	}
 }