13 Commits

Author SHA1 Message Date
d4b43ebb30 Merge branch 'size-census': hash only files with shared sizes
All checks were successful
check / check (push) Successful in 1m8s
2026-07-25 06:06:30 +07:00
b62b4f297f Stat in the walk, hash only shared sizes, flush batches mid-scan
Restructure scan into three phases: walk+stat, hash, update.

The stat pass is folded into the walk workers: each regular file is
lstatted as its directory is read, while the metadata is hot. The
walk builds a scan-wide size census (walked files plus records
outside the scan roots), and unchanged already-hashed files resolve
during the walk without further work.

Only files whose size at least one other file shares are ever read:
a size-unique file cannot be a duplicate, so it is recorded without
hashes (head and tail empty). When a later scan makes its size
shared, the file is hashed then, even if otherwise unchanged. report
excludes unhashed records; trees gives them a never-matching
signature so a tree containing one never compares equal to another.

Hashed records are committed in batched transactions while the hash
phase runs, so an interrupted scan keeps everything hashed so far
and the next run resumes cheaply. The hash phase total is exact,
giving a meaningful ETA.

Memory drops accordingly: the existing-record index holds only path,
size, mtime, and a hashed flag (no hash values); the walk carries one
small record per candidate file; overlapping operands are pruned up
front instead of deduplicating every walked path in a scan-wide set.
Files no bigger than one chunk are hashed with a single read.
2026-07-25 06:06:22 +07:00
340bdbe39e Merge branch 'make-default-target': plain make builds the binary
All checks were successful
check / check (push) Successful in 1m12s
2026-07-24 11:21:18 +07:00
a1c3b852c3 Make the binary the default Make target
All checks were successful
check / check (push) Successful in 4s
Plain make now builds sfdupes (previously the default was all =
check + build); make build remains as an alias, so the Dockerfile
and existing habits keep working. The sfdupes target is phony: go
build's own cache decides what to recompile.
2026-07-24 11:21:16 +07:00
9f03eb3e2a Merge branch 'scan-wide-phases': scan-wide phases, concurrent operands, batched updates
All checks were successful
check / check (push) Successful in 1m8s
2026-07-24 10:26:54 +07:00
3ecf73c80a Make phases scan-wide, walk operands concurrently, batch updates
All checks were successful
check / check (push) Successful in 5s
All PATH operands belong to a single scan: every operand seeds the
shared walk worker pool, and each pass (walk, stat, hash, update)
runs exactly once over the whole scan, so pass totals, percentages,
and ETAs are scan-global. The per-operand walk/hash/update cycles and
their stderr operand announcements are gone; duplicate paths from
overlapping operands are deduplicated before stat.

