Merge branch 'scan-wide-phases': scan-wide phases, concurrent operands, batched updates

README.md

@@ -130,12 +130,13 @@ All three subcommands operate on a single SQLite database file:
     database file is a fatal error (exit 1) telling the user to run
     `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
+    report while a cron `scan` is in progress is safe. The filesystem
-    the last committed state. Each `PATH` operand's changes are
+    is authoritative; the database is an eventually-consistent
-    committed in a single transaction when that operand completes, so
+    reflection of it. The update pass applies changes in batched
-    a report never observes a half-finished operand, and a scan that
+    transactions (keeping the WAL small and letting concurrent reports
-    dies partway keeps every operand completed so far (the interrupted
+    observe progress), so a report may see a scan's changes partially
-    operand's changes are lost, not corrupted).
+    applied, and a scan that dies partway leaves a valid database that
+    the next scan converges toward the filesystem.
 -   Schema (`PRAGMA user_version` is the schema version, currently 1; a
     database with any other version is a fatal error):
 
@@ -162,11 +163,12 @@ 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 processed in the
+not depend on the working directory. All operands belong to a single
-order given, each one walked and committed independently; overlapping
+scan and are enumerated concurrently: every operand seeds the shared
-operands (one containing another) are harmless — a file reached via
+walk worker pool, and every pass runs once over the whole scan, so
-multiple operands produces one database record (later operands see the
+pass totals and ETAs are scan-wide. Overlapping operands (one
-records committed by earlier ones and reuse them unchanged).
+containing another) are harmless — a file reached via multiple
+operands is deduplicated by path and produces one database record.
 
 `scan` synchronizes the database with the filesystem state under the
 scanned operands:
@@ -188,18 +190,19 @@ scanned operands:
     disjoint trees can be scanned on different schedules into the same
     database.
 
-`scan` runs **four sequential passes per operand**, committing each
+`scan` runs **four sequential passes over the whole scan**.
-operand before starting the next. Parallelism lives strictly inside
+Parallelism lives strictly inside each pass; the passes themselves
-each pass; the passes themselves never overlap:
+never overlap:
 
-1.  **walk** — enumerate the tree with the worker pool: each worker
+1.  **walk** — enumerate the trees under all `PATH` operands
-    reads one directory at a time, collecting regular-file paths and
+    concurrently with the worker pool: every operand seeds the shared
-    handing discovered subdirectories back to the shared queue.
+    queue, and each worker reads one directory at a time, collecting
-    Sequential directory enumeration is metadata-latency-bound and
+    regular-file paths and handing discovered subdirectories back to
-    takes hours at tens of millions of files; per-directory
+    the queue. Sequential directory enumeration is
-    parallelism is what makes the walk tractable on large or busy
+    metadata-latency-bound and takes hours at tens of millions of
-    pools. Total unknown while running: show a live count, not a
+    files; per-directory parallelism is what makes the walk tractable
-    percentage.
+    on large or busy pools. Total unknown while running: show a live
+    count, not a percentage.
 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
@@ -208,17 +211,8 @@ each pass; the passes themselves never overlap:
     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.
-4.  **update** — apply the operand's insertions, updates, and
+4.  **update** — apply the scan's insertions, updates, and deletions
-    deletions to the database in a single transaction.
+    to the database in batched transactions.
-
-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:
 
@@ -360,8 +354,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, repeated per operand. Required
+Each scan pass gets its own bar. Required elements for the stat,
-elements for the stat, hash, and update passes (known totals):
+hash, and update passes (known totals):
 
 -   elapsed time
 -   estimated time remaining

TODO.md

@@ -19,6 +19,14 @@
 
 # Completed Steps
 
+- scan-wide phases, concurrent operands, batched updates (2026-07-24,
+  branch `scan-wide-phases`): all operands seed the shared walk pool
+  and every pass runs once over the whole scan, so totals and ETAs
+  are scan-global; the per-operand walk/hash/update cycles and their
+  stderr announcements are gone; the update pass commits in batched
+  transactions — the filesystem is authoritative and the database an
+  eventually-consistent reflection, so scan-level atomicity is not
+  required
 - 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

db.go

@@ -8,6 +8,7 @@ import (
 	"io/fs"
 	"os"
 	"path/filepath"
+	"slices"
 	"strconv"
 
 	// The pure-Go SQLite driver, registered as "sqlite"; keeps cgo
@@ -237,12 +238,40 @@ func loadFileRows(db *sql.DB) ([]scanRec, error) {
 	return recs, nil
 }
 
+// updateBatchSize is the number of record changes committed per
+// transaction during the update pass. The filesystem is authoritative
+// and the database an eventually-consistent reflection of it, so
+// scan-level atomicity is not required; smaller transactions keep the
+// WAL small and let concurrent reports observe progress.
+const updateBatchSize = 10000
+
 // applyChanges writes one scan's database changes — upserts for new and
-// changed files, deletes for vanished ones — in a single transaction,
+// changed files, deletes for vanished ones — in batched transactions.
-// so a concurrent report never observes a half-finished scan. Progress
+// Progress is rendered on prog (one increment per change).
-// is rendered on prog (one increment per change).
 func applyChanges(db *sql.DB, upserts []scanRec, deletes []string,
 	prog *progress,
+) error {
+	for batch := range slices.Chunk(upserts, updateBatchSize) {
+		err := applyBatch(db, batch, nil, prog)
+		if err != nil {
+			return err
+		}
+	}
+
+	for batch := range slices.Chunk(deletes, updateBatchSize) {
+		err := applyBatch(db, nil, batch, prog)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// applyBatch commits one batch of upserts and deletes in a single
+// transaction.
+func applyBatch(db *sql.DB, upserts []scanRec, deletes []string,
+	prog *progress,
 ) error {
 	ctx := context.Background()
 

db_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"errors"
+	"fmt"
 	"path/filepath"
 	"slices"
 	"strings"
@@ -178,3 +179,46 @@ func TestApplyChangesRoundTrip(t *testing.T) {
 		t.Fatalf("rows = %+v, want just %+v", got, upd)
 	}
 }
+
+func TestApplyChangesBatching(t *testing.T) {
+	t.Parallel()
+
+	db := openTestDB(t)
+
+	// One more change than the batch size, so the update spans two
+	// transactions.
+	n := updateBatchSize + 1
+
+	recs := make([]scanRec, 0, n)
+	for i := range n {
+		recs = append(recs, scanRec{
+			size: int64(i), mtime: 1, head: "h", tail: "t",
+			path: fmt.Sprintf("/batch/%07d", i),
+		})
+	}
+
+	err := applyChanges(db, recs, nil, newProgress("update", int64(n)))
+	if err != nil {
+		t.Fatalf("applyChanges: %v", err)
+	}
+
+	got, err := loadFileRows(db)
+	if err != nil || len(got) != n {
+		t.Fatalf("loadFileRows = %d rows, %v; want %d", len(got), err, n)
+	}
+
+	deletes := make([]string, 0, n)
+	for _, r := range recs {
+		deletes = append(deletes, r.path)
+	}
+
+	err = applyChanges(db, nil, deletes, newProgress("update", int64(n)))
+	if err != nil {
+		t.Fatalf("applyChanges deletes: %v", err)
+	}
+
+	got, err = loadFileRows(db)
+	if err != nil || len(got) != 0 {
+		t.Fatalf("loadFileRows = %d rows, %v; want 0", len(got), err)
+	}
+}

scan.go

@@ -99,49 +99,27 @@ type scanStats struct {
 	skipped   int
 }
 
-// syncScan synchronizes the database with the filesystem under roots.
+// syncScan synchronizes the database with the filesystem under roots:
-// Operands are processed in order, each one walked, hashed, and
+// four scan-wide passes — walk, stat, hash, update — each parallel
-// committed independently, so an interrupted scan keeps every operand
+// internally, run strictly in sequence over all operands together.
-// completed so far. Records outside the roots are never touched.
+// Records outside the roots are never touched.
 func syncScan(db *sql.DB, roots []string, workers int,
 	oneFS bool,
 ) (scanStats, error) {
 	var st scanStats
 
-	for i, root := range roots {
+	existing, err := loadScopedRows(db, 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
 	}
-	}
 
-	return st, nil
+	paths, walkErrs := walkPass(roots, oneFS, workers)
-}
+	recs, statErrs := statPass(uniquePaths(paths), workers)
-
-// 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
-	}
-
-	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.unchanged = len(unchanged)
-	st.skipped += walkErrs + statErrs + hashErrs
+	st.skipped = walkErrs + statErrs + hashErrs
 
 	for _, r := range hashed {
 		if _, ok := existing[r.path]; ok {
@@ -152,15 +130,15 @@ func syncRoot(db *sql.DB, root string, workers int, oneFS bool,
 	}
 
 	deletes := collectDeletes(existing, unchanged, hashed)
-	st.removed += len(deletes)
+	st.removed = len(deletes)
 
-	return applyPass(db, hashed, deletes)
+	return st, applyPass(db, hashed, deletes)
 }
 
-// loadScopedRows loads the database records whose paths lie under
+// loadScopedRows loads the database records whose paths lie under any
-// root, keyed by path. Records outside the scanned operands belong to
+// of the scan roots, keyed by path. Records outside the roots belong
-// other trees and are left untouched.
+// to other trees and are left untouched.
-func loadScopedRows(db *sql.DB, root string) (map[string]scanRec, error) {
+func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
 	all, err := loadFileRows(db)
 	if err != nil {
 		return nil, err
@@ -169,7 +147,7 @@ func loadScopedRows(db *sql.DB, root string) (map[string]scanRec, error) {
 	scoped := make(map[string]scanRec)
 
 	for _, r := range all {
-		if underRoot(r.path, root) {
+		if underAnyRoot(r.path, roots) {
 			scoped[r.path] = r
 		}
 	}
@@ -177,6 +155,38 @@ func loadScopedRows(db *sql.DB, root 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
+}
+
+// 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
+}
+
 // underRoot reports whether path is root itself or lies under it. Both
 // must be absolute and lexically clean.
 func underRoot(path, root string) bool {
@@ -278,15 +288,28 @@ type walkEvent struct {
 	fail bool
 }
 
-// walkPass enumerates every regular file under root with a
+// walkPass enumerates every regular file under all roots concurrently
-// per-directory worker pool. It never follows symlinks, never
+// with a per-directory worker pool; every operand seeds the shared
-// descends into directories named .zfs, and warns and continues on
+// queue. It never follows symlinks, never descends into directories
-// any per-path error. With oneFS set it never descends into a
+// named .zfs, and warns and continues on any per-path error. With
-// directory on a different filesystem than root.
+// oneFS set it never descends into a directory on a different
-func walkPass(root string, oneFS bool, workers int) ([]string, int) {
+// filesystem than its root operand.
+func walkPass(roots []string, oneFS bool, workers int) ([]string, int) {
 	prog := newProgress("walk", -1)
 
-	paths, initial, errs := seedRoot(root, prog)
+	var (
+		paths   []string
+		initial []dirJob
+		errs    int
+	)
+
+	for _, root := range roots {
+		p, jobs, e := seedRoot(root, prog)
+		paths = append(paths, p...)
+		initial = append(initial, jobs...)
+		errs += e
+	}
+
 	jobs, subdirs, events := startWalkWorkers(workers, oneFS)
 
 	dispatchDirs(initial, jobs, subdirs)

scan_test.go

@@ -131,7 +131,7 @@ func TestWalkPass(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	paths, errs := walkPass(dir, false, 4)
+	paths, errs := walkPass([]string{dir}, false, 4)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
@@ -161,7 +161,7 @@ func TestWalkPassDeepAndWide(t *testing.T) {
 
 	slices.Sort(want)
 
-	paths, errs := walkPass(dir, false, 8)
+	paths, errs := walkPass([]string{dir}, false, 8)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
@@ -173,6 +173,33 @@ func TestWalkPassDeepAndWide(t *testing.T) {
 	}
 }
 
+func TestWalkPassMultipleRoots(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")),
+	}
+
+	slices.Sort(want)
+
+	// Operands are enumerated concurrently by the shared pool; order
+	// is unspecified.
+	paths, errs := walkPass([]string{rootA, rootB}, false, 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)
+	}
+}
+
 func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
 	t.Parallel()
 
@@ -187,13 +214,13 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
 	}
 
 	// A regular-file operand is emitted as itself.
-	paths, errs := walkPass(f, false, 2)
+	paths, errs := walkPass([]string{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.
-	paths, errs = walkPass(link, false, 2)
+	paths, errs = walkPass([]string{link}, false, 2)
 	if errs != 0 || len(paths) != 0 {
 		t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs)
 	}
@@ -209,7 +236,7 @@ func TestWalkPassOneFilesystemSameFS(t *testing.T) {
 		writeFile(t, dir, "sub/deep/b", []byte("b")),
 	}
 
-	paths, errs := walkPass(dir, true, 4)
+	paths, errs := walkPass([]string{dir}, true, 4)
 	if errs != 0 {
 		t.Fatalf("errs = %d, want 0", errs)
 	}
@@ -623,13 +650,11 @@ func TestSyncScanOverlappingRoots(t *testing.T) {
 
 	writeFile(t, dir, "sub/f", pattern(1, 10))
 
-	// A file reachable via two overlapping operands yields one
+	// A file reachable via two overlapping operands is deduplicated
-	// record: the second operand sees the rows the first operand
+	// by path in the shared walk and processed once.
-	// 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 != (scanStats{added: 1, unchanged: 1}) {
+	if st != (scanStats{added: 1}) {
-		t.Fatalf("stats = %+v, want 1 added 1 unchanged", st)
+		t.Fatalf("stats = %+v, want 1 added", st)
 	}
 
 	if got := recordPaths(dbRecords(t, db)); len(got) != 1 {
@@ -637,6 +662,17 @@ func TestSyncScanOverlappingRoots(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)
+	}
+}
+
 func TestReportsNeverTouchFilesystem(t *testing.T) {
 	t.Parallel()