sfdupes/scan.go

package main

import (
	"cmp"
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"sync"
	"syscall"
)

// The duplicate ladder (see hashSignature and README "Duplicate
// detection"). A same-size candidate below headTailMin is hashed in
// full and compared directly; a larger one is separated first by the
// hashes of its end windows, then by a content hash that is exact below
// wholeFileMax and deliberately sampled at or above it. The hash phase
// reads only the end windows of a larger file; the content phase reads
// it for its content hash only once its size, head, and tail match
// another file's.

// headTailMin is the size threshold for the end-window gate. A file
// smaller than this is hashed in full directly, with no separate head
// and tail step: its head, tail, and content all carry the whole-file
// hash. A file this size or larger is separated first by its end
// windows.
const headTailMin = 10 * 1024 * 1024

// headTailWindow is the number of bytes hashed from each end of a file
// at or above headTailMin (the head and tail rungs). Because
// headTailMin is far larger than two windows, the head and tail windows
// never overlap.
const headTailWindow = 64 * 1024

// wholeFileMax is the size boundary between the two content rungs: a
// file strictly smaller than this is content-hashed in full; a file
// this size or larger is content-hashed by sampling.
const wholeFileMax = 50 * 1024 * 1024

// sampleStride is the spacing between content samples for large files:
// one window is read at each gigabyte-aligned offset (0, 1 GiB, ...).
const sampleStride = 1024 * 1024 * 1024

// sampleWindow is the number of bytes read at each large-file sample
// offset, truncated at end of file.
const sampleWindow = 1024 * 1024

// workQueueDepth bounds the job and result channels feeding the walk
// and hash worker pools.
const workQueueDepth = 1024

// fileRec carries one statted file between the scan phases. dev and
// ino identify the underlying inode so hard-linked paths can share
// one read; both are zero when the platform exposes no inode.
type fileRec struct {
	path  string
	size  int64
	mtime int64
	dev   uint64
	ino   uint64
}

// fileMeta is the in-memory index entry for one existing database
// record: just enough for change detection, plus whether the record
// carries hashes. Hashes stay on disk; at tens of millions of records
// they would dominate the scan's memory.
type fileMeta struct {
	size   int64
	mtime  int64
	hashed bool
}

// runScan implements the scan subcommand: four sequential phases —
// walk (which stats each file as it is discovered), hash, update,
// content — that synchronize the persistent database with the
// filesystem state under the PATH operands. Only files whose size at
// least one other file shares are ever hashed: a size-unique file
// cannot be a duplicate. A file of headTailMin or more gets its content
// hash only when its size, head, and tail match another file's. 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
// takes the database out of WAL mode — always runs. 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
	}

	roots, err := resolveRoots(roots)
	if err != nil {
		return err
	}

	dbPath := databasePath()

	db, err := openScanDatabase(ctx, dbPath)
	if err != nil {
		return err
	}

	defer closeScanDatabase(ctx, db, dbPath)

	st, err := syncScan(ctx, db, roots, workers, oneFS)
	if err != nil {
		return fmt.Errorf("update database %s: %w", dbPath, err)
	}

	fmt.Fprintf(os.Stderr,
		"scan: %d files seen (%d added, %d updated, %d removed, "+
			"%d unchanged), %d skipped\n",
		st.added+st.updated+st.unchanged, st.added, st.updated,
		st.removed, st.unchanged, st.skipped)

	return nil
}

// resolveRoots converts each PATH operand to an absolute, lexically
// cleaned path (symlinks are not resolved) and verifies that it
// exists. Database records are keyed by absolute path, so scan results
// must not depend on the working directory.
func resolveRoots(roots []string) ([]string, error) {
	abs := make([]string, 0, len(roots))

	for _, root := range roots {
		a, err := filepath.Abs(root)
		if err != nil {
			return nil, fmt.Errorf("resolve %s: %w", root, err)
		}

		// A nonexistent operand is a fatal error before any scanning.
		_, err = os.Lstat(a)
		if err != nil {
			return nil, err
		}

		abs = append(abs, a)
	}

	return abs, nil
}

