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 }