The update pass now commits in batched transactions (10k changes per
batch) instead of one scan-wide transaction: the filesystem is
authoritative and the database is an eventually-consistent reflection
of it, so scan-level atomicity buys nothing, while batches keep the
WAL small and let concurrent reports observe progress.
2026-07-24 10:26:52 +07:00
732fc351d7 Merge branch 'parallel-phases': sequential phases, parallelism within each
All checks were successful
check / check (push) Successful in 1m9s
2026-07-24 08:12:49 +07:00
1e7a519608 Split the stat pass back out of the walk
All checks were successful
check / check (push) Successful in 4s
Phases are strictly sequential again — walk, stat, hash, update per
operand — with parallelism only inside each phase. The walk
enumerates paths with per-directory workers (no lstat of file
entries); the stat pass lstats every collected path with per-file
workers, restoring its exact-total/ETA progress bar and per-file
parallelism inside wide flat directories.
2026-07-24 08:12:47 +07:00
09ff9b5f30 Merge branch 'scan-operand-progress': announce operands on stderr
All checks were successful
check / check (push) Successful in 1m15s
2026-07-24 07:52:55 +07:00
a0f0050ada Announce each operand on stderr before its passes
All checks were successful
check / check (push) Successful in 6s
With per-operand walk/hash/update cycles, a multi-operand invocation
(e.g. scan /srv/*) showed pass totals that looked like the whole
run's: an operator watching operand 3 of 14 hash 300k files concluded
the other 20M files were being skipped. Print the operand path and
its position before each cycle.
2026-07-24 07:52:54 +07:00
6a15b879de Merge branch 'parallel-walk': parallel per-directory walk, per-operand commits
All checks were successful
check / check (push) Successful in 1m10s
2026-07-24 07:44:28 +07:00
dced5cf0d2 Parallelize the walk and commit per operand
All checks were successful
check / check (push) Successful in 5s
Replace the single-goroutine WalkDir traversal with a per-directory
worker pool: workers read directories concurrently and lstat entries
while each directory is fresh in cache, recording size and mtime
during the walk. This folds the separate stat pass away (halving
metadata I/O per run) and overlaps metadata latency, which dominated
on busy pools — a sequential walk of a ~22M-file tree was observed
taking over 4 hours.

Each PATH operand now loads its scope, walks, hashes, and commits in
its own transaction, so an interrupted scan keeps every operand
completed so far; a later overlapping operand sees the records
committed by earlier ones and reuses them unchanged.
2026-07-24 07:44:18 +07:00
3ebf98940a Specify parallel per-directory walk and per-operand commits
The walk pass is a single goroutine; on a busy ZFS pool it manages
only a few thousand directory entries per second and takes hours at
~20M files. Respecify it as a worker-pool traversal that reads
directories concurrently and records size/mtime during the walk,
folding away the separate stat pass and halving metadata I/O. Each
PATH operand now commits in its own transaction so an interrupted
scan keeps the operands completed so far.
2026-07-24 06:28:12 +07:00
11 changed files with 1038 additions and 462 deletions

View File

@@ -6,13 +6,15 @@ BINARY := sfdupes
VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo dev) VERSION := $(shell git describe --tags --always --dirty 2>/dev/null || echo dev)
LDFLAGS := -X main.Version=$(VERSION) LDFLAGS := -X main.Version=$(VERSION)
.PHONY: all build test lint fmt fmt-check check docker hooks clean .PHONY: sfdupes build test lint fmt fmt-check check docker hooks clean
all: check build # Default target: build the binary. Phony so go build (which has its
# own build cache) always decides what to recompile.
build: sfdupes:
go build -ldflags "$(LDFLAGS)" -o $(BINARY) go build -ldflags "$(LDFLAGS)" -o $(BINARY)
build: sfdupes
test: test:
@go test -timeout 30s -cover ./... || \ @go test -timeout 30s -cover ./... || \
{ echo "--- Rerunning with -v for details ---"; \ { echo "--- Rerunning with -v for details ---"; \

151
README.md
View File

@@ -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
@@ -130,10 +135,15 @@ All three subcommands operate on a single SQLite database file:
database file is a fatal error (exit 1) telling the user to run database file is a fatal error (exit 1) telling the user to run
`scan` first. `scan` first.
- 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 reports see report while a cron `scan` is in progress is safe. The filesystem
the last committed state. Each scan commits its changes in a single is authoritative; the database is an eventually-consistent
transaction, so a report never observes a half-finished scan and a reflection of it. Hashed records are committed in batched
scan that dies partway leaves the previous state intact. 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 - 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):
@@ -149,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
@@ -160,20 +174,34 @@ or a regular file; an operand that does not exist is a fatal error
(exit 1). Because database records persist between runs and are keyed (exit 1). Because database records persist between runs and are keyed
by absolute path, each operand is resolved to an absolute, lexically 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. Operands are walked in the order not depend on the working directory. All operands belong to a single
given; overlapping operands (one containing another) are harmless — a scan and are enumerated concurrently: every operand seeds the shared
file reached via multiple operands 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 `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
@@ -184,20 +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**, in this order, so that every `scan` runs **three sequential phases over the whole scan**.
expensive pass has an exact total for meaningful progress and ETA: Parallelism lives inside each phase; batched database writes begin
during the hash phase:
1. **walk** — recursively enumerate the tree under each `PATH` in 1. **walk + stat** — enumerate the trees under all `PATH` operands
turn, collecting the list of regular-file paths. Total unknown concurrently with the walk worker pool: every operand seeds the
while running: show a live count, not a percentage. shared queue, and each worker reads one directory at a time,
2. **stat** — `lstat` every collected path, recording size and mtime. handing discovered subdirectories back to the queue and running
3. **hash** — for each new or changed file (per the rules above), read `lstat` on each regular file as it is discovered (while the
the first `min(1024, size)` bytes and the last `min(1024, size)` directory's metadata is still hot). Sequential directory
bytes (the two reads overlap when `size < 2048`; for `size == 0` enumeration is metadata-latency-bound and takes hours at tens of
hash the empty input) and compute the SHA-256 of each. Unchanged millions of files; per-directory parallelism is what makes the
files are not read and do not appear in this pass's total. walk tractable on large or busy pools. The walk builds the size
4. **update** — apply all insertions, updates, and deletions to the census and resolves unchanged already-hashed files on the fly;
database in a single transaction. 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: Rules for the walk:
@@ -219,9 +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 stat and hash passes use a worker pool (`--workers`, Concurrency: the walk phase (which also stats files) and the hash
default `runtime.NumCPU()`). The main goroutine owns database writes phase each use a worker pool of `--workers` workers (default
and progress rendering; progress display must never block the workers. `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 `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:
@@ -246,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`.
@@ -283,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
@@ -333,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, hash, Each phase gets its own display. The walk has no known total while
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
@@ -347,12 +410,9 @@ 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
@@ -375,7 +435,8 @@ Additional requirements:
The `Makefile` is the single source of truth for all operations: The `Makefile` is the single source of truth for all operations:
- `make build` — build the `sfdupes` binary (cgo disabled). - `make` / `make build` — build the `sfdupes` binary (cgo
disabled); building is the default target.
- `make test` — run the test suite (30-second timeout; reruns with - `make test` — run the test suite (30-second timeout; reruns with
`-v` on failure). `-v` on failure).
- `make lint` — run `golangci-lint` with the repo config. - `make lint` — run `golangci-lint` with the repo config.

31
TODO.md
View File

@@ -19,6 +19,37 @@
# Completed Steps # Completed Steps
- make the binary the default Make target (2026-07-24, branch
`make-default-target`): plain `make` now builds `sfdupes`
(previously it ran `check` plus `build`); `make build` remains as
an alias
- scan-wide phases, concurrent operands, batched updates (2026-07-24,
branch `scan-wide-phases`): all operands seed the shared walk pool
and every pass runs once over the whole scan, so totals and ETAs
are scan-global; the per-operand walk/hash/update cycles and their
stderr announcements are gone; the update pass commits in batched
transactions — the filesystem is authoritative and the database an
eventually-consistent reflection, so scan-level atomicity is not
required
- split the stat pass back out of the walk (2026-07-24, branch
`parallel-phases`): phases are strictly sequential again — walk,
stat, hash, update per operand — with parallelism only inside each
phase; the walk enumerates paths with per-directory workers and the
stat pass lstats them with per-file workers, restoring the exact
total/ETA stat bar
- announce each operand on stderr before its passes (2026-07-24,
branch `scan-operand-progress`): with per-operand walk/hash/update
cycles, a multi-operand run (e.g. `scan /srv/*`) showed pass totals
that looked like the whole run's — an operator watching operand 3 of
14 hash 300k files concluded 20M files were being skipped
- parallel walk (2026-07-24, branch `parallel-walk`): the walk pass
was a single goroutine and took hours at ~20M files on a busy pool
(observed: 22M files in 4h on a ZFS server); it is now a
per-directory worker-pool traversal that records size/mtime during
the walk (folding away the separate stat pass, halving metadata
I/O), and each `PATH` operand commits in its own transaction so an
interrupted scan keeps completed operands
- persistent scan database (2026-07-24, branch `persistent-database`): - persistent scan database (2026-07-24, branch `persistent-database`):
`scan` now maintains a SQLite database (`modernc.org/sqlite`, pure `scan` now maintains a SQLite database (`modernc.org/sqlite`, pure
Go, cgo stays disabled) keyed by absolute path that survives between Go, cgo stays disabled) keyed by absolute path that survives between

73
db.go
View File

@@ -8,6 +8,7 @@ import (
"io/fs" "io/fs"
"os" "os"
"path/filepath" "path/filepath"
"slices"
"strconv" "strconv"
// The pure-Go SQLite driver, registered as "sqlite"; keeps cgo // The pure-Go SQLite driver, registered as "sqlite"; keeps cgo
@@ -237,12 +238,78 @@ 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
// transaction during the update pass. The filesystem is authoritative
// and the database an eventually-consistent reflection of it, so
// scan-level atomicity is not required; smaller transactions keep the
// WAL small and let concurrent reports observe progress.
const updateBatchSize = 10000
// applyChanges writes one scan's database changes — upserts for new and // applyChanges writes one scan's database changes — upserts for new and
// changed files, deletes for vanished ones — in a single transaction, // changed files, deletes for vanished ones — in batched transactions.
// so a concurrent report never observes a half-finished scan. Progress // Progress is rendered on prog (one increment per change).
// is rendered on prog (one increment per change).
func applyChanges(db *sql.DB, upserts []scanRec, deletes []string, func applyChanges(db *sql.DB, upserts []scanRec, deletes []string,
prog *progress, prog *progress,
) error {
for batch := range slices.Chunk(upserts, updateBatchSize) {
err := applyBatch(db, batch, nil, prog)
if err != nil {
return err
}
}
for batch := range slices.Chunk(deletes, updateBatchSize) {
err := applyBatch(db, nil, batch, prog)
if err != nil {
return err
}
}
return nil
}
// applyBatch commits one batch of upserts and deletes in a single
// transaction.
func applyBatch(db *sql.DB, upserts []scanRec, deletes []string,
prog *progress,
) error { ) error {
ctx := context.Background() ctx := context.Background()

View File

@@ -4,6 +4,7 @@ import (
"context" "context"
"database/sql" "database/sql"
"errors" "errors"
"fmt"
"path/filepath" "path/filepath"
"slices" "slices"
"strings" "strings"
@@ -178,3 +179,46 @@ func TestApplyChangesRoundTrip(t *testing.T) {
t.Fatalf("rows = %+v, want just %+v", got, upd) t.Fatalf("rows = %+v, want just %+v", got, upd)
} }
} }
func TestApplyChangesBatching(t *testing.T) {
t.Parallel()
db := openTestDB(t)
// One more change than the batch size, so the update spans two
// transactions.
n := updateBatchSize + 1
recs := make([]scanRec, 0, n)
for i := range n {
recs = append(recs, scanRec{
size: int64(i), mtime: 1, head: "h", tail: "t",
path: fmt.Sprintf("/batch/%07d", i),
})
}
err := applyChanges(db, recs, nil, newProgress("update", int64(n)))
if err != nil {
t.Fatalf("applyChanges: %v", err)
}
got, err := loadFileRows(db)
if err != nil || len(got) != n {
t.Fatalf("loadFileRows = %d rows, %v; want %d", len(got), err, n)
}
deletes := make([]string, 0, n)
for _, r := range recs {
deletes = append(deletes, r.path)
}
err = applyChanges(db, nil, deletes, newProgress("update", int64(n)))
if err != nil {
t.Fatalf("applyChanges deletes: %v", err)
}
got, err = loadFileRows(db)
if err != nil || len(got) != 0 {
t.Fatalf("loadFileRows = %d rows, %v; want 0", len(got), err)
}
}

View File

@@ -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")

View File

@@ -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()

View File

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

870
scan.go
View File

@@ -4,38 +4,47 @@ import (
"crypto/sha256" "crypto/sha256"
"database/sql" "database/sql"
"encoding/hex" "encoding/hex"
"errors"
"fmt" "fmt"
"io/fs" "io/fs"
"os" "os"
"path/filepath" "path/filepath"
"slices" "slices"
"strings" "strings"
"sync"
"syscall" "syscall"
) )
// 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 that stopped being a regular file // fileRec carries one statted file between the scan phases.
// between the walk and stat passes.
var errNotRegular = errors.New("no longer a regular file")
// fileRec carries one file between the stat and hash passes.
type fileRec struct { 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
@@ -88,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 {
@@ -98,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.
// walk and stat everything, hash only new or changed files, and apply // Entries consumed from existing mark files verified this run;
// the resulting record insertions, updates, and deletions in a single // whatever remains after the walk and hash phases is deleted by the
// transaction. 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) 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 by this scan. // 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
@@ -181,194 +446,240 @@ func underRoot(path, root string) bool {
return strings.HasPrefix(path, prefix) return strings.HasPrefix(path, prefix)
} }
// uniquePaths deduplicates the walked paths, preserving order. // dirJob is one directory awaiting traversal by the walk workers. It
// Overlapping operands can reach the same file more than once, but the // carries its operand's filesystem device so -x can stop at
// database keys records by path, so each file is processed once. // filesystem boundaries.
func uniquePaths(paths []string) []string { type dirJob struct {
seen := make(map[string]bool, len(paths)) path string
out := make([]string, 0, len(paths)) rootDev uint64
rootDevOK bool
}
for _, p := range paths { // walkEvent is one walk result delivered to the walk phase: a regular
if seen[p] { // file's statted record, or a warning when fail is set.
continue type walkEvent struct {
rec fileRec
warn string
fail bool
}
// startWalk seeds every root into the shared walk worker pool and
// returns the event stream: one record per regular file, one warning
// event per per-path error. The channel is closed when the walk
// completes.
func startWalk(roots []string, oneFS bool, workers int) <-chan walkEvent {
jobs, subdirs, events := startWalkWorkers(workers, oneFS)
go func() {
initial := make([]dirJob, 0, len(roots))
for _, root := range roots {
initial = append(initial, seedRoot(root, events)...)
} }
seen[p] = true dispatchDirs(initial, jobs, subdirs)
}()
out = append(out, p) return events
}
return out
} }
// partitionChanged splits the stat results into files that must be // seedRoot turns one PATH operand into the walk's starting state: a
// hashed (new, or changed) and files whose existing records are reused // regular-file operand is statted and emitted directly, a directory
// without reading them: a file is unchanged when its statted size // operand becomes an initial job, and a symlink or other non-regular
// equals the recorded size and its statted mtime is not newer than the // operand yields nothing (symlinks are never followed, including as
// recorded mtime. // operands).
func partitionChanged(recs []fileRec, func seedRoot(root string, events chan<- walkEvent) []dirJob {
existing map[string]scanRec, fi, err := os.Lstat(root)
) ([]fileRec, []scanRec) { if err != nil {
var ( events <- walkEvent{
toHash []fileRec warn: fmt.Sprintf("walk %s: %v", root, err),
unchanged []scanRec fail: true,
)
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 nil
} }
return toHash, unchanged switch {
} case fi.IsDir():
if filepath.Base(root) == ".zfs" {
// collectDeletes returns the existing record paths that were not
// successfully processed this run: vanished files, plus paths that
// failed to stat or hash. The database keeps only records verified by
// the latest scan covering them. The result is sorted so the update
// pass is deterministic.
func collectDeletes(existing map[string]scanRec,
unchanged, hashed []scanRec,
) []string {
kept := make(map[string]bool, len(unchanged)+len(hashed))
for _, r := range unchanged {
kept[r.path] = true
}
for _, r := range hashed {
kept[r.path] = true
}
var deletes []string
for p := range existing {
if !kept[p] {
deletes = append(deletes, p)
}
}
slices.Sort(deletes)
return deletes
}
// applyPass writes the scan's changes to the database under an update
// progress display (one item per insertion, update, or deletion).
func applyPass(db *sql.DB, upserts []scanRec, deletes []string) error {
prog := newProgress("update", int64(len(upserts)+len(deletes)))
err := applyChanges(db, upserts, deletes, prog)
prog.finish()
return err
}
// treeWalker carries the walk-pass state shared by all PATH operands.
type treeWalker struct {
prog *progress
oneFS bool
paths []string
errs int
}
// walkPass enumerates every regular file under each root operand in
// order. It never follows symlinks, never descends into directories
// named .zfs, and warns and continues on any per-path error. With
// oneFS set it never descends into a directory on a different
// filesystem than its root operand.
func walkPass(roots []string, oneFS bool) ([]string, int) {
w := &treeWalker{prog: newProgress("walk", -1), oneFS: oneFS}
for _, root := range roots {
w.walkRoot(root)
}
w.prog.finish()
return w.paths, w.errs
}
// walkRoot walks a single PATH operand, appending regular-file paths.
func (w *treeWalker) walkRoot(root string) {
rootDev, rootDevOK := deviceOf(root)
walkErr := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
if err != nil {
w.errs++
w.prog.warnf("walk %s: %v", p, err)
if d != nil && d.IsDir() {
return filepath.SkipDir
}
return nil return nil
} }
if d.IsDir() { dev, ok := deviceOfInfo(fi)
return w.dirAction(p, d, rootDev, rootDevOK)
return []dirJob{{path: root, rootDev: dev, rootDevOK: ok}}
case fi.Mode().IsRegular():
events <- walkEvent{rec: fileRec{
path: root,
size: fi.Size(),
mtime: fi.ModTime().Unix(),
}}
return nil
default:
return nil
}
}
// startWalkWorkers starts the walk worker pool. Each worker processes
// one directory at a time, emitting an event per regular file and
// handing discovered subdirectories back to the dispatcher; events is
// closed once every worker has finished.
func startWalkWorkers(workers int,
oneFS bool,
) (chan dirJob, chan []dirJob, chan walkEvent) {
jobs := make(chan dirJob, workQueueDepth)
subdirs := make(chan []dirJob, workers)
events := make(chan walkEvent, workQueueDepth)
var wg sync.WaitGroup
for range workers {
wg.Go(func() {
for job := range jobs {
subdirs <- walkOneDir(job, oneFS, events)
}
})
}
go func() {
wg.Wait()
close(events)
}()
return jobs, subdirs, events
}
// dispatchDirs feeds directory jobs to the walk workers, queueing
// newly discovered subdirectories (newest first, which keeps the
// frontier small) until every directory has been processed, then
// closes jobs.
func dispatchDirs(initial []dirJob, jobs chan<- dirJob,
subdirs <-chan []dirJob,
) {
go func() {
queue := slices.Clone(initial)
pending := len(queue)
for pending > 0 {
var (
out chan<- dirJob
next dirJob
)
if len(queue) > 0 {
out = jobs
next = queue[len(queue)-1]
}
select {
case out <- next:
queue = queue[:len(queue)-1]
case subs := <-subdirs:
pending += len(subs) - 1
queue = append(queue, subs...)
}
}
close(jobs)
}()
}
// walkOneDir reads one directory, emitting an event per regular-file
// entry and a warning event per unreadable one, and returns the
// subdirectories to descend into.
func walkOneDir(job dirJob, oneFS bool, events chan<- walkEvent) []dirJob {
entries, err := os.ReadDir(job.path)
if err != nil {
events <- walkEvent{
warn: fmt.Sprintf("walk %s: %v", job.path, err),
fail: true,
}
return nil
}
var subs []dirJob
for _, e := range entries {
p := filepath.Join(job.path, e.Name())
if e.IsDir() {
if sub, ok := subdirJob(p, e, job, oneFS, events); ok {
subs = append(subs, sub)
}
continue
} }
// Regular files only: skip symlinks, sockets, FIFOs, and // Regular files only: skip symlinks, sockets, FIFOs, and
// device nodes. // device nodes.
if !d.Type().IsRegular() { if !e.Type().IsRegular() {
return nil continue
} }
w.paths = append(w.paths, p) emitFile(p, e, events)
w.prog.increment()
return nil
})
if walkErr != nil {
fatalf("walk %s: %v", root, walkErr)
} }
return subs
} }
// dirAction decides whether the walk descends into directory p. // emitFile stats one regular directory entry and emits its record.
func (w *treeWalker) dirAction(p string, d fs.DirEntry, rootDev uint64, rootDevOK bool) error { // The lstat happens here in the walk worker, while the directory's
// ZFS snapshot pseudo-dirs would list every file once per // metadata is still hot; a path that fails to stat (or stops being a
// snapshot; never descend. // regular file) between the directory read and the lstat is warned
if d.Name() == ".zfs" { // about and skipped.
return filepath.SkipDir func emitFile(p string, e fs.DirEntry, events chan<- walkEvent) {
} info, err := e.Info()
if !w.oneFS || !rootDevOK {
return nil
}
info, err := d.Info()
if err != nil { if err != nil {
w.errs++ events <- walkEvent{
warn: fmt.Sprintf("stat %s: %v", p, err),
fail: true,
}
w.prog.warnf("walk %s: %v", p, err) return
return filepath.SkipDir
} }
if dev, ok := deviceOfInfo(info); ok && dev != rootDev { if !info.Mode().IsRegular() {
return filepath.SkipDir return
} }
return nil events <- walkEvent{rec: fileRec{
path: p,
size: info.Size(),
mtime: info.ModTime().Unix(),
}}
} }
// deviceOf returns the filesystem device ID of path without following // subdirJob applies the descent rules to directory p: never enter
// symlinks. // .zfs (ZFS snapshot pseudo-dirs would list every file once per
func deviceOf(path string) (uint64, bool) { // snapshot), and with -x never enter a directory on a different
fi, err := os.Lstat(path) // filesystem than its operand.
if err != nil { func subdirJob(p string, e fs.DirEntry, parent dirJob, oneFS bool,
return 0, false events chan<- walkEvent,
) (dirJob, bool) {
if e.Name() == ".zfs" {
return dirJob{}, false
} }
return deviceOfInfo(fi) job := dirJob{path: p, rootDev: parent.rootDev, rootDevOK: parent.rootDevOK}
if !oneFS || !parent.rootDevOK {
return job, true
}
info, err := e.Info()
if err != nil {
events <- walkEvent{
warn: fmt.Sprintf("walk %s: %v", p, err),
fail: true,
}
return dirJob{}, false
}
if dev, ok := deviceOfInfo(info); ok && dev != parent.rootDev {
return dirJob{}, false
}
return job, true
} }
// deviceOfInfo extracts the filesystem device ID from a FileInfo, when // deviceOfInfo extracts the filesystem device ID from a FileInfo, when
@@ -382,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
@@ -457,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)
@@ -548,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)

View File

@@ -5,6 +5,7 @@ import (
"crypto/sha256" "crypto/sha256"
"database/sql" "database/sql"
"encoding/hex" "encoding/hex"
"fmt"
"os" "os"
"path/filepath" "path/filepath"
"slices" "slices"
@@ -109,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)
@@ -130,19 +168,57 @@ func TestWalkPass(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
paths, errs := walkPass([]string{dir}, false) 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 TestWalkPassMultipleRoots(t *testing.T) { func TestWalkDeepAndWide(t *testing.T) {
t.Parallel()
// Exercise the dispatcher with more directories than workers and
// with nesting deeper than the worker count.
dir := t.TempDir()
deep := "deep" + strings.Repeat("/d", 30)
want := make([]string, 0, 41)
want = append(want, writeFile(t, dir, deep+"/f", []byte("x")))
for i := range 40 {
want = append(want, writeFile(t, dir,
fmt.Sprintf("wide/%02d/f", i), []byte("y")))
}
slices.Sort(want)
recs, errs := collectWalk(t, []string{dir}, false, 8)
if errs != 0 {
t.Fatalf("errs = %d, want 0", errs)
}
if got := walkedPaths(recs); !slices.Equal(got, want) {
t.Fatalf("walked %d paths, want %d", len(got), len(want))
}
}
func TestWalkMultipleRoots(t *testing.T) {
t.Parallel() t.Parallel()
rootA := t.TempDir() rootA := t.TempDir()
@@ -153,18 +229,21 @@ func TestWalkPassMultipleRoots(t *testing.T) {
writeFile(t, rootB, "b1", []byte("3")), writeFile(t, rootB, "b1", []byte("3")),
} }
paths, errs := walkPass([]string{rootA, rootB}, false) slices.Sort(want)
// Operands are enumerated concurrently by the shared pool; order
// is unspecified.
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)
} }
// Operands are walked in the order given. if got := walkedPaths(recs); !slices.Equal(got, want) {
if !slices.Equal(paths, want) { t.Fatalf("paths = %q, want %q", got, 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()
@@ -177,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) 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) 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.
@@ -200,58 +279,37 @@ 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) 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 TestDeviceOf(t *testing.T) { func TestDeviceOfInfo(t *testing.T) {
t.Parallel() t.Parallel()
dir := t.TempDir() dir := t.TempDir()
dev1, ok1 := deviceOf(dir) fi1, err := os.Lstat(dir)
if err != nil {
t.Fatal(err)
}
dev2, ok2 := deviceOf(dir) fi2, err := os.Lstat(dir)
if err != nil {
t.Fatal(err)
}
dev1, ok1 := deviceOfInfo(fi1)
dev2, ok2 := deviceOfInfo(fi2)
if !ok1 || !ok2 || dev1 != dev2 { if !ok1 || !ok2 || dev1 != dev2 {
t.Fatalf("deviceOf unstable: %d/%v vs %d/%v", dev1, ok1, dev2, ok2) t.Fatalf("deviceOfInfo unstable: %d/%v vs %d/%v",
} dev1, ok1, dev2, ok2)
if _, ok := deviceOf(filepath.Join(dir, "missing")); ok {
t.Fatal("deviceOf reported ok for a missing path")
}
}
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)
} }
} }
@@ -607,11 +665,101 @@ func TestSyncScanOverlappingRoots(t *testing.T) {
writeFile(t, dir, "sub/f", pattern(1, 10)) writeFile(t, dir, "sub/f", pattern(1, 10))
// A file reachable via two overlapping operands yields one record. // A file reachable via two overlapping operands is deduplicated
// by path in the shared walk and processed once.
st := syncTree(t, db, dir, filepath.Join(dir, "sub")) st := syncTree(t, db, dir, filepath.Join(dir, "sub"))
if st.added != 1 { if st != (scanStats{added: 1}) {
t.Fatalf("stats = %+v, want 1 added", st) t.Fatalf("stats = %+v, want 1 added", st)
} }
if got := recordPaths(dbRecords(t, db)); len(got) != 1 {
t.Fatalf("records = %q, want exactly one", got)
}
}
func TestScanSkipsUniqueSizes(t *testing.T) {
t.Parallel()
dir := t.TempDir()
db := openTestDB(t)
a := writeFile(t, dir, "a.bin", pattern(1, 500))
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("pruneRoots = %q, want %q", got, want)
}
} }
func TestReportsNeverTouchFilesystem(t *testing.T) { func TestReportsNeverTouchFilesystem(t *testing.T) {
@@ -659,14 +807,3 @@ func TestUnderRoot(t *testing.T) {
} }
} }
} }
func TestUniquePaths(t *testing.T) {
t.Parallel()
got := uniquePaths([]string{"/a", "/b", "/a", "/c", "/b"})
want := []string{"/a", "/b", "/c"}
if !slices.Equal(got, want) {
t.Fatalf("uniquePaths = %q, want %q", got, want)
}
}

View File

@@ -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