// pruneRoots drops operands already covered by another operand:
// duplicates and any operand lying under another one. Every remaining
// file is then reachable through exactly one root, so walked paths are
// unique without keeping a scan-wide set of every path seen.
func pruneRoots(roots []string) []string {
	// Shorter-first ordering guarantees an ancestor is kept before any
	// operand under it is considered.
	sorted := slices.Clone(roots)
	slices.SortFunc(sorted, func(a, b string) int {
		if len(a) != len(b) {
			return len(a) - len(b)
		}

		return strings.Compare(a, b)
	})

	kept := make([]string, 0, len(sorted))

	for _, r := range sorted {
		if !underAnyRoot(r, kept) {
			kept = append(kept, r)
		}
	}

	return kept
}

// scanStats summarizes one scan's database synchronization for the
// final stderr summary.
type scanStats struct {
	added     int
	updated   int
	removed   int
	unchanged int
	skipped   int
}

// scanState carries one scan's evolving state across its phases.
// Entries consumed from existing mark files verified this run;
// whatever remains after the walk and hash phases is deleted by the
// update phase.
type scanState struct {
	db        *sql.DB
	existing  map[string]fileMeta
	sizes     []int64   // size census: every walked file, plus records outside the roots
	toHash    []fileRec // files whose size is shared: must be read
	sentinels []fileRec // new/changed size-unique files: recorded without hashes
	batch     []scanRec
	st        scanStats
}

// syncScan synchronizes the database with the filesystem under roots
// in four sequential phases: walk (enumerate and stat every file,
// building a complete size census), hash (read only the new or
// changed — or previously unhashed — files whose size at least one
// other file shares, committing results in batches as they arrive),
// update (record the size-unique files without reading them, and
// delete the records the scan no longer verifies), and content (fill
// in the content hash of every record of headTailMin or more whose
// size, head, and tail match another record's). Records outside the
// roots are never touched, except that the content phase fills in
// their content hash. Operands the walk cannot start from are dropped
// first, so the records beneath them count as outside the roots unless
// they lie under another root.
func syncScan(ctx context.Context, db *sql.DB, roots []string,
	workers int, oneFS bool,
) (scanStats, error) {
	s := &scanState{db: db}

	// Types are checked before pruning so that an operand under a
	// dropped one is still scanned, not dropped as lying under it.
	roots = pruneRoots(s.walkableRoots(roots))

	err := s.loadIndex(ctx, roots)
	if err != nil {
		return s.st, err
	}

	changed, unhashed := s.walkPhase(startWalk(ctx, roots, oneFS, workers))

	// A cancelled walk stops early, so its size census covers only part
	// 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
	}

	s.partition(changed, unhashed)

	err = s.hashPhase(ctx, workers)
	if err != nil {
		return s.st, err
	}

	err = s.updatePhase(ctx)
	if err != nil {
		return s.st, err
	}

	return s.st, s.contentPhase(ctx, workers)
}

// walkableRoots returns the operands the walk can start from: regular
// files, and directories not named .zfs. Every other operand is warned
// about, counted as skipped, and dropped. A dropped operand is no
// longer a root, so the records stored beneath it count as outside the
// roots and are not deleted as unverified, unless it lies under another
// root. An operand that fails lstat here is kept, and the walk warns
// about it.
func (s *scanState) walkableRoots(roots []string) []string {
	kept := make([]string, 0, len(roots))

	for _, root := range roots {
		fi, err := os.Lstat(root)
		if err == nil {
			warn := operandWarning(root, fi)
			if warn != "" {
				s.st.skipped++

				fmt.Fprintln(os.Stderr, escapePath(warn))

				continue
			}
		}

		kept = append(kept, root)
	}

	return kept
}

