sfdupes/scan.go

package main

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

// chunk is the number of bytes hashed from each end of a file.
const chunk = 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.
type fileRec struct {
	path  string
	size  int64
	mtime int64
}

// 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: three sequential phases —
// walk (which stats each file as it is discovered), hash, update —
// 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. Flag parsing and the at-least-one-operand check are done
// by cobra.
func runScan(roots []string, workers int, oneFS bool) {
	if workers < 1 {
		workers = 1
	}

	roots = resolveRoots(roots)
	dbPath := databasePath()

	db, err := openScanDatabase(dbPath)
	if err != nil {
		fatalf("%v", err)
	}

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

	st, err := syncScan(db, roots, workers, oneFS)
	if err != nil {
		fatalf("update database %s: %v", 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)
}

// 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 {
	abs := make([]string, 0, len(roots))

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

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

		abs = append(abs, a)
	}

	return abs
}

// 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 three 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),
// and update (record the size-unique files without reading them, and
// delete the records the scan no longer verifies). Records outside
// the roots are never touched.
func syncScan(db *sql.DB, roots []string, workers int,
	oneFS bool,
) (scanStats, error) {
	roots = pruneRoots(roots)

	s := &scanState{db: db}

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

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

	s.partition(changed, unhashed)

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

	return s.st, s.updatePhase()
}

// 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(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(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 further. 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++
}

// hashPhase hashes every queued file with the worker pool, 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).
// The total is exact, so the bar shows a real ETA. Files that fail to
// hash are warned about and skipped; their stale records, if any, are
// deleted by the update phase.
func (s *scanState) hashPhase(workers int) error {
	jobs := make(chan fileRec, workQueueDepth)
	results := make(chan hashResult, workQueueDepth)

	startHashWorkers(jobs, results, workers)

	// The feeder ranges over its own reference: s.toHash is released
	// below while the feeder may still be running.
	toHash := s.toHash
	s.toHash = nil

	go func() {
		for _, rec := range toHash {
			jobs <- rec
		}

		close(jobs)
	}()

	prog := newProgress("hash", int64(len(toHash)))
	defer prog.finish()

	for range toHash {
		r := <-results

		prog.increment()

		if r.err != nil {
			s.st.skipped++

			prog.warnf("hash %s: %v", r.rec.path, r.err)

			continue
		}

		s.resolve(r.rec.path)

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

		if len(s.batch) < updateBatchSize {
			continue
		}

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

		s.batch = s.batch[:0]
	}

	return nil
}

// 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() 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(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(s.db, recs, nil, prog)
		if err != nil {
			return err
		}
	}

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

// 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.
func startWalk(roots []string, oneFS bool, workers int) <-chan walkEvent {
	jobs, subdirs, events := startWalkWorkers(workers, oneFS)

	go func() {
		initial := make([]dirJob, 0, len(roots))
		for _, root := range roots {
			initial = append(initial, seedRoot(root, events)...)
		}

		dispatchDirs(initial, jobs, subdirs)
	}()

	return events
}

// seedRoot turns one PATH operand into the walk's starting state: a
// regular-file operand is statted and emitted directly, a directory
// operand becomes an initial job, and a symlink or other non-regular
// operand yields nothing (symlinks are never followed, including as
// operands).
func seedRoot(root string, events chan<- walkEvent) []dirJob {
	fi, err := os.Lstat(root)
	if err != nil {
		events <- walkEvent{
			warn: fmt.Sprintf("walk %s: %v", root, err),
			fail: true,
		}

		return nil
	}

	switch {
	case fi.IsDir():
		if filepath.Base(root) == ".zfs" {
			return nil
		}

		dev, ok := deviceOfInfo(fi)

		return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
	case fi.Mode().IsRegular():
		events <- walkEvent{rec: fileRec{
			path:  root,
			size:  fi.Size(),
			mtime: fi.ModTime().Unix(),
		}}

		return nil
	default:
		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.
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
	}

	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 !e.Type().IsRegular() {
			continue
		}

		emitFile(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(p string, e fs.DirEntry, events chan<- walkEvent) {
	info, err := e.Info()
	if err != nil {
		events <- walkEvent{
			warn: fmt.Sprintf("stat %s: %v", p, err),
			fail: true,
		}

		return
	}

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

	events <- walkEvent{rec: fileRec{
		path:  p,
		size:  info.Size(),
		mtime: info.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
// 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
	}

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

	info, err := e.Info()
	if err != nil {
		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
}

// hashResult carries one file's head/tail hashes (or the error that
// prevented hashing it) from the hash workers to the hash phase.
type hashResult struct {
	rec  fileRec
	head string
	tail string
	err  error
}

// startHashWorkers starts the hash worker pool: workers read jobs,
// write one result per record, and exit when jobs is closed.
func startHashWorkers(jobs <-chan fileRec, results chan<- hashResult,
	workers int,
) {
	for range workers {
		go func() {
			for rec := range jobs {
				head, tail, err := hashHeadTail(rec.path, rec.size)
				results <- hashResult{
					rec: rec, head: head, tail: tail, err: err,
				}
			}
		}()
	}
}

// hashHeadTail returns the lowercase-hex SHA-256 of the first
// min(chunk, size) bytes and of the last min(chunk, size) bytes of the
// file at path. The two reads overlap when size < 2*chunk; for
// size == 0 both hashes are of the empty input. size is the value
// recorded when the file was statted.
func hashHeadTail(path string, size int64) (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() }()

	n := min(int64(chunk), size)

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

	h := sha256.Sum256(buf)

	// When the whole file fits in one chunk the tail window is exactly
	// the bytes just read: reuse the head hash instead of issuing a
	// second read for every small file.
	if size <= int64(chunk) {
		hh := hex.EncodeToString(h[:])

		return hh, hh, nil
	}

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

	t := sha256.Sum256(buf)

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