sfdupes/scan.go

package main

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"slices"
	"strings"
	"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 that stopped being a regular file
// between the walk and stat passes.
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:
// walk and stat everything, hash only new or changed files, and apply
// the resulting record insertions, updates, and deletions in a single
// transaction. Records outside the roots are never touched.
func syncScan(db *sql.DB, roots []string, workers int,
	oneFS bool,
) (scanStats, error) {
	var st scanStats

	existing, err := loadScopedRows(db, roots)
	if err != nil {
		return st, err
	}

	paths, walkErrs := walkPass(roots, oneFS)
	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

	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 st, applyPass(db, hashed, deletes)
}

// 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 by this scan.
func loadScopedRows(db *sql.DB, roots []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 underAnyRoot(r.path, roots) {
			scoped[r.path] = r
		}
	}

	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
}

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

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

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

// treeWalker carries the walk-pass state shared by all PATH operands.
type treeWalker struct {
	prog  *progress
	oneFS bool
	paths []string
	errs  int
}

// walkPass enumerates every regular file under each root operand in
// order. 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) ([]string, int) {
	w := &treeWalker{prog: newProgress("walk", -1), oneFS: oneFS}

	for _, root := range roots {
		w.walkRoot(root)
	}

	w.prog.finish()

	return w.paths, w.errs
}

// walkRoot walks a single PATH operand, appending regular-file paths.
func (w *treeWalker) walkRoot(root string) {
	rootDev, rootDevOK := deviceOf(root)

	walkErr := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
		if err != nil {
			w.errs++

			w.prog.warnf("walk %s: %v", p, err)

			if d != nil && d.IsDir() {
				return filepath.SkipDir
			}

			return nil
		}

		if d.IsDir() {
			return w.dirAction(p, d, rootDev, rootDevOK)
		}
		// Regular files only: skip symlinks, sockets, FIFOs, and
		// device nodes.
		if !d.Type().IsRegular() {
			return nil
		}

		w.paths = append(w.paths, p)

		w.prog.increment()

		return nil
	})
	if walkErr != nil {
		fatalf("walk %s: %v", root, walkErr)
	}
}

// dirAction decides whether the walk descends into directory p.
func (w *treeWalker) dirAction(p string, d fs.DirEntry, rootDev uint64, rootDevOK bool) error {
	// ZFS snapshot pseudo-dirs would list every file once per
	// snapshot; never descend.
	if d.Name() == ".zfs" {
		return filepath.SkipDir
	}

	if !w.oneFS || !rootDevOK {
		return nil
	}

	info, err := d.Info()
	if err != nil {
		w.errs++

		w.prog.warnf("walk %s: %v", p, err)

		return filepath.SkipDir
	}

	if dev, ok := deviceOfInfo(info); ok && dev != rootDev {
		return filepath.SkipDir
	}

	return nil
}

// deviceOf returns the filesystem device ID of path without following
// symlinks.
func deviceOf(path string) (uint64, bool) {
	fi, err := os.Lstat(path)
	if err != nil {
		return 0, false
	}

	return deviceOfInfo(fi)
}

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

// statPass lstats every collected path in a worker pool, recording size
// and mtime. Paths that fail to stat (or are no longer regular files)
// are warned about and dropped.
func statPass(paths []string, workers int) ([]fileRec, int) {
	type result struct {
		rec fileRec
		err error
	}

	jobs := make(chan string, workQueueDepth)
	results := make(chan result, workQueueDepth)

	for range workers {
		go func() {
			for p := range jobs {
				fi, err := os.Lstat(p)
				switch {
				case err != nil:
					results <- result{rec: fileRec{path: p}, err: err}
				case !fi.Mode().IsRegular():
					results <- result{
						rec: fileRec{path: p},
						err: errNotRegular,
					}
				default:
					results <- result{rec: fileRec{
						path:  p,
						size:  fi.Size(),
						mtime: fi.ModTime().Unix(),
					}}
				}
			}
		}()
	}

	go func() {
		for _, p := range paths {
			jobs <- p
		}

		close(jobs)
	}()

	prog := newProgress("stat", int64(len(paths)))

	var errs int

	recs := make([]fileRec, 0, len(paths))
	for range paths {
		r := <-results
		if r.err != nil {
			errs++

			prog.warnf("stat %s: %v", r.rec.path, r.err)
		} else {
			recs = append(recs, r.rec)
		}

		prog.increment()
	}

	prog.finish()

	return recs, errs
}

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