// loadIndex indexes the database records under the scan roots for
// change detection and collects the sizes of every record outside
// 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(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)
	defer prog.finish()

	s.existing = make(map[string]fileMeta)

	return loadFileMeta(ctx, s.db,
		func(path string, size, mtime int64, hashed bool) {
			prog.increment()

			if underAnyRoot(path, roots) {
				s.existing[path] = fileMeta{
					size: size, mtime: mtime, hashed: hashed,
				}

				return
			}

			s.sizes = append(s.sizes, size)
		})
}

// walkPhase drains the walk, appending every walked file's size to
// the census and resolving what it can immediately: an unchanged file
// whose record already has hashes needs nothing from the hash phase
// (the content phase may still fill in its content hash). It returns
// the new-or-changed files and the unchanged files whose records lack
// hashes; both remain candidates until the census decides whether
// their sizes are shared.
func (s *scanState) walkPhase(
	events <-chan walkEvent,
) ([]fileRec, []fileRec) {
	prog := newProgress("walk", -1)

	var changed, unhashed []fileRec

	for ev := range events {
		if ev.fail {
			s.st.skipped++

			prog.warnf("%s", ev.warn)

			continue
		}

		s.sizes = append(s.sizes, ev.rec.size)

		prog.increment()

		old, ok := s.existing[ev.rec.path]

		switch {
		case !ok || old.size != ev.rec.size || old.mtime < ev.rec.mtime:
			changed = append(changed, ev.rec)
		case old.hashed:
			delete(s.existing, ev.rec.path)

			s.st.unchanged++
		default:
			unhashed = append(unhashed, ev.rec)
		}
	}

	prog.finish()

	return changed, unhashed
}

// partition decides each candidate's disposition now that the size
// census is complete. A file whose size no other file shares cannot
// be a duplicate and is never read: a new or changed one is recorded
// without hashes, an unchanged unhashed one keeps its record. Every
// file with a shared size queues for the hash phase.
func (s *scanState) partition(changed, unhashed []fileRec) {
	slices.Sort(s.sizes)

	for _, rec := range changed {
		if s.sizeShared(rec.size) {
			s.toHash = append(s.toHash, rec)

			continue
		}

		s.sentinels = append(s.sentinels, rec)
		s.resolve(rec.path)
	}

	for _, rec := range unhashed {
		if s.sizeShared(rec.size) {
			s.toHash = append(s.toHash, rec)

			continue
		}

		delete(s.existing, rec.path)

		s.st.unchanged++
	}

	s.sizes = nil
}

// sizeShared reports whether at least two census entries have this
// size. Every candidate's own size is in the census exactly once, so
// a second entry means another file (or a record outside the scan
// roots) could share its content.
func (s *scanState) sizeShared(size int64) bool {
	i, found := slices.BinarySearch(s.sizes, size)

	return found && i+1 < len(s.sizes) && s.sizes[i+1] == size
}

// resolve counts one written record as added or updated and marks its
// path verified.
func (s *scanState) resolve(path string) {
	if _, ok := s.existing[path]; ok {
		s.st.updated++

		delete(s.existing, path)

		return
	}

	s.st.added++
}

// hashRuns sorts the queued files into inode order and groups paths
// sharing an inode into runs. Inode-ordered reads minimize seeking on
// spinning disks, and each run of hard-linked paths is read once,
// with every path sharing the result. Files without an inode identity
// are never merged.
func hashRuns(toHash []fileRec) [][]fileRec {
	slices.SortFunc(toHash, func(a, b fileRec) int {
		if c := cmp.Compare(a.dev, b.dev); c != 0 {
			return c
		}

		if c := cmp.Compare(a.ino, b.ino); c != 0 {
			return c
		}

		return strings.Compare(a.path, b.path)
	})

	var runs [][]fileRec

	start := 0

	for i := 1; i <= len(toHash); i++ {
		if i == len(toHash) || !sameInode(toHash[i-1], toHash[i]) {
			runs = append(runs, toHash[start:i])
			start = i
		}
	}

	return runs
}

