sfdupes/scan.go

package main

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"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 stat
// and hash worker pools.
const workQueueDepth = 1024

// errNotRegular reports a path whose type changed between the
// directory read and its lstat.
var errNotRegular = errors.New("no longer a regular file")

// fileRec carries one file between the stat and hash passes.
type fileRec struct {
	path  string
	size  int64
	mtime int64
}

// runScan implements the scan subcommand: four sequential passes
// (walk, stat, hash, update) that synchronize the persistent database
// with the filesystem state under the PATH operands. 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
}

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

// 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.
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)
	if err != nil {
		return err
	}

	walked, walkErrs := walkPass(root, oneFS, workers)
	toHash, unchanged := partitionChanged(walked, existing)
	hashed, hashErrs := hashPass(toHash, workers)

	st.unchanged += len(unchanged)
	st.skipped += walkErrs + hashErrs

	for _, r := range hashed {
		if _, ok := existing[r.path]; ok {
			st.updated++
		} else {
			st.added++
		}
	}

	deletes := collectDeletes(existing, unchanged, hashed)
	st.removed += len(deletes)

	return 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) {
	all, err := loadFileRows(db)
	if err != nil {
		return nil, err
	}

	scoped := make(map[string]scanRec)

	for _, r := range all {
		if underRoot(r.path, root) {
			scoped[r.path] = r
		}
	}

	return scoped, nil
}

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

// partitionChanged splits the stat results into files that must be
// hashed (new, or changed) and files whose existing records are reused
// without reading them: a file is unchanged when its statted size
// equals the recorded size and its statted mtime is not newer than the
// recorded mtime.
func partitionChanged(recs []fileRec,
	existing map[string]scanRec,
) ([]fileRec, []scanRec) {
	var (
		toHash    []fileRec
		unchanged []scanRec
	)

	for _, rec := range recs {
		old, ok := existing[rec.path]
		if ok && old.size == rec.size && old.mtime >= rec.mtime {
			unchanged = append(unchanged, old)

			continue
		}

		toHash = append(toHash, rec)
	}

	return toHash, unchanged
}

// collectDeletes returns the existing record paths that were not
// successfully processed this run: vanished files, plus paths that
// failed to stat or hash. The database keeps only records verified by
// the latest scan covering them. The result is sorted so the update
// pass is deterministic.
func collectDeletes(existing map[string]scanRec,
	unchanged, hashed []scanRec,
) []string {
	kept := make(map[string]bool, len(unchanged)+len(hashed))

	for _, r := range unchanged {
		kept[r.path] = true
	}

	for _, r := range hashed {
		kept[r.path] = true
	}

	var deletes []string

	for p := range existing {
		if !kept[p] {
			deletes = append(deletes, p)
		}
	}

	slices.Sort(deletes)

	return deletes
}

// applyPass writes the scan's changes to the database under an update
// progress display (one item per insertion, update, or deletion).
func applyPass(db *sql.DB, upserts []scanRec, deletes []string) error {
	prog := newProgress("update", int64(len(upserts)+len(deletes)))
	err := applyChanges(db, upserts, deletes, prog)

	prog.finish()

	return err
}

// 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 main goroutine: a
// regular file's stat record, or a warning when fail is set.
type walkEvent struct {
	rec  fileRec
	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
// directory on a different filesystem than root.
func walkPass(root string, oneFS bool, workers int) ([]fileRec, int) {
	prog := newProgress("walk", -1)

	recs, initial, errs := seedRoot(root, prog)
	jobs, subdirs, events := startWalkWorkers(workers, oneFS)

	dispatchDirs(initial, jobs, subdirs)

	for ev := range events {
		if ev.fail {
			errs++

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

			continue
		}

		recs = append(recs, ev.rec)

		prog.increment()
	}

	prog.finish()

	return recs, errs
}

// seedRoot turns the PATH operand into the walk's starting state: a
// regular-file operand becomes a record 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) {
	fi, err := os.Lstat(root)
	if err != nil {
		prog.warnf("walk %s: %v", root, err)

		return nil, nil, 1
	}

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

		dev, ok := deviceOfInfo(fi)

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

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

		events <- fileEvent(p, e)
	}

	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
// 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 main goroutine.
type hashResult struct {
	rec  fileRec
	head string
	tail string
	err  error
}

// startHashWorkers starts the hash worker pool over recs and returns
// the channel its results arrive on (one per record, in completion
// order).
func startHashWorkers(recs []fileRec, workers int) <-chan hashResult {
	jobs := make(chan fileRec, workQueueDepth)
	results := make(chan hashResult, workQueueDepth)

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

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

		close(jobs)
	}()

	return results
}

// hashPass hashes the first and last chunk bytes of every file in a
// worker pool and collects the resulting records on the main
// goroutine. Files that fail to open or read are warned about and
// dropped. Only new or changed files reach this pass.
func hashPass(recs []fileRec, workers int) ([]scanRec, int) {
	results := startHashWorkers(recs, workers)
	prog := newProgress("hash", int64(len(recs)))
	out := make([]scanRec, 0, len(recs))

	var errs int

	for range recs {
		r := <-results
		if r.err != nil {
			errs++

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

			continue
		}

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

		prog.increment()
	}

	prog.finish()

	return out, errs
}

// 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 by the stat pass.
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)

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

	t := sha256.Sum256(buf)

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