All checks were successful
check / check (push) Successful in 1m32s
TestSyncScanCancelledWalkKeepsRecords handed syncScan a context that was already cancelled. loadIndex is the first thing syncScan does, and its QueryContext fails on that context, so the scan returned before startWalk was ever called: no walk ran, no pool started, no write path was reachable, and all three of the test's assertions held for the wrong reason. The post-walk ctx.Err() guard, which is the highest-stakes line in the change, had no coverage at all — a panic in its body, or deleting it outright, left the suite green. Replace it with TestSyncScanCancelledMidWalkKeepsRecords, which cancels during the walk and so reaches the guard holding a genuinely partial census and a still-populated record index. The cancellation is driven by the scan's own progress rather than by a timer: walkClock is a context that cancels itself once its Done method has been consulted a set number of times, and since every blocking channel operation in the walk selects on Done — one consultation per event, a couple per directory, against the index load's fixed three — a threshold set to a quarter of the fixture's file count lands the cancellation deep inside the walk on every run. The census settles at around 380 of 2000 files, leaving some 1600 records that a complete-looking census would have handed to the update phase as deletions. The already-cancelled case is kept, renamed to what it actually tests and with its goroutine assertion dropped, since nothing that could leak is ever started. Direct tests cover the remaining cancellation branches of both pools: sendEvent abandoning a blocked send, walk workers dropping queued directories, a walk worker abandoning its subdirectory hand-off, dispatchDirs closing jobs on its way out, feedHashJobs doing the same, hashWorker dropping queued runs, and hashPhase leaving its result loop. Each is deterministic — the channels involved are unbuffered, unread or pre-filled, so the cancellation case is the only one that can be ready. Also correct two overstated claims. The hashLeakFiles comment described a mechanism that does not occur: the surplus is absorbed exactly by the two pool channels plus the workers in flight, so the feeder drains and exits, and what an abandoned pool leaves parked is the workers and the goroutine waiting on them. And the guard is defence in depth, not the sole barrier against data loss: the update phase's BeginTx fails on the same cancelled context before deleting anything today. The guard is what keeps that true once an interrupted scan is allowed to commit what it has.
997 lines
25 KiB
Go
997 lines
25 KiB
Go
package main
|
|
|
|
import (
|
|
"cmp"
|
|
"context"
|
|
"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. 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: 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. Errors are returned rather than exiting, so that the
|
|
// deferred close — which checkpoints the SQLite WAL — 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 func() { _ = db.Close() }()
|
|
|
|
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 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(ctx context.Context, db *sql.DB, roots []string,
|
|
workers int, oneFS bool,
|
|
) (scanStats, error) {
|
|
roots = pruneRoots(roots)
|
|
|
|
s := &scanState{db: db}
|
|
|
|
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
|
|
}
|
|
|
|
return s.st, s.updatePhase(ctx)
|
|
}
|
|
|
|
// 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 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++
|
|
}
|
|
|
|
// 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 the worker pool — one read
|
|
// per inode run, in inode order — 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 counts actual
|
|
// reads, so the bar shows a real ETA. A run that fails to hash is
|
|
// warned about and skipped; stale records for its paths, if any, are
|
|
// deleted by the update phase.
|
|
//
|
|
// 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) hashPhase(ctx context.Context, workers int) error {
|
|
runs := hashRuns(s.toHash)
|
|
s.toHash = nil
|
|
|
|
pool := startHashPool(ctx, runs, workers)
|
|
defer pool.stop()
|
|
|
|
prog := newProgress("hash", 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("hash %s: %v", r.run[0].path, r.err)
|
|
|
|
continue
|
|
}
|
|
|
|
err := s.recordRun(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,
|
|
path: rec.path,
|
|
})
|
|
}
|
|
|
|
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)
|
|
}
|
|
|
|
// 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():
|
|
}
|
|
}
|
|
|
|
// 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(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
|
|
}
|
|
|
|
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():
|
|
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
|
|
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. 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 one inode run's head/tail hashes (or the error
|
|
// that prevented hashing it) from the hash workers to the hash phase.
|
|
type hashResult struct {
|
|
run []fileRec
|
|
head string
|
|
tail 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 (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,
|
|
) *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) })
|
|
}
|
|
|
|
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 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,
|
|
) {
|
|
for run := range jobs {
|
|
if ctx.Err() != nil {
|
|
continue
|
|
}
|
|
|
|
head, tail, err := hashHeadTail(run[0].path, run[0].size)
|
|
|
|
select {
|
|
case results <- hashResult{
|
|
run: run, head: head, tail: tail, err: err,
|
|
}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// emptyHash is the lowercase-hex SHA-256 of the empty input: the head
|
|
// and tail hash of every zero-length file.
|
|
const emptyHash = "e3b0c44298fc1c149afbf4c8996fb924" +
|
|
"27ae41e4649b934ca495991b7852b855"
|
|
|
|
// 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. size is the
|
|
// value recorded when the file was statted; a zero-length file's
|
|
// hashes are constant, so it is never even opened.
|
|
func hashHeadTail(path string, size int64) (string, string, error) {
|
|
if size == 0 {
|
|
return 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() }()
|
|
|
|
n := min(int64(chunk), size)
|
|
|
|
buf := make([]byte, n)
|
|
|
|
_, 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
|
|
}
|