Compare commits
2 Commits
340bdbe39e
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| d4b43ebb30 | |||
| b62b4f297f |
158
README.md
158
README.md
@@ -50,11 +50,13 @@ completed scan.
|
||||
|
||||
Duplicate finders that hash entire files do not scale to the target
|
||||
environment: ~10 million files and ~150 TB on possibly slow or busy
|
||||
disks (a ZFS pool under resilver). Reading at most 2 KiB per file makes
|
||||
a full-filesystem sweep tractable, and the signatures are kept in a
|
||||
persistent database, so the expensive filesystem pass is incremental: a
|
||||
rescan re-hashes only files whose recorded mtime or size changed, and
|
||||
all analysis happens offline from the database alone. The end goal is
|
||||
disks (a ZFS pool under resilver). Reading at most 2 KiB per file — and
|
||||
only from files whose size at least one other file shares, since a
|
||||
size-unique file cannot be a duplicate — makes a full-filesystem sweep
|
||||
tractable, and the signatures are kept in a persistent database, so
|
||||
the expensive filesystem pass is incremental: a rescan re-hashes only
|
||||
files whose recorded mtime or size changed, and all analysis happens
|
||||
offline from the database alone. The end goal is
|
||||
not individual files but whole duplicated trees — duplicate
|
||||
extractions, duplicate downloads, copied project trees — which an
|
||||
operator can consider removing as a unit.
|
||||
@@ -70,10 +72,13 @@ Goals, in order:
|
||||
removing an entire subtree at once. File-level duplicate detection is
|
||||
the foundation; tree-level detection is built on top of it.
|
||||
2. **Never read full file contents.** At most 2 KiB is read per file
|
||||
(first and last 1024 bytes). Scale target: ~10 million files, ~150 TB
|
||||
(first and last 1024 bytes), and only files whose size at least
|
||||
one other file shares are read at all — a size-unique file cannot
|
||||
be a duplicate. Scale target: tens of millions of files, ~150 TB
|
||||
filesystem, possibly slow or busy disks (ZFS pool under resilver).
|
||||
Holding the full file list in memory is acceptable; reading file
|
||||
contents beyond 2 KiB per file is not.
|
||||
Holding one small record (path, size, mtime) per file in memory
|
||||
during a scan is acceptable; holding every file's hashes is not
|
||||
(they stay in the database).
|
||||
3. **Scan incrementally, analyze offline.** The expensive filesystem
|
||||
scan maintains a persistent database; an unchanged file is never
|
||||
read again on a rescan. All analysis (`report`, `trees`) works from
|
||||
@@ -132,11 +137,13 @@ All three subcommands operate on a single SQLite database file:
|
||||
- The database uses WAL journal mode and a busy timeout, so running a
|
||||
report while a cron `scan` is in progress is safe. The filesystem
|
||||
is authoritative; the database is an eventually-consistent
|
||||
reflection of it. The update pass applies changes in batched
|
||||
transactions (keeping the WAL small and letting concurrent reports
|
||||
observe progress), so a report may see a scan's changes partially
|
||||
applied, and a scan that dies partway leaves a valid database that
|
||||
the next scan converges toward the filesystem.
|
||||
reflection of it. Hashed records are committed in batched
|
||||
transactions while the scan is still running (keeping the WAL
|
||||
small and letting concurrent reports observe progress), so a
|
||||
report may see a scan's changes partially applied, and a scan
|
||||
that dies partway leaves a valid database holding everything
|
||||
hashed so far; the next scan skips those records and converges
|
||||
toward the filesystem.
|
||||
- Schema (`PRAGMA user_version` is the schema version, currently 1; a
|
||||
database with any other version is a fatal error):
|
||||
|
||||
@@ -152,7 +159,11 @@ All three subcommands operate on a single SQLite database file:
|
||||
|
||||
Paths are stored as BLOBs because Unix paths are raw bytes, not
|
||||
guaranteed UTF-8. `mtime` is used only for change detection; it is
|
||||
not part of the duplicate key.
|
||||
not part of the duplicate key. `head` and `tail` are empty strings
|
||||
when the file has never been hashed because its size was unique as
|
||||
of the last scan that covered it; such records still define the
|
||||
file for tree reconstruction but never participate in duplicate
|
||||
groups.
|
||||
|
||||
### `scan` mode
|
||||
|
||||
@@ -165,21 +176,32 @@ by absolute path, each operand is resolved to an absolute, lexically
|
||||
cleaned path (symlinks are not resolved) before walking, so results do
|
||||
not depend on the working directory. All operands belong to a single
|
||||
scan and are enumerated concurrently: every operand seeds the shared
|
||||
walk worker pool, and every pass runs once over the whole scan, so
|
||||
pass totals and ETAs are scan-wide. Overlapping operands (one
|
||||
containing another) are harmless — a file reached via multiple
|
||||
operands is deduplicated by path and produces one database record.
|
||||
walk worker pool. Overlapping operands are harmless — an operand that
|
||||
duplicates another or lies under another is dropped before walking,
|
||||
so every file is reached exactly once and produces one database
|
||||
record.
|
||||
|
||||
`scan` synchronizes the database with the filesystem state under the
|
||||
scanned operands:
|
||||
|
||||
- A file not yet in the database is hashed and inserted.
|
||||
- Only a file whose size at least one other file shares is ever
|
||||
read: a size-unique file cannot be a duplicate, so it is recorded
|
||||
without hashes (`head` and `tail` empty). The size census covers
|
||||
every file walked this scan plus every database record outside
|
||||
the scanned operands, so a possible duplicate of a separately
|
||||
scanned tree is still recognized.
|
||||
- A file not yet in the database is inserted: hashed when its size
|
||||
is shared, without hashes otherwise.
|
||||
- A file already in the database is **skipped without reading its
|
||||
contents** when its lstat size equals the recorded size and its
|
||||
lstat mtime is not newer than the recorded mtime. This is what
|
||||
makes a daily rescan cheap.
|
||||
makes a daily rescan cheap. Exception: an unchanged file whose
|
||||
record lacks hashes is hashed — and its record updated — once its
|
||||
size becomes shared, so hashing deferred by size-uniqueness
|
||||
happens as soon as it could matter.
|
||||
- A file whose mtime is newer than recorded, or whose size differs,
|
||||
is re-hashed and its record updated.
|
||||
is processed as if new: re-hashed, or recorded without hashes,
|
||||
per the shared-size rule.
|
||||
- A database record whose path lies under one of the scanned operands
|
||||
but was not successfully processed this run is deleted. This
|
||||
removes records for deleted files. It also removes records for
|
||||
@@ -190,29 +212,39 @@ scanned operands:
|
||||
disjoint trees can be scanned on different schedules into the same
|
||||
database.
|
||||
|
||||
`scan` runs **four sequential passes over the whole scan**.
|
||||
Parallelism lives strictly inside each pass; the passes themselves
|
||||
never overlap:
|
||||
`scan` runs **three sequential phases over the whole scan**.
|
||||
Parallelism lives inside each phase; batched database writes begin
|
||||
during the hash phase:
|
||||
|
||||
1. **walk** — enumerate the trees under all `PATH` operands
|
||||
concurrently with the worker pool: every operand seeds the shared
|
||||
queue, and each worker reads one directory at a time, collecting
|
||||
regular-file paths and handing discovered subdirectories back to
|
||||
the queue. Sequential directory enumeration is
|
||||
metadata-latency-bound and takes hours at tens of millions of
|
||||
files; per-directory parallelism is what makes the walk tractable
|
||||
on large or busy pools. Total unknown while running: show a live
|
||||
count, not a percentage.
|
||||
2. **stat** — `lstat` every collected path with the worker pool
|
||||
(per-file parallelism), recording size and mtime.
|
||||
3. **hash** — for each new or changed file (per the rules above), read
|
||||
the first `min(1024, size)` bytes and the last `min(1024, size)`
|
||||
bytes (the two reads overlap when `size < 2048`; for `size == 0`
|
||||
hash the empty input) and compute the SHA-256 of each. Unchanged
|
||||
files are not read and do not appear in this pass's total, which
|
||||
is therefore exact for meaningful progress and ETA.
|
||||
4. **update** — apply the scan's insertions, updates, and deletions
|
||||
to the database in batched transactions.
|
||||
1. **walk + stat** — enumerate the trees under all `PATH` operands
|
||||
concurrently with the walk worker pool: every operand seeds the
|
||||
shared queue, and each worker reads one directory at a time,
|
||||
handing discovered subdirectories back to the queue and running
|
||||
`lstat` on each regular file as it is discovered (while the
|
||||
directory's metadata is still hot). Sequential directory
|
||||
enumeration is metadata-latency-bound and takes hours at tens of
|
||||
millions of files; per-directory parallelism is what makes the
|
||||
walk tractable on large or busy pools. The walk builds the size
|
||||
census and resolves unchanged already-hashed files on the fly;
|
||||
every other file is carried to the hash phase as a (path, size,
|
||||
mtime) record.
|
||||
2. **hash** — with the census complete, each carried file's size
|
||||
decides its fate. Size-unique files are never read: new or
|
||||
changed ones are recorded without hashes in the update phase,
|
||||
unchanged unhashed ones simply keep their records. Every file
|
||||
with a shared size is hashed by the worker pool: read the first
|
||||
`min(1024, size)` bytes and the last `min(1024, size)` bytes
|
||||
(one read when `size <= 1024`, since the two windows coincide;
|
||||
for `size == 0` hash the empty input) and compute the SHA-256 of
|
||||
each. The phase total is exact, so progress and ETA are
|
||||
meaningful. Completed records are committed in batched
|
||||
transactions **while hashing runs**, so a scan interrupted after
|
||||
hours keeps everything hashed so far and the next scan resumes
|
||||
cheaply, skipping records already written.
|
||||
3. **update** — commit the final partial batch, the hash-less
|
||||
records for size-unique new and changed files, and the deletions
|
||||
for records the scan did not verify (vanished files, plus paths
|
||||
that failed to stat or hash).
|
||||
|
||||
Rules for the walk:
|
||||
|
||||
@@ -234,12 +266,14 @@ Rules for the walk:
|
||||
records (accepted: the database mirrors what the latest scan could
|
||||
actually verify).
|
||||
|
||||
Concurrency: the walk, stat, and hash passes each use a worker pool
|
||||
(`--workers`, default `runtime.NumCPU()`); the walk parallelizes
|
||||
across directories, stat and hash across files, so raising
|
||||
`--workers` can speed up metadata-bound passes on busy pools. The
|
||||
main goroutine owns database writes and progress rendering; progress
|
||||
display must never block the workers.
|
||||
Concurrency: the walk phase (which also stats files) and the hash
|
||||
phase each use a worker pool of `--workers` workers (default
|
||||
`runtime.NumCPU()`); the walk parallelizes across directories,
|
||||
hashing across files. Both phases are seek-bound on spinning disks,
|
||||
so raising `--workers` well past the core count can help on pools
|
||||
with many spindles. The main goroutine owns partitioning, database
|
||||
writes, and progress rendering; progress display must never block
|
||||
the workers.
|
||||
|
||||
`scan` writes nothing to stdout. The summary line on stderr reports the
|
||||
files seen this run broken down by disposition, plus skips:
|
||||
@@ -264,7 +298,11 @@ mounted.
|
||||
|
||||
Processing:
|
||||
|
||||
- Group records by the key `(size, head_hash, tail_hash)`.
|
||||
- Records without hashes (size-unique when last scanned) are
|
||||
excluded: their content is unknown, so they are never reported as
|
||||
duplicates.
|
||||
- Group the remaining records by the key
|
||||
`(size, head_hash, tail_hash)`.
|
||||
- Every group with two or more paths is a duplicate group.
|
||||
- Within each group, sort paths lexicographically (byte order). The
|
||||
first path is the group's `first`; every other path is a `dupe`.
|
||||
@@ -301,7 +339,10 @@ the paths in the records, split on `/`.
|
||||
Definitions:
|
||||
|
||||
- A file's **signature** is `(size, head_hash, tail_hash)` — mtime is
|
||||
informational and excluded.
|
||||
informational and excluded. An unhashed record (empty hashes) has
|
||||
unknown content: its signature is treated as unique to that file,
|
||||
so a tree containing an unhashed file never compares equal to any
|
||||
other tree.
|
||||
- A directory's **digest** is a SHA-256 Merkle digest computed
|
||||
bottom-up: serialize the directory's child entries — for a file
|
||||
child, its name and signature; for a subdirectory child, its name
|
||||
@@ -351,11 +392,15 @@ all dupe rows) in human units.
|
||||
|
||||
### Progress
|
||||
|
||||
Use the progress-bar library for all scan-pass progress; rendering in the
|
||||
Use the progress-bar library for all scan progress; rendering in the
|
||||
style of `pv` is the model. All progress goes to stderr.
|
||||
|
||||
Each scan pass gets its own bar. Required elements for the stat,
|
||||
hash, and update passes (known totals):
|
||||
Each phase gets its own display. The walk has no known total while
|
||||
running: show a live file count, rate, and elapsed time
|
||||
(spinner-style, no percentage or ETA). The hash and update phases
|
||||
have exact totals — only files that actually need hashing appear in
|
||||
the hash total, so its ETA is meaningful. Required elements for the
|
||||
bars with known totals:
|
||||
|
||||
- elapsed time
|
||||
- estimated time remaining
|
||||
@@ -365,12 +410,9 @@ hash, and update passes (known totals):
|
||||
Example shape (exact layout is flexible, content is not):
|
||||
|
||||
```
|
||||
hash: [1234567/9876543] 12% |████ | 8123 files/s elapsed 2:32 eta 17:54
|
||||
hash: [12345/98765] 12% |████ | 92 files/s elapsed 2:32 eta 17:54
|
||||
```
|
||||
|
||||
The walk pass has no known total: show a live file count and elapsed time
|
||||
(spinner-style, no percentage or ETA).
|
||||
|
||||
Additional requirements:
|
||||
|
||||
- When stderr is not a TTY, do not emit ANSI redraws: print a plain
|
||||
|
||||
38
db.go
38
db.go
@@ -238,6 +238,44 @@ func loadFileRows(db *sql.DB) ([]scanRec, error) {
|
||||
return recs, nil
|
||||
}
|
||||
|
||||
// loadFileMeta streams every record's path, size, mtime, and whether
|
||||
// it carries hashes to fn. Scan change detection needs no hash
|
||||
// values, and skipping the hash columns keeps the scan's in-memory
|
||||
// index small on multi-million-file databases.
|
||||
func loadFileMeta(db *sql.DB,
|
||||
fn func(path string, size, mtime int64, hashed bool),
|
||||
) error {
|
||||
rows, err := db.QueryContext(context.Background(),
|
||||
"SELECT path, size, mtime, head <> '' FROM files")
|
||||
if err != nil {
|
||||
return fmt.Errorf("read records: %w", err)
|
||||
}
|
||||
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
for rows.Next() {
|
||||
var (
|
||||
path []byte
|
||||
size, mtime int64
|
||||
hashed int64
|
||||
)
|
||||
|
||||
err = rows.Scan(&path, &size, &mtime, &hashed)
|
||||
if err != nil {
|
||||
return fmt.Errorf("read record: %w", err)
|
||||
}
|
||||
|
||||
fn(string(path), size, mtime, hashed != 0)
|
||||
}
|
||||
|
||||
err = rows.Err()
|
||||
if err != nil {
|
||||
return fmt.Errorf("read records: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// updateBatchSize is the number of record changes committed per
|
||||
// transaction during the update pass. The filesystem is authoritative
|
||||
// and the database an eventually-consistent reflection of it, so
|
||||
|
||||
2
main.go
2
main.go
@@ -69,7 +69,7 @@ func main() {
|
||||
},
|
||||
}
|
||||
scanCmd.Flags().IntVar(&scanWorkers, "workers", runtime.NumCPU(),
|
||||
"concurrent workers for the stat and hash passes")
|
||||
"concurrent workers for the walk and hash phases")
|
||||
scanCmd.Flags().BoolVarP(&scanOneFS, "one-file-system", "x", false,
|
||||
"do not cross filesystem boundaries")
|
||||
|
||||
|
||||
17
progress.go
17
progress.go
@@ -38,7 +38,10 @@ func stderrIsTTY() bool {
|
||||
// stderr is not a TTY it emits no ANSI redraws: it prints a plain
|
||||
// one-line update no more often than every plainInterval.
|
||||
//
|
||||
// All methods must be called from the main goroutine only.
|
||||
// All methods must be called from the main goroutine only. A nil
|
||||
// *progress is a valid no-display receiver: every method is a no-op,
|
||||
// so batched database flushes during the streaming pass can reuse the
|
||||
// update-pass helpers without rendering anything.
|
||||
type progress struct {
|
||||
label string
|
||||
total int64 // -1 when unknown (walk pass)
|
||||
@@ -82,6 +85,10 @@ func newProgress(label string, total int64) *progress {
|
||||
|
||||
// increment records one completed item and refreshes the display.
|
||||
func (p *progress) increment() {
|
||||
if p == nil {
|
||||
return
|
||||
}
|
||||
|
||||
p.count++
|
||||
if p.bar != nil {
|
||||
_ = p.bar.Add(1)
|
||||
@@ -97,6 +104,10 @@ func (p *progress) increment() {
|
||||
|
||||
// warnf prints a one-line warning to stderr without corrupting the bar.
|
||||
func (p *progress) warnf(format string, args ...any) {
|
||||
if p == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if p.bar != nil {
|
||||
_ = p.bar.Clear()
|
||||
}
|
||||
@@ -106,6 +117,10 @@ func (p *progress) warnf(format string, args ...any) {
|
||||
|
||||
// finish terminates the pass's display.
|
||||
func (p *progress) finish() {
|
||||
if p == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if p.bar != nil {
|
||||
_ = p.bar.Finish()
|
||||
|
||||
|
||||
@@ -105,6 +105,13 @@ func collectDupeGroups(recs []scanRec) []dupeGroup {
|
||||
groups := make(map[fileSig][]string)
|
||||
|
||||
for _, r := range recs {
|
||||
// A record without hashes (its size was unique when last
|
||||
// scanned) has unknown content and is never reported as a
|
||||
// duplicate.
|
||||
if r.head == "" {
|
||||
continue
|
||||
}
|
||||
|
||||
k := fileSig{size: r.size, head: r.head, tail: r.tail}
|
||||
groups[k] = append(groups[k], r.path)
|
||||
}
|
||||
|
||||
714
scan.go
714
scan.go
@@ -4,7 +4,6 @@ import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
@@ -18,25 +17,34 @@ import (
|
||||
// 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
|
||||
// workQueueDepth bounds the job and result channels feeding the walk
|
||||
// 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.
|
||||
// fileRec carries one statted file between the scan phases.
|
||||
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.
|
||||
// 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
|
||||
@@ -89,6 +97,33 @@ func resolveRoots(roots []string) []string {
|
||||
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 {
|
||||
@@ -99,60 +134,289 @@ type scanStats struct {
|
||||
skipped int
|
||||
}
|
||||
|
||||
// syncScan synchronizes the database with the filesystem under roots:
|
||||
// four scan-wide passes — walk, stat, hash, update — each parallel
|
||||
// internally, run strictly in sequence over all operands together.
|
||||
// Records outside the roots are never touched.
|
||||
// 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) {
|
||||
var st scanStats
|
||||
roots = pruneRoots(roots)
|
||||
|
||||
existing, err := loadScopedRows(db, roots)
|
||||
s := &scanState{db: db}
|
||||
|
||||
err := s.loadIndex(roots)
|
||||
if err != nil {
|
||||
return st, err
|
||||
return s.st, err
|
||||
}
|
||||
|
||||
paths, walkErrs := walkPass(roots, oneFS, workers)
|
||||
recs, statErrs := statPass(uniquePaths(paths), workers)
|
||||
toHash, unchanged := partitionChanged(recs, existing)
|
||||
hashed, hashErrs := hashPass(toHash, workers)
|
||||
changed, unhashed := s.walkPhase(startWalk(roots, oneFS, workers))
|
||||
|
||||
st.unchanged = len(unchanged)
|
||||
st.skipped = walkErrs + statErrs + hashErrs
|
||||
s.partition(changed, unhashed)
|
||||
|
||||
for _, r := range hashed {
|
||||
if _, ok := existing[r.path]; ok {
|
||||
st.updated++
|
||||
} else {
|
||||
st.added++
|
||||
}
|
||||
err = s.hashPhase(workers)
|
||||
if err != nil {
|
||||
return s.st, err
|
||||
}
|
||||
|
||||
deletes := collectDeletes(existing, unchanged, hashed)
|
||||
st.removed = len(deletes)
|
||||
|
||||
return st, applyPass(db, hashed, deletes)
|
||||
return s.st, s.updatePhase()
|
||||
}
|
||||
|
||||
// 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.
|
||||
func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
|
||||
all, err := loadFileRows(db)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// 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 {
|
||||
s.existing = make(map[string]fileMeta)
|
||||
|
||||
scoped := make(map[string]scanRec)
|
||||
return loadFileMeta(s.db,
|
||||
func(path string, size, mtime int64, hashed bool) {
|
||||
if underAnyRoot(path, roots) {
|
||||
s.existing[path] = fileMeta{
|
||||
size: size, mtime: mtime, hashed: hashed,
|
||||
}
|
||||
|
||||
for _, r := range all {
|
||||
if underAnyRoot(r.path, roots) {
|
||||
scoped[r.path] = r
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
return scoped, nil
|
||||
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
|
||||
@@ -167,26 +431,6 @@ func underAnyRoot(path string, roots []string) bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// underRoot reports whether path is root itself or lies under it. Both
|
||||
// must be absolute and lexically clean.
|
||||
func underRoot(path, root string) bool {
|
||||
@@ -202,75 +446,6 @@ func underRoot(path, root string) bool {
|
||||
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.
|
||||
@@ -280,87 +455,68 @@ type dirJob struct {
|
||||
rootDevOK bool
|
||||
}
|
||||
|
||||
// walkEvent is one walk result delivered to the main goroutine: a
|
||||
// regular-file path, or a warning when fail is set.
|
||||
// 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 {
|
||||
path string
|
||||
rec fileRec
|
||||
warn string
|
||||
fail bool
|
||||
}
|
||||
|
||||
// walkPass enumerates every regular file under all roots concurrently
|
||||
// with a per-directory worker pool; every operand seeds the shared
|
||||
// queue. 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, workers int) ([]string, int) {
|
||||
prog := newProgress("walk", -1)
|
||||
|
||||
var (
|
||||
paths []string
|
||||
initial []dirJob
|
||||
errs int
|
||||
)
|
||||
|
||||
for _, root := range roots {
|
||||
p, jobs, e := seedRoot(root, prog)
|
||||
paths = append(paths, p...)
|
||||
initial = append(initial, jobs...)
|
||||
errs += e
|
||||
}
|
||||
|
||||
// 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)
|
||||
|
||||
dispatchDirs(initial, jobs, subdirs)
|
||||
|
||||
for ev := range events {
|
||||
if ev.fail {
|
||||
errs++
|
||||
|
||||
prog.warnf("%s", ev.warn)
|
||||
|
||||
continue
|
||||
go func() {
|
||||
initial := make([]dirJob, 0, len(roots))
|
||||
for _, root := range roots {
|
||||
initial = append(initial, seedRoot(root, events)...)
|
||||
}
|
||||
|
||||
paths = append(paths, ev.path)
|
||||
dispatchDirs(initial, jobs, subdirs)
|
||||
}()
|
||||
|
||||
prog.increment()
|
||||
}
|
||||
|
||||
prog.finish()
|
||||
|
||||
return paths, errs
|
||||
return events
|
||||
}
|
||||
|
||||
// seedRoot turns the PATH operand into the walk's starting state: a
|
||||
// regular-file operand becomes a path directly, a directory operand
|
||||
// becomes the initial job, and a symlink or other non-regular operand
|
||||
// yields nothing (symlinks are never followed, including as
|
||||
// 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, prog *progress) ([]string, []dirJob, int) {
|
||||
func seedRoot(root string, events chan<- walkEvent) []dirJob {
|
||||
fi, err := os.Lstat(root)
|
||||
if err != nil {
|
||||
prog.warnf("walk %s: %v", root, err)
|
||||
events <- walkEvent{
|
||||
warn: fmt.Sprintf("walk %s: %v", root, err),
|
||||
fail: true,
|
||||
}
|
||||
|
||||
return nil, nil, 1
|
||||
return nil
|
||||
}
|
||||
|
||||
switch {
|
||||
case fi.IsDir():
|
||||
if filepath.Base(root) == ".zfs" {
|
||||
return nil, nil, 0
|
||||
return nil
|
||||
}
|
||||
|
||||
dev, ok := deviceOfInfo(fi)
|
||||
|
||||
return nil, []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}, 0
|
||||
return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
|
||||
case fi.Mode().IsRegular():
|
||||
prog.increment()
|
||||
events <- walkEvent{rec: fileRec{
|
||||
path: root,
|
||||
size: fi.Size(),
|
||||
mtime: fi.ModTime().Unix(),
|
||||
}}
|
||||
|
||||
return []string{root}, nil, 0
|
||||
return nil
|
||||
default:
|
||||
return nil, nil, 0
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
@@ -460,12 +616,39 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
|
||||
continue
|
||||
}
|
||||
|
||||
events <- walkEvent{path: p}
|
||||
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
|
||||
@@ -510,74 +693,8 @@ func deviceOfInfo(fi fs.FileInfo) (uint64, bool) {
|
||||
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.
|
||||
// prevented hashing it) from the hash workers to the hash phase.
|
||||
type hashResult struct {
|
||||
rec fileRec
|
||||
head string
|
||||
@@ -585,76 +702,28 @@ type hashResult struct {
|
||||
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)
|
||||
|
||||
// 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}
|
||||
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.
|
||||
// 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)
|
||||
@@ -676,11 +745,18 @@ func hashHeadTail(path string, size int64) (string, string, error) {
|
||||
|
||||
h := sha256.Sum256(buf)
|
||||
|
||||
if n > 0 {
|
||||
_, err = f.ReadAt(buf, size-n)
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
// 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)
|
||||
|
||||
215
scan_test.go
215
scan_test.go
@@ -110,14 +110,51 @@ func TestHashHeadTailErrors(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalkPass(t *testing.T) {
|
||||
// collectWalk runs a walk over roots and returns the emitted records
|
||||
// and the number of warning events.
|
||||
func collectWalk(t *testing.T, roots []string, oneFS bool,
|
||||
workers int,
|
||||
) ([]fileRec, int) {
|
||||
t.Helper()
|
||||
|
||||
var (
|
||||
recs []fileRec
|
||||
errs int
|
||||
)
|
||||
|
||||
for ev := range startWalk(roots, oneFS, workers) {
|
||||
if ev.fail {
|
||||
errs++
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
recs = append(recs, ev.rec)
|
||||
}
|
||||
|
||||
return recs, errs
|
||||
}
|
||||
|
||||
// walkedPaths returns the sorted paths of the walked records.
|
||||
func walkedPaths(recs []fileRec) []string {
|
||||
paths := make([]string, 0, len(recs))
|
||||
for _, r := range recs {
|
||||
paths = append(paths, r.path)
|
||||
}
|
||||
|
||||
slices.Sort(paths)
|
||||
|
||||
return paths
|
||||
}
|
||||
|
||||
func TestWalk(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
want := []string{
|
||||
writeFile(t, dir, "a.txt", []byte("a")),
|
||||
writeFile(t, dir, "sub/b.txt", []byte("b")),
|
||||
writeFile(t, dir, "sub/deeper/c.txt", []byte("c")),
|
||||
writeFile(t, dir, "sub/b.txt", []byte("bb")),
|
||||
writeFile(t, dir, "sub/deeper/c.txt", []byte("ccc")),
|
||||
}
|
||||
|
||||
slices.Sort(want)
|
||||
@@ -131,19 +168,29 @@ func TestWalkPass(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
paths, errs := walkPass([]string{dir}, false, 4)
|
||||
recs, errs := collectWalk(t, []string{dir}, false, 4)
|
||||
if errs != 0 {
|
||||
t.Fatalf("errs = %d, want 0", errs)
|
||||
}
|
||||
|
||||
slices.Sort(paths)
|
||||
if got := walkedPaths(recs); !slices.Equal(got, want) {
|
||||
t.Fatalf("paths = %q, want %q", got, want)
|
||||
}
|
||||
|
||||
if !slices.Equal(paths, want) {
|
||||
t.Fatalf("paths = %q, want %q", paths, want)
|
||||
// The walk stats each file as it is discovered: every record must
|
||||
// carry the real size and a plausible mtime.
|
||||
for _, r := range recs {
|
||||
if r.size < 1 || r.size > 3 {
|
||||
t.Errorf("%s: size = %d, want 1..3", r.path, r.size)
|
||||
}
|
||||
|
||||
if r.mtime <= 0 {
|
||||
t.Errorf("%s: mtime = %d, want positive", r.path, r.mtime)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalkPassDeepAndWide(t *testing.T) {
|
||||
func TestWalkDeepAndWide(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Exercise the dispatcher with more directories than workers and
|
||||
@@ -161,19 +208,17 @@ func TestWalkPassDeepAndWide(t *testing.T) {
|
||||
|
||||
slices.Sort(want)
|
||||
|
||||
paths, errs := walkPass([]string{dir}, false, 8)
|
||||
recs, errs := collectWalk(t, []string{dir}, false, 8)
|
||||
if errs != 0 {
|
||||
t.Fatalf("errs = %d, want 0", errs)
|
||||
}
|
||||
|
||||
slices.Sort(paths)
|
||||
|
||||
if !slices.Equal(paths, want) {
|
||||
t.Fatalf("walked %d paths, want %d", len(paths), len(want))
|
||||
if got := walkedPaths(recs); !slices.Equal(got, want) {
|
||||
t.Fatalf("walked %d paths, want %d", len(got), len(want))
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalkPassMultipleRoots(t *testing.T) {
|
||||
func TestWalkMultipleRoots(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
rootA := t.TempDir()
|
||||
@@ -188,19 +233,17 @@ func TestWalkPassMultipleRoots(t *testing.T) {
|
||||
|
||||
// Operands are enumerated concurrently by the shared pool; order
|
||||
// is unspecified.
|
||||
paths, errs := walkPass([]string{rootA, rootB}, false, 4)
|
||||
recs, errs := collectWalk(t, []string{rootA, rootB}, false, 4)
|
||||
if errs != 0 {
|
||||
t.Fatalf("errs = %d, want 0", errs)
|
||||
}
|
||||
|
||||
slices.Sort(paths)
|
||||
|
||||
if !slices.Equal(paths, want) {
|
||||
t.Fatalf("paths = %q, want %q", paths, want)
|
||||
if got := walkedPaths(recs); !slices.Equal(got, want) {
|
||||
t.Fatalf("paths = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
|
||||
func TestWalkFileAndSymlinkOperands(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
@@ -213,20 +256,20 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// A regular-file operand is emitted as itself.
|
||||
paths, errs := walkPass([]string{f}, false, 2)
|
||||
if errs != 0 || !slices.Equal(paths, []string{f}) {
|
||||
t.Fatalf("file operand: paths = %q, errs = %d", paths, errs)
|
||||
// A regular-file operand is emitted as itself, statted.
|
||||
recs, errs := collectWalk(t, []string{f}, false, 2)
|
||||
if errs != 0 || len(recs) != 1 || recs[0].path != f || recs[0].size != 4 {
|
||||
t.Fatalf("file operand: recs = %+v, errs = %d", recs, errs)
|
||||
}
|
||||
|
||||
// A symlink operand is not followed and yields nothing.
|
||||
paths, errs = walkPass([]string{link}, false, 2)
|
||||
if errs != 0 || len(paths) != 0 {
|
||||
t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs)
|
||||
recs, errs = collectWalk(t, []string{link}, false, 2)
|
||||
if errs != 0 || len(recs) != 0 {
|
||||
t.Fatalf("symlink operand: recs = %+v, errs = %d", recs, errs)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWalkPassOneFilesystemSameFS(t *testing.T) {
|
||||
func TestWalkOneFilesystemSameFS(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Everything in one filesystem: -x must not skip anything.
|
||||
@@ -236,41 +279,13 @@ func TestWalkPassOneFilesystemSameFS(t *testing.T) {
|
||||
writeFile(t, dir, "sub/deep/b", []byte("b")),
|
||||
}
|
||||
|
||||
paths, errs := walkPass([]string{dir}, true, 4)
|
||||
recs, errs := collectWalk(t, []string{dir}, true, 4)
|
||||
if errs != 0 {
|
||||
t.Fatalf("errs = %d, want 0", errs)
|
||||
}
|
||||
|
||||
slices.Sort(paths)
|
||||
|
||||
if !slices.Equal(paths, want) {
|
||||
t.Fatalf("paths = %q, want %q", paths, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStatPass(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
a := writeFile(t, dir, "a", pattern(1, 10))
|
||||
b := writeFile(t, dir, "b", pattern(2, 20))
|
||||
missing := filepath.Join(dir, "vanished")
|
||||
|
||||
recs, errs := statPass([]string{a, b, missing}, 2)
|
||||
if errs != 1 {
|
||||
t.Fatalf("errs = %d, want 1 for the vanished file", errs)
|
||||
}
|
||||
|
||||
slices.SortFunc(recs, func(x, y fileRec) int {
|
||||
return strings.Compare(x.path, y.path)
|
||||
})
|
||||
|
||||
if len(recs) != 2 || recs[0].size != 10 || recs[1].size != 20 {
|
||||
t.Fatalf("recs = %+v, want sizes 10 and 20", recs)
|
||||
}
|
||||
|
||||
if recs[0].mtime <= 0 || recs[1].mtime <= 0 {
|
||||
t.Fatalf("recs = %+v, want positive mtimes", recs)
|
||||
if got := walkedPaths(recs); !slices.Equal(got, want) {
|
||||
t.Fatalf("paths = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -662,14 +677,88 @@ func TestSyncScanOverlappingRoots(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestUniquePaths(t *testing.T) {
|
||||
func TestScanSkipsUniqueSizes(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
got := uniquePaths([]string{"/a", "/b", "/a", "/c", "/b"})
|
||||
dir := t.TempDir()
|
||||
db := openTestDB(t)
|
||||
a := writeFile(t, dir, "a.bin", pattern(1, 500))
|
||||
|
||||
want := []string{"/a", "/b", "/c"}
|
||||
writeFile(t, dir, "b.bin", pattern(2, 600))
|
||||
|
||||
// Neither size is shared, so neither file is read: both records
|
||||
// are written without hashes and no duplicates are reported.
|
||||
st := syncTree(t, db, dir)
|
||||
if st != (scanStats{added: 2}) {
|
||||
t.Fatalf("stats = %+v, want 2 added", st)
|
||||
}
|
||||
|
||||
recs := dbRecords(t, db)
|
||||
for _, r := range recs {
|
||||
if r.head != "" || r.tail != "" {
|
||||
t.Errorf("%s: head = %q tail = %q, want unhashed",
|
||||
r.path, r.head, r.tail)
|
||||
}
|
||||
}
|
||||
|
||||
if groups := collectDupeGroups(recs); len(groups) != 0 {
|
||||
t.Fatalf("groups = %+v, want none from unhashed records", groups)
|
||||
}
|
||||
|
||||
// A new same-size file makes 500 a shared size: the next scan
|
||||
// hashes both the new file and the previously unhashed unchanged
|
||||
// one, and they group as duplicates.
|
||||
c := writeFile(t, dir, "c.bin", pattern(1, 500))
|
||||
|
||||
st = syncTree(t, db, dir)
|
||||
if st != (scanStats{added: 1, updated: 1, unchanged: 1}) {
|
||||
t.Fatalf("rescan stats = %+v, want 1 added 1 updated 1 unchanged",
|
||||
st)
|
||||
}
|
||||
|
||||
groups := collectDupeGroups(dbRecords(t, db))
|
||||
if len(groups) != 1 {
|
||||
t.Fatalf("groups = %+v, want the a/c pair", groups)
|
||||
}
|
||||
|
||||
if want := []string{a, c}; !slices.Equal(groups[0].paths, want) {
|
||||
t.Fatalf("group paths = %q, want %q", groups[0].paths, want)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTreesUnhashedNeverEqual(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Two trees identical except for unhashed same-name, same-size
|
||||
// files (possible when the trees were scanned separately) must not
|
||||
// compare equal: unhashed content is unknown.
|
||||
shared := pattern(1, 100)
|
||||
recs := []scanRec{
|
||||
{path: "/x/t1/f1", size: 100, head: hexSum(shared), tail: hexSum(shared)},
|
||||
{path: "/x/t2/f1", size: 100, head: hexSum(shared), tail: hexSum(shared)},
|
||||
{path: "/x/t1/u", size: 50},
|
||||
{path: "/x/t2/u", size: 50},
|
||||
}
|
||||
|
||||
super, dirs := buildHierarchy(recs)
|
||||
super.compute()
|
||||
|
||||
if tg := collectTreeGroups(dirs, super); len(tg) != 0 {
|
||||
t.Fatalf("tree groups = %d, want 0 (unhashed files differ)",
|
||||
len(tg))
|
||||
}
|
||||
}
|
||||
|
||||
func TestPruneRoots(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// Duplicates and operands under other operands are dropped; /cc is
|
||||
// not under /c (sibling with a shared prefix).
|
||||
got := pruneRoots([]string{"/a/b", "/a", "/c", "/a", "/a/b/c", "/cc"})
|
||||
|
||||
want := []string{"/a", "/c", "/cc"}
|
||||
if !slices.Equal(got, want) {
|
||||
t.Fatalf("uniquePaths = %q, want %q", got, want)
|
||||
t.Fatalf("pruneRoots = %q, want %q", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
12
trees.go
12
trees.go
@@ -116,9 +116,17 @@ func buildHierarchy(recs []scanRec) (*treeNode, []*treeNode) {
|
||||
node.files = make(map[string]fileSig)
|
||||
}
|
||||
|
||||
node.files[comps[len(comps)-1]] = fileSig{
|
||||
size: r.size, head: r.head, tail: r.tail,
|
||||
sig := fileSig{size: r.size, head: r.head, tail: r.tail}
|
||||
|
||||
// An unhashed record (its size was unique when last scanned)
|
||||
// has unknown content: give it a signature no other file can
|
||||
// share, so trees containing it never compare equal. Real
|
||||
// heads are hex, so the NUL-prefixed form cannot collide.
|
||||
if sig.head == "" {
|
||||
sig.head = "unhashed\x00" + r.path
|
||||
}
|
||||
|
||||
node.files[comps[len(comps)-1]] = sig
|
||||
}
|
||||
|
||||
return super, allDirs
|
||||
|
||||
Reference in New Issue
Block a user