Compare commits
3 Commits
make-defau
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| d4b43ebb30 | |||
| b62b4f297f | |||
| 340bdbe39e |
158
README.md
158
README.md
@@ -50,11 +50,13 @@ completed scan.
|
|||||||
|
|
||||||
Duplicate finders that hash entire files do not scale to the target
|
Duplicate finders that hash entire files do not scale to the target
|
||||||
environment: ~10 million files and ~150 TB on possibly slow or busy
|
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
|
disks (a ZFS pool under resilver). Reading at most 2 KiB per file — and
|
||||||
a full-filesystem sweep tractable, and the signatures are kept in a
|
only from files whose size at least one other file shares, since a
|
||||||
persistent database, so the expensive filesystem pass is incremental: a
|
size-unique file cannot be a duplicate — makes a full-filesystem sweep
|
||||||
rescan re-hashes only files whose recorded mtime or size changed, and
|
tractable, and the signatures are kept in a persistent database, so
|
||||||
all analysis happens offline from the database alone. The end goal is
|
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
|
not individual files but whole duplicated trees — duplicate
|
||||||
extractions, duplicate downloads, copied project trees — which an
|
extractions, duplicate downloads, copied project trees — which an
|
||||||
operator can consider removing as a unit.
|
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
|
removing an entire subtree at once. File-level duplicate detection is
|
||||||
the foundation; tree-level detection is built on top of it.
|
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
|
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).
|
filesystem, possibly slow or busy disks (ZFS pool under resilver).
|
||||||
Holding the full file list in memory is acceptable; reading file
|
Holding one small record (path, size, mtime) per file in memory
|
||||||
contents beyond 2 KiB per file is not.
|
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
|
3. **Scan incrementally, analyze offline.** The expensive filesystem
|
||||||
scan maintains a persistent database; an unchanged file is never
|
scan maintains a persistent database; an unchanged file is never
|
||||||
read again on a rescan. All analysis (`report`, `trees`) works from
|
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
|
- 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
|
report while a cron `scan` is in progress is safe. The filesystem
|
||||||
is authoritative; the database is an eventually-consistent
|
is authoritative; the database is an eventually-consistent
|
||||||
reflection of it. The update pass applies changes in batched
|
reflection of it. Hashed records are committed in batched
|
||||||
transactions (keeping the WAL small and letting concurrent reports
|
transactions while the scan is still running (keeping the WAL
|
||||||
observe progress), so a report may see a scan's changes partially
|
small and letting concurrent reports observe progress), so a
|
||||||
applied, and a scan that dies partway leaves a valid database that
|
report may see a scan's changes partially applied, and a scan
|
||||||
the next scan converges toward the filesystem.
|
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
|
- Schema (`PRAGMA user_version` is the schema version, currently 1; a
|
||||||
database with any other version is a fatal error):
|
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
|
Paths are stored as BLOBs because Unix paths are raw bytes, not
|
||||||
guaranteed UTF-8. `mtime` is used only for change detection; it is
|
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
|
### `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
|
cleaned path (symlinks are not resolved) before walking, so results do
|
||||||
not depend on the working directory. All operands belong to a single
|
not depend on the working directory. All operands belong to a single
|
||||||
scan and are enumerated concurrently: every operand seeds the shared
|
scan and are enumerated concurrently: every operand seeds the shared
|
||||||
walk worker pool, and every pass runs once over the whole scan, so
|
walk worker pool. Overlapping operands are harmless — an operand that
|
||||||
pass totals and ETAs are scan-wide. Overlapping operands (one
|
duplicates another or lies under another is dropped before walking,
|
||||||
containing another) are harmless — a file reached via multiple
|
so every file is reached exactly once and produces one database
|
||||||
operands is deduplicated by path and produces one database record.
|
record.
|
||||||
|
|
||||||
`scan` synchronizes the database with the filesystem state under the
|
`scan` synchronizes the database with the filesystem state under the
|
||||||
scanned operands:
|
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
|
- A file already in the database is **skipped without reading its
|
||||||
contents** when its lstat size equals the recorded size and its
|
contents** when its lstat size equals the recorded size and its
|
||||||
lstat mtime is not newer than the recorded mtime. This is what
|
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,
|
- 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
|
- A database record whose path lies under one of the scanned operands
|
||||||
but was not successfully processed this run is deleted. This
|
but was not successfully processed this run is deleted. This
|
||||||
removes records for deleted files. It also removes records for
|
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
|
disjoint trees can be scanned on different schedules into the same
|
||||||
database.
|
database.
|
||||||
|
|
||||||
`scan` runs **four sequential passes over the whole scan**.
|
`scan` runs **three sequential phases over the whole scan**.
|
||||||
Parallelism lives strictly inside each pass; the passes themselves
|
Parallelism lives inside each phase; batched database writes begin
|
||||||
never overlap:
|
during the hash phase:
|
||||||
|
|
||||||
1. **walk** — enumerate the trees under all `PATH` operands
|
1. **walk + stat** — enumerate the trees under all `PATH` operands
|
||||||
concurrently with the worker pool: every operand seeds the shared
|
concurrently with the walk worker pool: every operand seeds the
|
||||||
queue, and each worker reads one directory at a time, collecting
|
shared queue, and each worker reads one directory at a time,
|
||||||
regular-file paths and handing discovered subdirectories back to
|
handing discovered subdirectories back to the queue and running
|
||||||
the queue. Sequential directory enumeration is
|
`lstat` on each regular file as it is discovered (while the
|
||||||
metadata-latency-bound and takes hours at tens of millions of
|
directory's metadata is still hot). Sequential directory
|
||||||
files; per-directory parallelism is what makes the walk tractable
|
enumeration is metadata-latency-bound and takes hours at tens of
|
||||||
on large or busy pools. Total unknown while running: show a live
|
millions of files; per-directory parallelism is what makes the
|
||||||
count, not a percentage.
|
walk tractable on large or busy pools. The walk builds the size
|
||||||
2. **stat** — `lstat` every collected path with the worker pool
|
census and resolves unchanged already-hashed files on the fly;
|
||||||
(per-file parallelism), recording size and mtime.
|
every other file is carried to the hash phase as a (path, size,
|
||||||
3. **hash** — for each new or changed file (per the rules above), read
|
mtime) record.
|
||||||
the first `min(1024, size)` bytes and the last `min(1024, size)`
|
2. **hash** — with the census complete, each carried file's size
|
||||||
bytes (the two reads overlap when `size < 2048`; for `size == 0`
|
decides its fate. Size-unique files are never read: new or
|
||||||
hash the empty input) and compute the SHA-256 of each. Unchanged
|
changed ones are recorded without hashes in the update phase,
|
||||||
files are not read and do not appear in this pass's total, which
|
unchanged unhashed ones simply keep their records. Every file
|
||||||
is therefore exact for meaningful progress and ETA.
|
with a shared size is hashed by the worker pool: read the first
|
||||||
4. **update** — apply the scan's insertions, updates, and deletions
|
`min(1024, size)` bytes and the last `min(1024, size)` bytes
|
||||||
to the database in batched transactions.
|
(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:
|
Rules for the walk:
|
||||||
|
|
||||||
@@ -234,12 +266,14 @@ Rules for the walk:
|
|||||||
records (accepted: the database mirrors what the latest scan could
|
records (accepted: the database mirrors what the latest scan could
|
||||||
actually verify).
|
actually verify).
|
||||||
|
|
||||||
Concurrency: the walk, stat, and hash passes each use a worker pool
|
Concurrency: the walk phase (which also stats files) and the hash
|
||||||
(`--workers`, default `runtime.NumCPU()`); the walk parallelizes
|
phase each use a worker pool of `--workers` workers (default
|
||||||
across directories, stat and hash across files, so raising
|
`runtime.NumCPU()`); the walk parallelizes across directories,
|
||||||
`--workers` can speed up metadata-bound passes on busy pools. The
|
hashing across files. Both phases are seek-bound on spinning disks,
|
||||||
main goroutine owns database writes and progress rendering; progress
|
so raising `--workers` well past the core count can help on pools
|
||||||
display must never block the workers.
|
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
|
`scan` writes nothing to stdout. The summary line on stderr reports the
|
||||||
files seen this run broken down by disposition, plus skips:
|
files seen this run broken down by disposition, plus skips:
|
||||||
@@ -264,7 +298,11 @@ mounted.
|
|||||||
|
|
||||||
Processing:
|
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.
|
- Every group with two or more paths is a duplicate group.
|
||||||
- Within each group, sort paths lexicographically (byte order). The
|
- Within each group, sort paths lexicographically (byte order). The
|
||||||
first path is the group's `first`; every other path is a `dupe`.
|
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:
|
Definitions:
|
||||||
|
|
||||||
- A file's **signature** is `(size, head_hash, tail_hash)` — mtime is
|
- 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
|
- A directory's **digest** is a SHA-256 Merkle digest computed
|
||||||
bottom-up: serialize the directory's child entries — for a file
|
bottom-up: serialize the directory's child entries — for a file
|
||||||
child, its name and signature; for a subdirectory child, its name
|
child, its name and signature; for a subdirectory child, its name
|
||||||
@@ -351,11 +392,15 @@ all dupe rows) in human units.
|
|||||||
|
|
||||||
### Progress
|
### 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.
|
style of `pv` is the model. All progress goes to stderr.
|
||||||
|
|
||||||
Each scan pass gets its own bar. Required elements for the stat,
|
Each phase gets its own display. The walk has no known total while
|
||||||
hash, and update passes (known totals):
|
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
|
- elapsed time
|
||||||
- estimated time remaining
|
- estimated time remaining
|
||||||
@@ -365,12 +410,9 @@ hash, and update passes (known totals):
|
|||||||
Example shape (exact layout is flexible, content is not):
|
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:
|
Additional requirements:
|
||||||
|
|
||||||
- When stderr is not a TTY, do not emit ANSI redraws: print a plain
|
- 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
|
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
|
// updateBatchSize is the number of record changes committed per
|
||||||
// transaction during the update pass. The filesystem is authoritative
|
// transaction during the update pass. The filesystem is authoritative
|
||||||
// and the database an eventually-consistent reflection of it, so
|
// 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(),
|
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,
|
scanCmd.Flags().BoolVarP(&scanOneFS, "one-file-system", "x", false,
|
||||||
"do not cross filesystem boundaries")
|
"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
|
// stderr is not a TTY it emits no ANSI redraws: it prints a plain
|
||||||
// one-line update no more often than every plainInterval.
|
// 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 {
|
type progress struct {
|
||||||
label string
|
label string
|
||||||
total int64 // -1 when unknown (walk pass)
|
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.
|
// increment records one completed item and refreshes the display.
|
||||||
func (p *progress) increment() {
|
func (p *progress) increment() {
|
||||||
|
if p == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
p.count++
|
p.count++
|
||||||
if p.bar != nil {
|
if p.bar != nil {
|
||||||
_ = p.bar.Add(1)
|
_ = p.bar.Add(1)
|
||||||
@@ -97,6 +104,10 @@ func (p *progress) increment() {
|
|||||||
|
|
||||||
// warnf prints a one-line warning to stderr without corrupting the bar.
|
// warnf prints a one-line warning to stderr without corrupting the bar.
|
||||||
func (p *progress) warnf(format string, args ...any) {
|
func (p *progress) warnf(format string, args ...any) {
|
||||||
|
if p == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
if p.bar != nil {
|
if p.bar != nil {
|
||||||
_ = p.bar.Clear()
|
_ = p.bar.Clear()
|
||||||
}
|
}
|
||||||
@@ -106,6 +117,10 @@ func (p *progress) warnf(format string, args ...any) {
|
|||||||
|
|
||||||
// finish terminates the pass's display.
|
// finish terminates the pass's display.
|
||||||
func (p *progress) finish() {
|
func (p *progress) finish() {
|
||||||
|
if p == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
if p.bar != nil {
|
if p.bar != nil {
|
||||||
_ = p.bar.Finish()
|
_ = p.bar.Finish()
|
||||||
|
|
||||||
|
|||||||
@@ -105,6 +105,13 @@ func collectDupeGroups(recs []scanRec) []dupeGroup {
|
|||||||
groups := make(map[fileSig][]string)
|
groups := make(map[fileSig][]string)
|
||||||
|
|
||||||
for _, r := range recs {
|
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}
|
k := fileSig{size: r.size, head: r.head, tail: r.tail}
|
||||||
groups[k] = append(groups[k], r.path)
|
groups[k] = append(groups[k], r.path)
|
||||||
}
|
}
|
||||||
|
|||||||
714
scan.go
714
scan.go
@@ -4,7 +4,6 @@ import (
|
|||||||
"crypto/sha256"
|
"crypto/sha256"
|
||||||
"database/sql"
|
"database/sql"
|
||||||
"encoding/hex"
|
"encoding/hex"
|
||||||
"errors"
|
|
||||||
"fmt"
|
"fmt"
|
||||||
"io/fs"
|
"io/fs"
|
||||||
"os"
|
"os"
|
||||||
@@ -18,25 +17,34 @@ import (
|
|||||||
// chunk is the number of bytes hashed from each end of a file.
|
// chunk is the number of bytes hashed from each end of a file.
|
||||||
const chunk = 1024
|
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.
|
// and hash worker pools.
|
||||||
const workQueueDepth = 1024
|
const workQueueDepth = 1024
|
||||||
|
|
||||||
// errNotRegular reports a path whose type changed between the
|
// fileRec carries one statted file between the scan phases.
|
||||||
// directory read and its lstat.
|
|
||||||
var errNotRegular = errors.New("no longer a regular file")
|
|
||||||
|
|
||||||
// fileRec carries one file between the stat and hash passes.
|
|
||||||
type fileRec struct {
|
type fileRec struct {
|
||||||
path string
|
path string
|
||||||
size int64
|
size int64
|
||||||
mtime int64
|
mtime int64
|
||||||
}
|
}
|
||||||
|
|
||||||
// runScan implements the scan subcommand: four sequential passes
|
// fileMeta is the in-memory index entry for one existing database
|
||||||
// (walk, stat, hash, update) that synchronize the persistent database
|
// record: just enough for change detection, plus whether the record
|
||||||
// with the filesystem state under the PATH operands. Flag parsing and
|
// carries hashes. Hashes stay on disk; at tens of millions of records
|
||||||
// the at-least-one-operand check are done by cobra.
|
// 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) {
|
func runScan(roots []string, workers int, oneFS bool) {
|
||||||
if workers < 1 {
|
if workers < 1 {
|
||||||
workers = 1
|
workers = 1
|
||||||
@@ -89,6 +97,33 @@ func resolveRoots(roots []string) []string {
|
|||||||
return abs
|
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
|
// scanStats summarizes one scan's database synchronization for the
|
||||||
// final stderr summary.
|
// final stderr summary.
|
||||||
type scanStats struct {
|
type scanStats struct {
|
||||||
@@ -99,60 +134,289 @@ type scanStats struct {
|
|||||||
skipped int
|
skipped int
|
||||||
}
|
}
|
||||||
|
|
||||||
// syncScan synchronizes the database with the filesystem under roots:
|
// scanState carries one scan's evolving state across its phases.
|
||||||
// four scan-wide passes — walk, stat, hash, update — each parallel
|
// Entries consumed from existing mark files verified this run;
|
||||||
// internally, run strictly in sequence over all operands together.
|
// whatever remains after the walk and hash phases is deleted by the
|
||||||
// Records outside the roots are never touched.
|
// 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,
|
func syncScan(db *sql.DB, roots []string, workers int,
|
||||||
oneFS bool,
|
oneFS bool,
|
||||||
) (scanStats, error) {
|
) (scanStats, error) {
|
||||||
var st scanStats
|
roots = pruneRoots(roots)
|
||||||
|
|
||||||
existing, err := loadScopedRows(db, roots)
|
s := &scanState{db: db}
|
||||||
|
|
||||||
|
err := s.loadIndex(roots)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return st, err
|
return s.st, err
|
||||||
}
|
}
|
||||||
|
|
||||||
paths, walkErrs := walkPass(roots, oneFS, workers)
|
changed, unhashed := s.walkPhase(startWalk(roots, oneFS, workers))
|
||||||
recs, statErrs := statPass(uniquePaths(paths), workers)
|
|
||||||
toHash, unchanged := partitionChanged(recs, existing)
|
|
||||||
hashed, hashErrs := hashPass(toHash, workers)
|
|
||||||
|
|
||||||
st.unchanged = len(unchanged)
|
s.partition(changed, unhashed)
|
||||||
st.skipped = walkErrs + statErrs + hashErrs
|
|
||||||
|
|
||||||
for _, r := range hashed {
|
err = s.hashPhase(workers)
|
||||||
if _, ok := existing[r.path]; ok {
|
if err != nil {
|
||||||
st.updated++
|
return s.st, err
|
||||||
} else {
|
|
||||||
st.added++
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
deletes := collectDeletes(existing, unchanged, hashed)
|
return s.st, s.updatePhase()
|
||||||
st.removed = len(deletes)
|
|
||||||
|
|
||||||
return st, applyPass(db, hashed, deletes)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// loadScopedRows loads the database records whose paths lie under any
|
// loadIndex indexes the database records under the scan roots for
|
||||||
// of the scan roots, keyed by path. Records outside the roots belong
|
// change detection and collects the sizes of every record outside
|
||||||
// to other trees and are left untouched.
|
// them: out-of-scope records join the size census so a scanned file
|
||||||
func loadScopedRows(db *sql.DB, roots []string) (map[string]scanRec, error) {
|
// can be recognized as a possible duplicate of a tree scanned
|
||||||
all, err := loadFileRows(db)
|
// separately into the same database.
|
||||||
if err != nil {
|
func (s *scanState) loadIndex(roots []string) error {
|
||||||
return nil, err
|
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 {
|
return
|
||||||
if underAnyRoot(r.path, roots) {
|
}
|
||||||
scoped[r.path] = r
|
|
||||||
|
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
|
// 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
|
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
|
// underRoot reports whether path is root itself or lies under it. Both
|
||||||
// must be absolute and lexically clean.
|
// must be absolute and lexically clean.
|
||||||
func underRoot(path, root string) bool {
|
func underRoot(path, root string) bool {
|
||||||
@@ -202,75 +446,6 @@ func underRoot(path, root string) bool {
|
|||||||
return strings.HasPrefix(path, prefix)
|
return strings.HasPrefix(path, prefix)
|
||||||
}
|
}
|
||||||
|
|
||||||
// partitionChanged splits the stat results into files that must be
|
|
||||||
// hashed (new, or changed) and files whose existing records are reused
|
|
||||||
// without reading them: a file is unchanged when its statted size
|
|
||||||
// equals the recorded size and its statted mtime is not newer than the
|
|
||||||
// recorded mtime.
|
|
||||||
func partitionChanged(recs []fileRec,
|
|
||||||
existing map[string]scanRec,
|
|
||||||
) ([]fileRec, []scanRec) {
|
|
||||||
var (
|
|
||||||
toHash []fileRec
|
|
||||||
unchanged []scanRec
|
|
||||||
)
|
|
||||||
|
|
||||||
for _, rec := range recs {
|
|
||||||
old, ok := existing[rec.path]
|
|
||||||
if ok && old.size == rec.size && old.mtime >= rec.mtime {
|
|
||||||
unchanged = append(unchanged, old)
|
|
||||||
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
|
|
||||||
toHash = append(toHash, rec)
|
|
||||||
}
|
|
||||||
|
|
||||||
return toHash, unchanged
|
|
||||||
}
|
|
||||||
|
|
||||||
// collectDeletes returns the existing record paths that were not
|
|
||||||
// successfully processed this run: vanished files, plus paths that
|
|
||||||
// failed to stat or hash. The database keeps only records verified by
|
|
||||||
// the latest scan covering them. The result is sorted so the update
|
|
||||||
// pass is deterministic.
|
|
||||||
func collectDeletes(existing map[string]scanRec,
|
|
||||||
unchanged, hashed []scanRec,
|
|
||||||
) []string {
|
|
||||||
kept := make(map[string]bool, len(unchanged)+len(hashed))
|
|
||||||
|
|
||||||
for _, r := range unchanged {
|
|
||||||
kept[r.path] = true
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, r := range hashed {
|
|
||||||
kept[r.path] = true
|
|
||||||
}
|
|
||||||
|
|
||||||
var deletes []string
|
|
||||||
|
|
||||||
for p := range existing {
|
|
||||||
if !kept[p] {
|
|
||||||
deletes = append(deletes, p)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
slices.Sort(deletes)
|
|
||||||
|
|
||||||
return deletes
|
|
||||||
}
|
|
||||||
|
|
||||||
// applyPass writes the scan's changes to the database under an update
|
|
||||||
// progress display (one item per insertion, update, or deletion).
|
|
||||||
func applyPass(db *sql.DB, upserts []scanRec, deletes []string) error {
|
|
||||||
prog := newProgress("update", int64(len(upserts)+len(deletes)))
|
|
||||||
err := applyChanges(db, upserts, deletes, prog)
|
|
||||||
|
|
||||||
prog.finish()
|
|
||||||
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
|
|
||||||
// dirJob is one directory awaiting traversal by the walk workers. It
|
// dirJob is one directory awaiting traversal by the walk workers. It
|
||||||
// carries its operand's filesystem device so -x can stop at
|
// carries its operand's filesystem device so -x can stop at
|
||||||
// filesystem boundaries.
|
// filesystem boundaries.
|
||||||
@@ -280,87 +455,68 @@ type dirJob struct {
|
|||||||
rootDevOK bool
|
rootDevOK bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// walkEvent is one walk result delivered to the main goroutine: a
|
// walkEvent is one walk result delivered to the walk phase: a regular
|
||||||
// regular-file path, or a warning when fail is set.
|
// file's statted record, or a warning when fail is set.
|
||||||
type walkEvent struct {
|
type walkEvent struct {
|
||||||
path string
|
rec fileRec
|
||||||
warn string
|
warn string
|
||||||
fail bool
|
fail bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// walkPass enumerates every regular file under all roots concurrently
|
// startWalk seeds every root into the shared walk worker pool and
|
||||||
// with a per-directory worker pool; every operand seeds the shared
|
// returns the event stream: one record per regular file, one warning
|
||||||
// queue. It never follows symlinks, never descends into directories
|
// event per per-path error. The channel is closed when the walk
|
||||||
// named .zfs, and warns and continues on any per-path error. With
|
// completes.
|
||||||
// oneFS set it never descends into a directory on a different
|
func startWalk(roots []string, oneFS bool, workers int) <-chan walkEvent {
|
||||||
// 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
|
|
||||||
}
|
|
||||||
|
|
||||||
jobs, subdirs, events := startWalkWorkers(workers, oneFS)
|
jobs, subdirs, events := startWalkWorkers(workers, oneFS)
|
||||||
|
|
||||||
dispatchDirs(initial, jobs, subdirs)
|
go func() {
|
||||||
|
initial := make([]dirJob, 0, len(roots))
|
||||||
for ev := range events {
|
for _, root := range roots {
|
||||||
if ev.fail {
|
initial = append(initial, seedRoot(root, events)...)
|
||||||
errs++
|
|
||||||
|
|
||||||
prog.warnf("%s", ev.warn)
|
|
||||||
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
|
|
||||||
paths = append(paths, ev.path)
|
dispatchDirs(initial, jobs, subdirs)
|
||||||
|
}()
|
||||||
|
|
||||||
prog.increment()
|
return events
|
||||||
}
|
|
||||||
|
|
||||||
prog.finish()
|
|
||||||
|
|
||||||
return paths, errs
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// seedRoot turns the PATH operand into the walk's starting state: a
|
// seedRoot turns one PATH operand into the walk's starting state: a
|
||||||
// regular-file operand becomes a path directly, a directory operand
|
// regular-file operand is statted and emitted directly, a directory
|
||||||
// becomes the initial job, and a symlink or other non-regular operand
|
// operand becomes an initial job, and a symlink or other non-regular
|
||||||
// yields nothing (symlinks are never followed, including as
|
// operand yields nothing (symlinks are never followed, including as
|
||||||
// operands).
|
// operands).
|
||||||
func seedRoot(root string, prog *progress) ([]string, []dirJob, int) {
|
func seedRoot(root string, events chan<- walkEvent) []dirJob {
|
||||||
fi, err := os.Lstat(root)
|
fi, err := os.Lstat(root)
|
||||||
if err != nil {
|
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 {
|
switch {
|
||||||
case fi.IsDir():
|
case fi.IsDir():
|
||||||
if filepath.Base(root) == ".zfs" {
|
if filepath.Base(root) == ".zfs" {
|
||||||
return nil, nil, 0
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
dev, ok := deviceOfInfo(fi)
|
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():
|
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:
|
default:
|
||||||
return nil, nil, 0
|
return nil
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -460,12 +616,39 @@ func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
events <- walkEvent{path: p}
|
emitFile(p, e, events)
|
||||||
}
|
}
|
||||||
|
|
||||||
return subs
|
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
|
// subdirJob applies the descent rules to directory p: never enter
|
||||||
// .zfs (ZFS snapshot pseudo-dirs would list every file once per
|
// .zfs (ZFS snapshot pseudo-dirs would list every file once per
|
||||||
// snapshot), and with -x never enter a directory on a different
|
// 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
|
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
|
// 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 {
|
type hashResult struct {
|
||||||
rec fileRec
|
rec fileRec
|
||||||
head string
|
head string
|
||||||
@@ -585,76 +702,28 @@ type hashResult struct {
|
|||||||
err error
|
err error
|
||||||
}
|
}
|
||||||
|
|
||||||
// startHashWorkers starts the hash worker pool over recs and returns
|
// startHashWorkers starts the hash worker pool: workers read jobs,
|
||||||
// the channel its results arrive on (one per record, in completion
|
// write one result per record, and exit when jobs is closed.
|
||||||
// order).
|
func startHashWorkers(jobs <-chan fileRec, results chan<- hashResult,
|
||||||
func startHashWorkers(recs []fileRec, workers int) <-chan hashResult {
|
workers int,
|
||||||
jobs := make(chan fileRec, workQueueDepth)
|
) {
|
||||||
results := make(chan hashResult, workQueueDepth)
|
|
||||||
|
|
||||||
for range workers {
|
for range workers {
|
||||||
go func() {
|
go func() {
|
||||||
for rec := range jobs {
|
for rec := range jobs {
|
||||||
head, tail, err := hashHeadTail(rec.path, rec.size)
|
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
|
// hashHeadTail returns the lowercase-hex SHA-256 of the first
|
||||||
// min(chunk, size) bytes and of the last min(chunk, size) bytes of the
|
// 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
|
// 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
|
// 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) {
|
func hashHeadTail(path string, size int64) (string, string, error) {
|
||||||
//nolint:gosec // hashing operator-supplied paths is the tool's purpose
|
//nolint:gosec // hashing operator-supplied paths is the tool's purpose
|
||||||
f, err := os.Open(path)
|
f, err := os.Open(path)
|
||||||
@@ -676,11 +745,18 @@ func hashHeadTail(path string, size int64) (string, string, error) {
|
|||||||
|
|
||||||
h := sha256.Sum256(buf)
|
h := sha256.Sum256(buf)
|
||||||
|
|
||||||
if n > 0 {
|
// When the whole file fits in one chunk the tail window is exactly
|
||||||
_, err = f.ReadAt(buf, size-n)
|
// the bytes just read: reuse the head hash instead of issuing a
|
||||||
if err != nil {
|
// second read for every small file.
|
||||||
return "", "", err
|
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)
|
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()
|
t.Parallel()
|
||||||
|
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
want := []string{
|
want := []string{
|
||||||
writeFile(t, dir, "a.txt", []byte("a")),
|
writeFile(t, dir, "a.txt", []byte("a")),
|
||||||
writeFile(t, dir, "sub/b.txt", []byte("b")),
|
writeFile(t, dir, "sub/b.txt", []byte("bb")),
|
||||||
writeFile(t, dir, "sub/deeper/c.txt", []byte("c")),
|
writeFile(t, dir, "sub/deeper/c.txt", []byte("ccc")),
|
||||||
}
|
}
|
||||||
|
|
||||||
slices.Sort(want)
|
slices.Sort(want)
|
||||||
@@ -131,19 +168,29 @@ func TestWalkPass(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
paths, errs := walkPass([]string{dir}, false, 4)
|
recs, errs := collectWalk(t, []string{dir}, false, 4)
|
||||||
if errs != 0 {
|
if errs != 0 {
|
||||||
t.Fatalf("errs = %d, want 0", errs)
|
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) {
|
// The walk stats each file as it is discovered: every record must
|
||||||
t.Fatalf("paths = %q, want %q", paths, want)
|
// 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()
|
t.Parallel()
|
||||||
|
|
||||||
// Exercise the dispatcher with more directories than workers and
|
// Exercise the dispatcher with more directories than workers and
|
||||||
@@ -161,19 +208,17 @@ func TestWalkPassDeepAndWide(t *testing.T) {
|
|||||||
|
|
||||||
slices.Sort(want)
|
slices.Sort(want)
|
||||||
|
|
||||||
paths, errs := walkPass([]string{dir}, false, 8)
|
recs, errs := collectWalk(t, []string{dir}, false, 8)
|
||||||
if errs != 0 {
|
if errs != 0 {
|
||||||
t.Fatalf("errs = %d, want 0", errs)
|
t.Fatalf("errs = %d, want 0", errs)
|
||||||
}
|
}
|
||||||
|
|
||||||
slices.Sort(paths)
|
if got := walkedPaths(recs); !slices.Equal(got, want) {
|
||||||
|
t.Fatalf("walked %d paths, want %d", len(got), len(want))
|
||||||
if !slices.Equal(paths, want) {
|
|
||||||
t.Fatalf("walked %d paths, want %d", len(paths), len(want))
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWalkPassMultipleRoots(t *testing.T) {
|
func TestWalkMultipleRoots(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
rootA := t.TempDir()
|
rootA := t.TempDir()
|
||||||
@@ -188,19 +233,17 @@ func TestWalkPassMultipleRoots(t *testing.T) {
|
|||||||
|
|
||||||
// Operands are enumerated concurrently by the shared pool; order
|
// Operands are enumerated concurrently by the shared pool; order
|
||||||
// is unspecified.
|
// is unspecified.
|
||||||
paths, errs := walkPass([]string{rootA, rootB}, false, 4)
|
recs, errs := collectWalk(t, []string{rootA, rootB}, false, 4)
|
||||||
if errs != 0 {
|
if errs != 0 {
|
||||||
t.Fatalf("errs = %d, want 0", errs)
|
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)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
|
func TestWalkFileAndSymlinkOperands(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
dir := t.TempDir()
|
dir := t.TempDir()
|
||||||
@@ -213,20 +256,20 @@ func TestWalkPassFileAndSymlinkOperands(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// A regular-file operand is emitted as itself.
|
// A regular-file operand is emitted as itself, statted.
|
||||||
paths, errs := walkPass([]string{f}, false, 2)
|
recs, errs := collectWalk(t, []string{f}, false, 2)
|
||||||
if errs != 0 || !slices.Equal(paths, []string{f}) {
|
if errs != 0 || len(recs) != 1 || recs[0].path != f || recs[0].size != 4 {
|
||||||
t.Fatalf("file operand: paths = %q, errs = %d", paths, errs)
|
t.Fatalf("file operand: recs = %+v, errs = %d", recs, errs)
|
||||||
}
|
}
|
||||||
|
|
||||||
// A symlink operand is not followed and yields nothing.
|
// A symlink operand is not followed and yields nothing.
|
||||||
paths, errs = walkPass([]string{link}, false, 2)
|
recs, errs = collectWalk(t, []string{link}, false, 2)
|
||||||
if errs != 0 || len(paths) != 0 {
|
if errs != 0 || len(recs) != 0 {
|
||||||
t.Fatalf("symlink operand: paths = %q, errs = %d", paths, errs)
|
t.Fatalf("symlink operand: recs = %+v, errs = %d", recs, errs)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestWalkPassOneFilesystemSameFS(t *testing.T) {
|
func TestWalkOneFilesystemSameFS(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
// Everything in one filesystem: -x must not skip anything.
|
// 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")),
|
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 {
|
if errs != 0 {
|
||||||
t.Fatalf("errs = %d, want 0", errs)
|
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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
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)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -662,14 +677,88 @@ func TestSyncScanOverlappingRoots(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestUniquePaths(t *testing.T) {
|
func TestScanSkipsUniqueSizes(t *testing.T) {
|
||||||
t.Parallel()
|
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) {
|
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 = make(map[string]fileSig)
|
||||||
}
|
}
|
||||||
|
|
||||||
node.files[comps[len(comps)-1]] = fileSig{
|
sig := fileSig{size: r.size, head: r.head, tail: r.tail}
|
||||||
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
|
return super, allDirs
|
||||||
|
|||||||
Reference in New Issue
Block a user