Make phases scan-wide, walk operands concurrently, batch updates

All PATH operands belong to a single scan: every operand seeds the
shared walk worker pool, and each pass (walk, stat, hash, update)
runs exactly once over the whole scan, so pass totals, percentages,
and ETAs are scan-global. The per-operand walk/hash/update cycles and
their stderr operand announcements are gone; duplicate paths from
overlapping operands are deduplicated before stat.

The update pass now commits in batched transactions (10k changes per
batch) instead of one scan-wide transaction: the filesystem is
authoritative and the database is an eventually-consistent reflection
of it, so scan-level atomicity buys nothing, while batches keep the
WAL small and let concurrent reports observe progress.

scan.go

@@ -99,49 +99,27 @@ type scanStats struct {
 	skipped   int
 }
 
-// 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.
+// syncScan synchronizes the database with the filesystem under roots:
+// four scan-wide passes — walk, stat, hash, update — each parallel
+// internally, run strictly in sequence over all operands together.
+// 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 {
-		// 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
-}
-
-// 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)
+	existing, err := loadScopedRows(db, roots)
 	if err != nil {
-		return err
+		return st, err
 	}
 
-	paths, walkErrs := walkPass(root, oneFS, workers)
-	recs, statErrs := statPass(paths, workers)
+	paths, walkErrs := walkPass(roots, oneFS, workers)
+	recs, statErrs := statPass(uniquePaths(paths), workers)
 	toHash, unchanged := partitionChanged(recs, existing)
 	hashed, hashErrs := hashPass(toHash, workers)
 
-	st.unchanged += len(unchanged)
-	st.skipped += walkErrs + statErrs + hashErrs
+	st.unchanged = len(unchanged)
+	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
-// 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) {
+// 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.
+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
-// 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) ([]string, int) {
+// walkPass enumerates every regular file under all roots concurrently
+// with a per-directory worker pool; every operand seeds the shared
+// queue. 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, 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)