// sameInode reports whether two records name the same underlying
// inode. Records without an inode identity never match.
func sameInode(a, b fileRec) bool {
	return (a.dev != 0 || a.ino != 0) && a.dev == b.dev && a.ino == b.ino
}

// hashPhase hashes every queued file with hashSignature — the head and
// tail of a file of headTailMin or more, the whole file below that —
// committing completed records to the database in batches as results
// arrive, so a long scan persists its progress as it goes (an
// interrupted scan resumes cheaply: the next run skips everything
// already recorded). 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(ctx context.Context, workers int) error {
	runs := hashRuns(s.toHash)
	s.toHash = nil

	return s.readRuns(ctx, workers, "hash", runs, hashSignature, s.recordRun)
}

// readRuns reads runs with the worker pool, one read per inode run, in
// the order given, under a progress display named label. The workers
// compute each run's hashes with hash, and each result goes to record;
// a run that fails to read is warned about and counted as skipped
// instead. The total counts actual reads, so the bar shows a real ETA.
//
// 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) readRuns(ctx context.Context, workers int,
	label string, runs [][]fileRec,
	hash func(path string, size int64) (string, string, string, error),
	record func(ctx context.Context, r hashResult) error,
) error {
	pool := startHashPool(ctx, runs, workers, hash)
	defer pool.stop()

	prog := newProgress(label, int64(len(runs)))
	defer prog.finish()

	for range runs {
		var r hashResult

		select {
		case r = <-pool.results:
		case <-ctx.Done():
			return ctx.Err()
		}

		prog.increment()

		if r.err != nil {
			s.st.skipped += len(r.run)

			prog.warnf("%s %s: %v", label, r.run[0].path, r.err)

			continue
		}

		err := record(ctx, r)
		if err != nil {
			return err
		}
	}

	return nil
}

// 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(ctx context.Context, r hashResult) error {
	for _, rec := range r.run {
		s.resolve(rec.path)

		s.batch = append(s.batch, scanRec{
			size:    rec.size,
			mtime:   rec.mtime,
			head:    r.head,
			tail:    r.tail,
			content: r.content,
			path:    rec.path,
		})
	}

	return s.commitFullBatch(ctx)
}

// commitFullBatch commits the running batch once it holds
// updateBatchSize records.
func (s *scanState) commitFullBatch(ctx context.Context) error {
	if len(s.batch) < updateBatchSize {
		return nil
	}

	err := applyBatch(ctx, s.db, s.batch, nil, nil)
	s.batch = s.batch[:0]

	return err
}

// updatePhase writes the scan's tail under one progress display: the
// final partial batch of hashed records, a hash-less record for every
// 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(ctx context.Context) error {
	deletes := make([]string, 0, len(s.existing))
	for path := range s.existing {
		deletes = append(deletes, path)
	}

	// Sorted deletes keep the update phase deterministic.
	slices.Sort(deletes)

	s.st.removed = len(deletes)

	total := len(s.batch) + len(s.sentinels) + len(deletes)
	prog := newProgress("update", int64(total))

	defer prog.finish()

	err := applyChanges(ctx, s.db, s.batch, nil, prog)
	if err != nil {
		return err
	}

	s.batch = nil

	// Sentinel records are converted in batch-sized chunks rather than
	// materialized all at once; a first scan can have millions.
	for chunk := range slices.Chunk(s.sentinels, updateBatchSize) {
		recs := make([]scanRec, 0, len(chunk))
		for _, rec := range chunk {
			recs = append(recs, scanRec{
				size: rec.size, mtime: rec.mtime, path: rec.path,
			})
		}

		err = applyBatch(ctx, s.db, recs, nil, prog)
		if err != nil {
			return err
		}
	}

	return applyChanges(ctx, s.db, nil, deletes, prog)
}

// contentPhase fills in the content hash of every record of headTailMin
// or more that lacks one and whose size, head, and tail equal another
// record's, anywhere in the database: records from this scan and
// records stored by earlier scans, inside or outside the roots. Only
// such a file can still be a duplicate, so no other file of headTailMin
// or more is read beyond its end windows. The files are read with the
// hash phase's worker pool and their records written back in batches. A
// failed read is warned about and counted as skipped; the record keeps
// its empty content, so it is never grouped, and a later scan tries
// again.
func (s *scanState) contentPhase(ctx context.Context, workers int) error {
	toRead, recs, err := s.contentCandidates(ctx)
	if err != nil {
		return err
	}

	err = s.readRuns(ctx, workers, "content", hashRuns(toRead),
		hashContentOnly, func(ctx context.Context, r hashResult) error {
			// Every path in the run keeps its record's head and tail
			// and gains the one content hash read for the run.
			for _, f := range r.run {
				rec := recs[f.path]
				rec.content = r.content
				s.batch = append(s.batch, rec)
			}

			return s.commitFullBatch(ctx)
		})
	if err != nil {
		return err
	}

	return applyChanges(ctx, s.db, s.batch, nil, nil)
}

// contentCandidates returns the files the content phase reads, and
// their records by path. Every record contentCandidatesSQL returns has
// its file checked with lstat, whether or not it already has a content
// hash: a file that is gone, is no longer a regular file, or has
// changed by the walk's rule keeps its record as it is and does not
// count as a match for the others, and any other lstat error is warned
// about and counted as skipped, with the same result. If such a record
// has no content hash, it stays out of duplicate groups; if it has one,
// it is still reported until a scan covering its own tree updates or
// removes it. The files of a group that pass and have no content hash
// are read only if at least minGroupSize of the group's files pass, so
// a group whose other members are all stale costs no reads. Only the
// records to be read are kept.
func (s *scanState) contentCandidates(
	ctx context.Context,
) ([]fileRec, map[string]scanRec, error) {
	// The query and the checks take real time on a large database;
	// without a display the scan looks hung before the reads begin.
	prog := newProgress("content", -1)
	defer prog.finish()

	var (
		toRead []fileRec
		first  scanRec   // the current group's first record
		passed int       // the current group's files that passed the check
		unread []fileRec // those of them without a content hash
	)

	recs := make(map[string]scanRec)

	// endGroup queues the current group's files to read if at least
	// minGroupSize of its files passed, and drops their records if not.
	endGroup := func() {
		if passed >= minGroupSize {
			toRead = append(toRead, unread...)
		} else {
			for _, f := range unread {
				delete(recs, f.path)
			}
		}

		passed, unread = 0, nil
	}

	err := loadContentCandidates(ctx, s.db, func(r scanRec, hashed bool) {
		prog.increment()

		if r.size != first.size || r.head != first.head || r.tail != first.tail {
			endGroup()

			first = r
		}

		f, ok, err := unchangedFile(r)
		if err != nil {
			s.st.skipped++

			prog.warnf("content %s: %v", r.path, err)
		}

		if !ok {
			return
		}

		passed++

		if !hashed {
			unread = append(unread, f)
			recs[r.path] = r
		}
	})
	if err != nil {
		return nil, nil, err
	}

	endGroup()

	return toRead, recs, nil
}

// unchangedFile lstats the file r names and returns it for reading if
// it is still the regular file r records: the same size, and an mtime
// no newer than recorded (the walk's change rule). A file that is gone
// or has changed reports false; any other lstat error is returned.
func unchangedFile(r scanRec) (fileRec, bool, error) {
	fi, err := os.Lstat(r.path)
	if errors.Is(err, fs.ErrNotExist) {
		return fileRec{}, false, nil
	}

	if err != nil {
		return fileRec{}, false, err
	}

	if !fi.Mode().IsRegular() || fi.Size() != r.size ||
		fi.ModTime().Unix() > r.mtime {
		return fileRec{}, false, nil
	}

	dev, ino := inodeOfInfo(fi)

	return fileRec{
		path: r.path, size: r.size, mtime: r.mtime, dev: dev, ino: ino,
	}, true, 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 {
	if path == root {
		return true
	}

	prefix := root
	if !strings.HasSuffix(prefix, "/") {
		prefix += "/"
	}

	return strings.HasPrefix(path, prefix)
}

// 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
}

// walkEvent is one walk result delivered to the walk phase: a regular
// file's statted record, or a warning when fail is set.
type walkEvent struct {
	rec  fileRec
	warn string
	fail bool
}

// 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, 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(ctx, root, events)...)
		}

		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():
	}
}

// operandWarning returns the one-line warning for an operand the walk
// does not start from, naming the path and what it is, or "" for one it
// does: a regular file, or a directory not named .zfs. Symlinks are
// never followed, including as operands.
func operandWarning(root string, fi fs.FileInfo) string {
	var kind string

	switch mode := fi.Mode(); {
	case mode.IsRegular():
		return ""
	case mode.IsDir():
		if filepath.Base(root) != ".zfs" {
			return ""
		}

		kind = ".zfs directory"
	case mode&fs.ModeSymlink != 0:
		kind = "symlink"
	case mode&fs.ModeSocket != 0:
		kind = "socket"
	case mode&fs.ModeNamedPipe != 0:
		kind = "FIFO"
	case mode&fs.ModeDevice != 0:
		kind = "device node"
	default:
		kind = "non-regular file"
	}

	return fmt.Sprintf("walk %s: skipping %s operand", root, kind)
}

// seedRoot turns one PATH operand into the walk's starting state: a
// regular-file operand is statted and emitted directly, and a directory
// operand becomes an initial job. walkableRoots has already dropped
// every other operand. One that has changed into something else since
// is warned about and skipped here; it is still a root, so the records
// stored beneath it are deleted as unverified.
func seedRoot(ctx context.Context, root string,
	events chan<- walkEvent,
) []dirJob {
	fi, err := os.Lstat(root)
	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", root, err),
			fail: true,
		})

		return nil
	}

	warn := operandWarning(root, fi)
	if warn != "" {
		sendEvent(ctx, events, walkEvent{warn: warn, fail: true})

		return nil
	}

	if fi.IsDir() {
		dev, ok := deviceOfInfo(fi)

		return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
	}

	dev, ino := inodeOfInfo(fi)

	sendEvent(ctx, events, walkEvent{rec: fileRec{
		path:  root,
		size:  fi.Size(),
		mtime: fi.ModTime().Unix(),
		dev:   dev,
		ino:   ino,
	}})

	return nil
}

// 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. 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)
	subdirs := make(chan []dirJob, workers)
	events := make(chan walkEvent, workQueueDepth)

	var wg sync.WaitGroup

	for range workers {
		wg.Go(func() {
			for job := range jobs {
				if ctx.Err() != nil {
					continue
				}

				select {
				case subdirs <- walkOneDir(ctx, job, oneFS, events):
				case <-ctx.Done():
				}
			}
		})
	}

	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. 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)

		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...)
			case <-ctx.Done():
				return
			}
		}
	}()
}

// 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(ctx context.Context, job dirJob, oneFS bool,
	events chan<- walkEvent,
) []dirJob {
	entries, err := os.ReadDir(job.path)
	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", job.path, err),
			fail: true,
		})

		return nil
	}

	var subs []dirJob

	for _, e := range entries {
		p := filepath.Join(job.path, e.Name())

		if e.IsDir() {
			if sub, ok := subdirJob(ctx, p, e, job, oneFS, events); ok {
				subs = append(subs, sub)
			}

			continue
		}
		// Regular files only: skip symlinks, sockets, FIFOs, and
		// device nodes.
		if !e.Type().IsRegular() {
			continue
		}

		emitFile(ctx, p, e, events)
	}

	return subs
}

// emitFile stats one regular directory entry and emits its record.
// The lstat happens here in the walk worker, while the directory's
// 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(ctx context.Context, p string, e fs.DirEntry,
	events chan<- walkEvent,
) {
	info, err := e.Info()
	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("stat %s: %v", p, err),
			fail: true,
		})

		return
	}

	if !info.Mode().IsRegular() {
		return
	}

	dev, ino := inodeOfInfo(info)

	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(ctx context.Context, p string, e fs.DirEntry,
	parent dirJob, oneFS bool, events chan<- walkEvent,
) (dirJob, bool) {
	if e.Name() == ".zfs" {
		return dirJob{}, false
	}

	job := dirJob{path: p, rootDev: parent.rootDev, rootDevOK: parent.rootDevOK}
	if !oneFS || !parent.rootDevOK {
		return job, true
	}

	info, err := e.Info()
	if err != nil {
		sendEvent(ctx, events, walkEvent{
			warn: fmt.Sprintf("walk %s: %v", p, err),
			fail: true,
		})

		return dirJob{}, false
	}

	if dev, ok := deviceOfInfo(info); ok && dev != parent.rootDev {
		return dirJob{}, false
	}

	return job, true
}

// deviceOfInfo extracts the filesystem device ID from a FileInfo, when
// the platform exposes one.
func deviceOfInfo(fi fs.FileInfo) (uint64, bool) {
	st, ok := fi.Sys().(*syscall.Stat_t)
	if !ok {
		return 0, false
	}

	return statDev(st), true
}

// inodeOfInfo extracts the (device, inode) pair identifying a file's
// underlying inode; (0, 0) when the platform exposes none (such files
// are never merged as hard links).
func inodeOfInfo(fi fs.FileInfo) (uint64, uint64) {
	st, ok := fi.Sys().(*syscall.Stat_t)
	if !ok {
		return 0, 0
	}

	return statDev(st), st.Ino
}

// hashResult carries the hashes computed for one inode run (or the
// error that prevented computing them) from the pool's workers to the
// phase that started the pool: head, tail, and content from
// hashSignature, content alone from hashContentOnly.
type hashResult struct {
	run     []fileRec
	head    string
	tail    string
	content string
	err     error
}

// 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 with hash (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,
	hash func(path string, size int64) (string, string, string, error),
) *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 {
		wg.Go(func() { hashWorker(ctx, jobs, results, hash) })
	}

	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 with hash 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,
	hash func(path string, size int64) (string, string, string, error),
) {
	for run := range jobs {
		if ctx.Err() != nil {
			continue
		}

		head, tail, content, err := hash(run[0].path, run[0].size)

		select {
		case results <- hashResult{
			run: run, head: head, tail: tail, content: content, err: err,
		}:
		case <-ctx.Done():
			return
		}
	}
}

// emptyHash is the lowercase-hex SHA-256 of the empty input: the head,
// tail, and content hash of every zero-length file.
const emptyHash = "e3b0c44298fc1c149afbf4c8996fb924" +
	"27ae41e4649b934ca495991b7852b855"

// hashSignature computes the hashes the hash phase records for a file
// whose size is shared; with the file size they form its duplicate
// signature. A file below headTailMin is hashed in full and its
// whole-file SHA-256 is returned as head, tail, and content alike —
// that range takes no separate end-window step. For a file at or above
// headTailMin only the head and tail are computed, the SHA-256 of its
// first and last headTailWindow bytes, and content is returned empty:
// the content phase computes it with hashContentOnly once the file's
// size, head, and tail match another file's. Two files are duplicates
// only when all four agree; any mismatch means not a duplicate. size
// is the value recorded when the file was statted; a zero-length file
// has constant hashes and is never opened.
func hashSignature(path string, size int64) (string, string, string, error) {
	if size == 0 {
		return emptyHash, emptyHash, emptyHash, nil
	}

	//nolint:gosec // hashing operator-supplied paths is the tool's purpose
	f, err := os.Open(path)
	if err != nil {
		return "", "", "", err
	}

	defer func() { _ = f.Close() }()

	// Below the threshold the whole file is hashed directly, with no
	// end-window step: head and tail both carry the whole-file hash.
	if size < int64(headTailMin) {
		content, err := hashWhole(f, size)
		if err != nil {
			return "", "", "", err
		}

		return content, content, content, nil
	}

	head, tail, err := hashEnds(f, size)
	if err != nil {
		return "", "", "", err
	}

	return head, tail, "", nil
}

// hashContentOnly returns the content hash of the file at path, which
// is at least headTailMin bytes: the content phase's read. head and
// tail are returned empty, because the content phase keeps the ones its
// records already hold.
func hashContentOnly(path string, size int64) (string, string, string, error) {
	//nolint:gosec // hashing operator-supplied paths is the tool's purpose
	f, err := os.Open(path)
	if err != nil {
		return "", "", "", err
	}

	defer func() { _ = f.Close() }()

	content, err := hashContent(f, size)

	return "", "", content, err
}

// hashEnds returns the SHA-256 of the first and last headTailWindow
// bytes of f. It is called only for files at least headTailMin, which
// is far larger than two windows, so the windows never overlap and both
// reads are always full.
func hashEnds(f *os.File, size int64) (string, string, error) {
	buf := make([]byte, headTailWindow)

	_, err := f.ReadAt(buf, 0)
	if err != nil {
		return "", "", err
	}

	h := sha256.Sum256(buf)
	head := hex.EncodeToString(h[:])

	_, err = f.ReadAt(buf, size-int64(headTailWindow))
	if err != nil {
		return "", "", err
	}

	t := sha256.Sum256(buf)

	return head, hex.EncodeToString(t[:]), nil
}

// hashContent returns the content-rung hash of f: the SHA-256 of the
// whole file when it is smaller than wholeFileMax, or of sampled
// windows when it is that size or larger.
func hashContent(f *os.File, size int64) (string, error) {
	if size >= int64(wholeFileMax) {
		return hashSamples(f, size)
	}

	return hashWhole(f, size)
}

// hashWhole returns the SHA-256 of the entire file. A SectionReader is
// used so the read is independent of the offset left by any end-window
// reads. Reading fewer than size bytes means the file shrank between
// the stat and the hash; that is an error rather than a hash of content
// that no longer matches the recorded size.
func hashWhole(f *os.File, size int64) (string, error) {
	h := sha256.New()

	n, err := io.Copy(h, io.NewSectionReader(f, 0, size))
	if err != nil {
		return "", err
	}

	if n != size {
		return "", fmt.Errorf("read %d of %d bytes: %w", n, size,
			io.ErrUnexpectedEOF)
	}

	return hex.EncodeToString(h.Sum(nil)), nil
}

// hashSamples feeds sampleWindow bytes at each gigabyte-aligned offset
// (0, sampleStride, 2*sampleStride, ... while inside the file), in
// order, into one hash, each window truncated at end of file. This is
// the probabilistic large-file rung: two files of equal size agreeing
// on every sample are reported as duplicates without every byte being
// read. Because size is part of the signature, files of different sizes
// never reach this comparison, so the sample boundaries always align.
func hashSamples(f *os.File, size int64) (string, error) {
	h := sha256.New()
	buf := make([]byte, sampleWindow)

	for off := int64(0); off < size; off += int64(sampleStride) {
		n := min(int64(sampleWindow), size-off)

		_, err := f.ReadAt(buf[:n], off)
		if err != nil {
			return "", err
		}

		h.Write(buf[:n])
	}

	return hex.EncodeToString(h.Sum(nil)), nil
}