Stop scan cleanly on SIGINT or SIGTERM (closes #5)
check / check (push) Failing after 3s

A first SIGINT or SIGTERM cancels the scan. It commits the hashed
records still in its batch, with a context that is not cancelled for
that one write, and starts no other write or deletion; deletions need
a complete walk, so records under paths an interrupted walk never
reached are kept. A batch whose commit failed is now kept for that
final commit instead of dropped. The progress display is finished (a
bar stopped short is no longer filled up), `scan: interrupted after N
files` goes to stderr, and the exit code is 1. A second signal ends the
process at once. A SIGINT inherited as ignored stays ignored.

Model: opus-5-5
This commit was merged in pull request #79.
This commit is contained in:
2026-10-04 09:30:38 +02:00
parent 33cf3dd29a
commit 4847882e46
8 changed files with 446 additions and 62 deletions
+25 -3
View File
@@ -184,8 +184,9 @@ All three subcommands operate on a single SQLite database file:
(keeping the WAL small and letting concurrent reports observe (keeping the WAL small and letting concurrent reports observe
progress), so a report may see a scan's changes partially applied, progress), so a report may see a scan's changes partially applied,
and a scan that dies partway leaves a valid database holding and a scan that dies partway leaves a valid database holding
everything hashed so far; the next scan skips those records and every batch committed so far (an interrupted scan also commits the
converges toward the filesystem. batch in progress, see "Error handling and exit codes"); the next
scan skips those records and converges toward the filesystem.
- `scan` switches the database back to rollback-journal mode when it - `scan` switches the database back to rollback-journal mode when it
closes it, so between scans the database file alone holds the whole closes it, so between scans the database file alone holds the whole
database. Each switch needs the database to itself: a `scan` that database. Each switch needs the database to itself: a `scan` that
@@ -616,6 +617,8 @@ Additional requirements:
waits for its next item. waits for its next item.
- A warning printed during a phase always lands on a line of its own, - A warning printed during a phase always lands on a line of its own,
never inside the progress display. never inside the progress display.
- A bar whose phase stops short of its total, as an interrupted one
does, is left as last drawn rather than filled up.
- `report` and `trees` modes need no progress display, only their - `report` and `trees` modes need no progress display, only their
stderr summaries. stderr summaries.
@@ -625,7 +628,8 @@ Additional requirements:
- `1`: fatal error (e.g., a `PATH` operand does not exist, another - `1`: fatal error (e.g., a `PATH` operand does not exist, another
`scan` is already running against the same database, the database `scan` is already running against the same database, the database
cannot be created/opened/read/written, a missing database for cannot be created/opened/read/written, a missing database for
`report`/`trees`, stdout write failure). `report`/`trees`, stdout write failure), or a `scan` stopped by
`SIGINT` or `SIGTERM` (see below).
- `2`: usage error (including `scan` with no `PATH` operand and - `2`: usage error (including `scan` with no `PATH` operand and
`report`/`trees` with any positional argument). `report`/`trees` with any positional argument).
@@ -641,6 +645,24 @@ stderr and exits 1. Two cases never reach sfdupes as a failed write:
the output is discarded and the run succeeds, as with the output is discarded and the run succeeds, as with
`> /dev/null`. `> /dev/null`.
`scan` stops cleanly on `SIGINT` (Ctrl-C) or `SIGTERM`. Its workers
stop taking work, each finishing at most the directory listing or file
it is reading; the progress display is finished; and the records it has
hashed but not yet committed are committed, so the next scan does not
hash them again. Apart from that commit it starts no further writes or
deletions: records are deleted only after a complete walk, so those
under paths an interrupted walk never reached are kept. The database is
closed and the lock released as on any other exit, the line
`scan: interrupted after N files` goes to stderr, N being the number of
files the walk reached, and the exit code is 1. The next scan skips the
records already written and converges as usual.
After the first signal `scan` stops catching them, so a second one ends
it at once, as an uncaught signal does: the records not yet committed
are lost, and the database is left valid, as when any scan dies (see
"Database"). A `SIGINT` that `scan` inherits as ignored, as a script's
background job does, stays ignored.
## Entrypoints ## Entrypoints
This repository adheres to the This repository adheres to the
+4
View File
@@ -29,6 +29,10 @@
# Completed Steps # Completed Steps
- `scan` stops cleanly on `SIGINT` or `SIGTERM`: commits what it has
hashed, deletes nothing more, exits 1 (2026-10-04,
https://git.eeqj.de/sneak/sfdupes/issues/5)
- `report` and `trees` stream the records instead of holding them all in - `report` and `trees` stream the records instead of holding them all in
memory; the schema gains the `files_signature` index (2026-10-04, memory; the schema gains the `files_signature` index (2026-10-04,
https://git.eeqj.de/sneak/sfdupes/issues/14) https://git.eeqj.de/sneak/sfdupes/issues/14)
+247 -13
View File
@@ -4,11 +4,16 @@ import (
"context" "context"
"database/sql" "database/sql"
"errors" "errors"
"fmt"
"os" "os"
"os/signal"
"path/filepath" "path/filepath"
"slices"
"strconv" "strconv"
"strings"
"sync" "sync"
"sync/atomic" "sync/atomic"
"syscall"
"testing" "testing"
"time" "time"
) )
@@ -150,9 +155,8 @@ func assertRecordsIntact(t *testing.T, db *sql.DB, before []string) {
// Every one of those records would look vanished to the update phase. // Every one of those records would look vanished to the update phase.
// The guard is what stops the scan there, and this test is what // The guard is what stops the scan there, and this test is what
// notices if it stops doing so: deleting the guard, or making it // notices if it stops doing so: deleting the guard, or making it
// unreachable, makes the scan carry its truncated view into a later // unreachable, makes the scan carry its truncated view into the update
// phase and fail there instead, with a wrapped error rather than the // phase, which counts every record the walk never reached for removal.
// bare cancellation.
// //
//nolint:paralleltest // counts goroutines: must not run beside others //nolint:paralleltest // counts goroutines: must not run beside others
func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) { func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) {
@@ -182,10 +186,10 @@ func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) {
// assertWalkGuardAborted checks that the scan stopped at the post-walk // assertWalkGuardAborted checks that the scan stopped at the post-walk
// guard: with a census that is neither empty (the walk really ran) // guard: with a census that is neither empty (the walk really ran)
// nor complete (it really was cut short), and with the guard's own // nor complete (it really was cut short), and with no record counted
// bare cancellation as the error. A wrapped error means the partial // for removal. A removal count means the partial census was carried
// census was carried past the guard into the hash or update phase, // past the guard into the update phase, which is the failure this test
// which is the failure this test exists to catch. // exists to catch.
func assertWalkGuardAborted(t *testing.T, st scanStats, err error) { func assertWalkGuardAborted(t *testing.T, st scanStats, err error) {
t.Helper() t.Helper()
@@ -194,12 +198,6 @@ func assertWalkGuardAborted(t *testing.T, st scanStats, err error) {
err, context.Canceled) err, context.Canceled)
} }
if errors.Unwrap(err) != nil {
t.Errorf("syncScan reported %q, want the guard's bare "+
"cancellation: a wrapped error means the truncated census "+
"reached a later phase", err)
}
if st.unchanged == 0 { if st.unchanged == 0 {
t.Fatalf("stats = %+v: the census is empty, so the walk never "+ t.Fatalf("stats = %+v: the census is empty, so the walk never "+
"ran and the guard was reached for the wrong reason", st) "ran and the guard was reached for the wrong reason", st)
@@ -260,6 +258,242 @@ func TestSyncScanCancelledBeforeLoadIndex(t *testing.T) {
assertRecordsIntact(t, db, before) assertRecordsIntact(t, db, before)
} }
// hashCancelAtDone is the consultation on which the mid-hash test's
// context cancels itself. The walk of buildWalkCancelTree spends about
// one per file and three per directory, and the hash phase then one per
// file hashed, so this lands about half way through the hash phase.
const hashCancelAtDone = walkCancelFiles + 3*walkCancelDirs +
walkCancelFiles/2
// TestSyncScanCancelledMidHashKeepsHashedRecords cancels a first scan
// part-way through its hash phase. The fixture holds fewer files than a
// batch, so every file hashed is still waiting to be committed: the scan
// must commit them all before it returns, and the next scan must hash
// only the rest.
func TestSyncScanCancelledMidHashKeepsHashedRecords(t *testing.T) {
t.Parallel()
dir := buildWalkCancelTree(t)
db := openTestDB(t)
st, err := syncScan(newWalkClock(hashCancelAtDone), db,
[]string{dir}, walkCancelWorkers, false)
if !errors.Is(err, context.Canceled) {
t.Fatalf("syncScan cancelled mid-hash = %v, want %v",
err, context.Canceled)
}
if st.walked != walkCancelFiles || st.added == 0 ||
st.added >= walkCancelFiles {
t.Fatalf("stats = %+v: want the walk complete and the hash phase "+
"cut short", st)
}
if got := len(dbRecords(t, db)); got != st.added {
t.Errorf("%d records after the cancelled scan, want the %d it hashed",
got, st.added)
}
hashed := st.added
st = syncTree(t, db, dir)
if st.added != walkCancelFiles-hashed || st.unchanged != hashed {
t.Errorf("next scan stats = %+v, want %d added %d unchanged",
st, walkCancelFiles-hashed, hashed)
}
}
// storedPaths opens the database at path as report does, which fails
// unless it is a valid database, and returns its records' paths.
func storedPaths(t *testing.T, path string) []string {
t.Helper()
db, err := openReportDatabase(t.Context(), path)
if err != nil {
t.Fatal(err)
}
defer func() { _ = db.Close() }()
return recordPaths(dbRecords(t, db))
}
// TestRunScanInterrupted calls the scan entrypoint with a context that
// is already cancelled, as when a signal arrives at once. It must return
// errInterrupted promptly with its one line on stderr, leave the
// database valid and as it was, and leave nothing in the way of the
// next scan, which must bring the database up to date.
func TestRunScanInterrupted(t *testing.T) {
path := testDBPath(t)
t.Setenv(databaseEnv, path)
stderr := captureStderr(t)
dir := buildSmokeTree(t)
err := runScan(t.Context(), []string{dir}, walkCancelWorkers, false)
if err != nil {
t.Fatal(err)
}
before := storedPaths(t, path)
// A vanished file and a new one: the interrupted scan records
// neither.
gone := filepath.Join(dir, "a", "unique.bin")
err = os.Remove(gone)
if err != nil {
t.Fatal(err)
}
added := writeFile(t, dir, "a/new.bin", pattern(50, 10))
shown := len(stderr())
done := make(chan struct{})
go func() {
defer close(done)
err = runScan(cancelledContext(t), []string{dir}, walkCancelWorkers,
false)
}()
awaitReturn(t, done, "runScan")
if !errors.Is(err, errInterrupted) {
t.Fatalf("runScan on a cancelled context = %v, want %v",
err, errInterrupted)
}
want := "scan: interrupted after 0 files\n"
if got := stderr()[shown:]; got != want {
t.Errorf("stderr = %q, want %q", got, want)
}
assertNoSidecars(t, path)
if got := storedPaths(t, path); !slices.Equal(got, before) {
t.Errorf("records = %q after the interrupted scan, want %q",
got, before)
}
err = runScan(t.Context(), []string{dir}, walkCancelWorkers, false)
if err != nil {
t.Fatal(err)
}
got := storedPaths(t, path)
if slices.Contains(got, gone) || !slices.Contains(got, added) {
t.Errorf("records = %q after the next scan, want %q gone and %q "+
"added", got, gone, added)
}
}
// TestRunScanInterruptedMidHash interrupts the scan entrypoint part-way
// through its hash phase, after the database is open. It must return
// errInterrupted, release the lock, end stderr with its line counting
// every file the walk reached, close the database out of WAL mode, and
// keep the records it hashed.
func TestRunScanInterruptedMidHash(t *testing.T) {
path := testDBPath(t)
t.Setenv(databaseEnv, path)
stderr := captureStderr(t)
dir := buildWalkCancelTree(t)
err := runScan(newWalkClock(hashCancelAtDone), []string{dir},
walkCancelWorkers, false)
if !errors.Is(err, errInterrupted) {
t.Fatalf("runScan interrupted mid-hash = %v, want %v",
err, errInterrupted)
}
holdScanLock(t, path)
want := fmt.Sprintf("scan: interrupted after %d files\n", walkCancelFiles)
if got := stderr(); !strings.HasSuffix(got, want) {
t.Errorf("stderr = %q, want it to end with %q", got, want)
}
assertNoSidecars(t, path)
db, err := openReportDatabase(t.Context(), path)
if err != nil {
t.Fatal(err)
}
defer func() { _ = db.Close() }()
// A plain close also removes the sidecars, but leaves WAL mode on.
var mode string
err = db.QueryRowContext(t.Context(), "PRAGMA journal_mode").Scan(&mode)
if err != nil {
t.Fatal(err)
}
if mode != "delete" {
t.Errorf("journal mode = %q after the interrupted scan, want %q",
mode, "delete")
}
kept := len(dbRecords(t, db))
if kept == 0 || kept >= walkCancelFiles {
t.Errorf("%d records after the interrupted scan, want those it "+
"hashed: some but not all of the %d files", kept, walkCancelFiles)
}
}
// TestInterruptContextCatchesSIGTERM sends SIGTERM to the test process
// while the scan's handler is installed, and checks that it cancels the
// scan's context.
//
//nolint:paralleltest // signals the whole process: must not run beside a scan
func TestInterruptContextCatchesSIGTERM(t *testing.T) {
// Caught here as well, so that a handler that misses SIGTERM fails
// this test instead of ending the test process.
caught := make(chan os.Signal, 1)
signal.Notify(caught, syscall.SIGTERM)
defer signal.Stop(caught)
ctx, stop := interruptContext(t.Context())
defer stop()
err := syscall.Kill(os.Getpid(), syscall.SIGTERM)
if err != nil {
t.Fatal(err)
}
select {
case <-ctx.Done():
case <-time.After(poolUnwind):
t.Fatal("SIGTERM did not cancel the scan's context")
}
}
// TestCommitFullBatchKeepsFailedBatch checks that a full batch whose
// commit fails, as it does once the scan is interrupted, stays in the
// batch, so that syncScan's final commit saves it.
func TestCommitFullBatchKeepsFailedBatch(t *testing.T) {
t.Parallel()
s := &scanState{db: openTestDB(t)}
for i := range updateBatchSize {
s.batch = append(s.batch, scanRec{path: "/f" + strconv.Itoa(i)})
}
err := s.commitFullBatch(cancelledContext(t))
if !errors.Is(err, context.Canceled) {
t.Fatalf("commitFullBatch on a cancelled context = %v, want %v",
err, context.Canceled)
}
if len(s.batch) != updateBatchSize {
t.Errorf("batch holds %d records after the failed commit, want %d",
len(s.batch), updateBatchSize)
}
}
// drainClosed counts the values received from ch until it closes, // drainClosed counts the values received from ch until it closes,
// failing the test if it does not close within poolUnwind. A pool that // failing the test if it does not close within poolUnwind. A pool that
// ignored its cancellation leaves its channel open with its goroutines // ignored its cancellation leaves its channel open with its goroutines
+6
View File
@@ -87,6 +87,9 @@ func run(args []string, stdout, stderr io.Writer) int {
switch { switch {
case err == nil: case err == nil:
return exitOK return exitOK
case errors.Is(err, errInterrupted):
// The interrupted scan has printed its own line.
return exitFatal
case errors.As(err, &fatal): case errors.As(err, &fatal):
// The command ran and failed: a runtime error, reported // The command ran and failed: a runtime error, reported
// without the usage text that a usage error gets. // without the usage text that a usage error gets.
@@ -132,6 +135,9 @@ func newRootCommand(stdout, stderr io.Writer) *cobra.Command {
Short: "Walk trees and synchronize the scan database", Short: "Walk trees and synchronize the scan database",
Args: cobra.MinimumNArgs(1), Args: cobra.MinimumNArgs(1),
RunE: runE(func(ctx context.Context, args []string) error { RunE: runE(func(ctx context.Context, args []string) error {
ctx, stop := interruptContext(ctx)
defer stop()
return runScan(ctx, args, scanWorkers, scanOneFS) return runScan(ctx, args, scanWorkers, scanOneFS)
}), }),
} }
+7 -1
View File
@@ -141,14 +141,20 @@ func (p *progress) warnf(format string, args ...any) {
fmt.Fprintln(os.Stderr, msg) fmt.Fprintln(os.Stderr, msg)
} }
// finish terminates the pass's display. // finish terminates the pass's display. A bar whose pass stopped short
// of its total, as an interrupted one does, is left as last drawn; the
// library's Finish would fill it up.
func (p *progress) finish() { func (p *progress) finish() {
if p == nil { if p == nil {
return return
} }
if p.bar != nil { if p.bar != nil {
if p.total >= 0 && p.count < p.total {
_ = p.bar.Exit()
} else {
_ = p.bar.Finish() _ = p.bar.Finish()
}
fmt.Fprintln(os.Stderr) fmt.Fprintln(os.Stderr)
+34 -6
View File
@@ -144,18 +144,46 @@ func TestSpinnerShowsCountAfterBurst(t *testing.T) {
time.Sleep(spinnerIdle) time.Sleep(spinnerIdle)
// A terminal shows the last frame drawn. The library starts each if shown := lastFrame(stderr()); !strings.Contains(shown, "(50/-,") {
// frame with a carriage return and erases the previous one with t.Errorf("terminal shows %q, want a count of 50", shown)
// spaces first. }
}
// lastFrame returns what a terminal shows of the frames a bar drew: the
// last one. The library starts each frame with a carriage return and
// erases the previous one with spaces first.
func lastFrame(out string) string {
var shown string var shown string
for frame := range strings.SplitSeq(stderr(), "\r") { for frame := range strings.SplitSeq(out, "\r") {
if strings.TrimSpace(frame) != "" { if strings.TrimSpace(frame) != "" {
shown = frame shown = frame
} }
} }
if !strings.Contains(shown, "(50/-,") { return shown
t.Errorf("terminal shows %q, want a count of 50", shown) }
// TestBarStoppedShortKeepsCount checks that the terminal display of a
// pass that stops before its total, as an interrupted one does, is left
// as last drawn instead of being filled up.
//
//nolint:paralleltest // captureStderr replaces the process-wide os.Stderr
func TestBarStoppedShortKeepsCount(t *testing.T) {
stderr := captureStderr(t)
p := &progress{
label: "hash", total: 10, start: time.Now(),
bar: newBar("hash", 10),
}
p.increment()
// Past the redraw limit, so the bar draws the next count.
time.Sleep(2 * barThrottle)
p.increment()
p.finish()
if shown := lastFrame(stderr()); !strings.Contains(shown, "(2/10,") {
t.Errorf("terminal shows %q, want a count of 2 of 10", shown)
} }
} }
+97 -22
View File
@@ -11,6 +11,7 @@ import (
"io" "io"
"io/fs" "io/fs"
"os" "os"
"os/signal"
"path/filepath" "path/filepath"
"slices" "slices"
"strings" "strings"
@@ -57,6 +58,10 @@ const sampleWindow = 1024 * 1024
// and hash worker pools. // and hash worker pools.
const workQueueDepth = 1024 const workQueueDepth = 1024
// errInterrupted reports a scan stopped by SIGINT or SIGTERM. runScan
// has already printed its line, so run prints nothing more.
var errInterrupted = errors.New("scan interrupted")
// fileRec carries one statted file between the scan phases. dev and // fileRec carries one statted file between the scan phases. dev and
// ino identify the underlying inode so hard-linked paths can share // ino identify the underlying inode so hard-linked paths can share
// one read; both are zero when the platform exposes no inode. // one read; both are zero when the platform exposes no inode.
@@ -90,8 +95,10 @@ type fileMeta struct {
// scan fails before it walks the filesystem or opens the database. // scan fails before it walks the filesystem or opens the database.
// Errors are returned rather than exiting, so that the deferred close — // Errors are returned rather than exiting, so that the deferred close —
// which takes the database out of WAL mode — always runs, and the lock // which takes the database out of WAL mode — always runs, and the lock
// is released after it. Cancelling ctx unwinds the worker pools and // is released after it. When ctx is cancelled, as by the SIGINT or
// aborts the scan with the context's error. // SIGTERM that interruptContext catches, the scan keeps what it has
// hashed (see syncScan), prints how many files its walk reached, and
// returns errInterrupted.
func runScan(ctx context.Context, roots []string, workers int, func runScan(ctx context.Context, roots []string, workers int,
oneFS bool, oneFS bool,
) error { ) error {
@@ -114,6 +121,12 @@ func runScan(ctx context.Context, roots []string, workers int,
defer func() { _ = lock.Close() }() defer func() { _ = lock.Close() }()
db, err := openScanDatabase(ctx, dbPath) db, err := openScanDatabase(ctx, dbPath)
if err != nil && ctx.Err() != nil {
// Interrupted while opening; SQLite may report that with an
// error of its own rather than the context's.
return interrupted(0)
}
if err != nil { if err != nil {
return err return err
} }
@@ -121,6 +134,10 @@ func runScan(ctx context.Context, roots []string, workers int,
defer closeScanDatabase(ctx, db, dbPath) defer closeScanDatabase(ctx, db, dbPath)
st, err := syncScan(ctx, db, roots, workers, oneFS) st, err := syncScan(ctx, db, roots, workers, oneFS)
if errors.Is(err, context.Canceled) {
return interrupted(st.walked)
}
if err != nil { if err != nil {
return fmt.Errorf("update database %s: %w", dbPath, err) return fmt.Errorf("update database %s: %w", dbPath, err)
} }
@@ -134,6 +151,34 @@ func runScan(ctx context.Context, roots []string, workers int,
return nil return nil
} }
// interruptContext returns a copy of ctx that the first SIGINT or
// SIGTERM cancels; the scan command runs the scan under it. stop
// releases the signals.
func interruptContext(ctx context.Context) (context.Context, func()) {
// A SIGINT ignored from the start, as by a script's background job,
// stays ignored.
signals := []os.Signal{syscall.SIGTERM}
if !signal.Ignored(syscall.SIGINT) {
signals = append(signals, syscall.SIGINT)
}
ctx, stop := signal.NotifyContext(ctx, signals...)
// Stopping restores the default handling, so a second signal ends
// the process at once.
context.AfterFunc(ctx, stop)
return ctx, stop
}
// interrupted prints the line for a scan stopped by a signal after its
// walk reached walked files, and returns errInterrupted.
func interrupted(walked int) error {
fmt.Fprintf(os.Stderr, "scan: interrupted after %d files\n", walked)
return errInterrupted
}
// resolveRoots converts each PATH operand to an absolute, lexically // resolveRoots converts each PATH operand to an absolute, lexically
// cleaned path (symlinks are not resolved) and verifies that it // cleaned path (symlinks are not resolved) and verifies that it
// exists. Database records are keyed by absolute path, so scan results // exists. Database records are keyed by absolute path, so scan results
@@ -187,8 +232,9 @@ func pruneRoots(roots []string) []string {
} }
// scanStats summarizes one scan's database synchronization for the // scanStats summarizes one scan's database synchronization for the
// final stderr summary. // final stderr summary, or for the line an interrupted scan prints.
type scanStats struct { type scanStats struct {
walked int // files the walk reached
added int added int
updated int updated int
removed int removed int
@@ -210,7 +256,30 @@ type scanState struct {
st scanStats st scanStats
} }
// syncScan synchronizes the database with the filesystem under roots // syncScan synchronizes the database with the filesystem under roots;
// see runPhases. When ctx is cancelled, as by an interrupt, it commits
// the hashed records still waiting in the batch, starts no other write
// or deletion, and returns the cancellation.
func syncScan(ctx context.Context, db *sql.DB, roots []string,
workers int, oneFS bool,
) (scanStats, error) {
s := &scanState{db: db}
err := s.runPhases(ctx, roots, workers, oneFS)
if err == nil || ctx.Err() == nil {
return s.st, err
}
// The one write made after the cancellation, so it cannot use ctx.
err = applyChanges(context.WithoutCancel(ctx), db, s.batch, nil, nil)
if err != nil {
return s.st, err
}
return s.st, ctx.Err()
}
// runPhases synchronizes the database with the filesystem under roots
// in four sequential phases: walk (enumerate and stat every file, // in four sequential phases: walk (enumerate and stat every file,
// building a complete size census), hash (read only the new or // building a complete size census), hash (read only the new or
// changed — or previously unhashed — files whose size at least one // changed — or previously unhashed — files whose size at least one
@@ -223,48 +292,41 @@ type scanState struct {
// their content hash. Operands the walk cannot start from are dropped // their content hash. Operands the walk cannot start from are dropped
// first, so the records beneath them count as outside the roots unless // first, so the records beneath them count as outside the roots unless
// they lie under another root. // they lie under another root.
func syncScan(ctx context.Context, db *sql.DB, roots []string, func (s *scanState) runPhases(ctx context.Context, roots []string,
workers int, oneFS bool, workers int, oneFS bool,
) (scanStats, error) { ) error {
s := &scanState{db: db}
// Types are checked before pruning so that an operand under a // Types are checked before pruning so that an operand under a
// dropped one is still scanned, not dropped as lying under it. // dropped one is still scanned, not dropped as lying under it.
roots = pruneRoots(s.walkableRoots(roots)) roots = pruneRoots(s.walkableRoots(roots))
err := s.loadIndex(ctx, roots) err := s.loadIndex(ctx, roots)
if err != nil { if err != nil {
return s.st, err return err
} }
changed, unhashed := s.walkPhase(startWalk(ctx, roots, oneFS, workers)) changed, unhashed := s.walkPhase(startWalk(ctx, roots, oneFS, workers))
// A cancelled walk stops early, so its size census covers only part // A cancelled walk stops early, so its size census covers only part
// of the roots, and every file it never reached looks vanished to // of the roots, and every file it never reached would look vanished
// the update phase. Defence in depth rather than the only barrier: // to the update phase. Stop before anything is written or deleted.
// that phase would today fail on its first BeginTx with the same
// cancelled context before deleting anything. But it is the barrier
// that survives a later decision to let an interrupted scan commit
// what it has, and it turns a confusing failure deep in the update
// phase into a clean abort at the phase boundary.
err = ctx.Err() err = ctx.Err()
if err != nil { if err != nil {
return s.st, err return err
} }
s.partition(changed, unhashed) s.partition(changed, unhashed)
err = s.hashPhase(ctx, workers) err = s.hashPhase(ctx, workers)
if err != nil { if err != nil {
return s.st, err return err
} }
err = s.updatePhase(ctx) err = s.updatePhase(ctx)
if err != nil { if err != nil {
return s.st, err return err
} }
return s.st, s.contentPhase(ctx, workers) return s.contentPhase(ctx, workers)
} }
// walkableRoots returns the operands the walk can start from: regular // walkableRoots returns the operands the walk can start from: regular
@@ -349,6 +411,7 @@ func (s *scanState) walkPhase(
} }
s.sizes = append(s.sizes, ev.rec.size) s.sizes = append(s.sizes, ev.rec.size)
s.st.walked++
prog.increment() prog.increment()
@@ -552,16 +615,22 @@ func (s *scanState) recordRun(ctx context.Context, r hashResult) error {
} }
// commitFullBatch commits the running batch once it holds // commitFullBatch commits the running batch once it holds
// updateBatchSize records. // updateBatchSize records. A batch that fails to commit is kept: the
// commit fails when the scan is interrupted, and syncScan then commits
// the batch itself.
func (s *scanState) commitFullBatch(ctx context.Context) error { func (s *scanState) commitFullBatch(ctx context.Context) error {
if len(s.batch) < updateBatchSize { if len(s.batch) < updateBatchSize {
return nil return nil
} }
err := applyBatch(ctx, s.db, s.batch, nil, nil) err := applyBatch(ctx, s.db, s.batch, nil, nil)
if err != nil {
return err
}
s.batch = s.batch[:0] s.batch = s.batch[:0]
return err return nil
} }
// updatePhase writes the scan's tail under one progress display: the // updatePhase writes the scan's tail under one progress display: the
@@ -1004,6 +1073,12 @@ func walkOneDir(ctx context.Context, job dirJob, oneFS bool,
var subs []dirJob var subs []dirJob
for _, e := range entries { for _, e := range entries {
// A cancelled scan wants nothing more from this directory: stop
// rather than lstat the rest of a large one.
if ctx.Err() != nil {
return nil
}
p := filepath.Join(job.path, e.Name()) p := filepath.Join(job.path, e.Name())
if e.IsDir() { if e.IsDir() {
+25 -16
View File
@@ -9,11 +9,13 @@ import (
"fmt" "fmt"
"io" "io"
"os" "os"
"os/signal"
"path/filepath" "path/filepath"
"runtime" "runtime"
"slices" "slices"
"strconv" "strconv"
"strings" "strings"
"syscall"
"testing" "testing"
"time" "time"
) )
@@ -456,7 +458,7 @@ func TestScanContentWithinOperand(t *testing.T) {
added := sparseFile(t, dir, "d2", headTailMin) added := sparseFile(t, dir, "d2", headTailMin)
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: 1, unchanged: 2}) { if st != (scanStats{walked: 3, added: 1, unchanged: 2}) {
t.Fatalf("rescan stats = %+v, want 1 added 2 unchanged", st) t.Fatalf("rescan stats = %+v, want 1 added 2 unchanged", st)
} }
@@ -502,7 +504,7 @@ func TestScanContentStalePartners(t *testing.T) {
sparseFile(t, dirB, "changed-copy", headTailMin+1) sparseFile(t, dirB, "changed-copy", headTailMin+1)
st := syncTree(t, db, dirB) st := syncTree(t, db, dirB)
if st != (scanStats{added: 2}) { if st != (scanStats{walked: 2, added: 2}) {
t.Errorf("stats = %+v, want 2 added and nothing skipped", st) t.Errorf("stats = %+v, want 2 added and nothing skipped", st)
} }
@@ -559,7 +561,7 @@ func TestScanContentHashedStalePartners(t *testing.T) {
b := sparseFile(t, t.TempDir(), "copy", headTailMin) b := sparseFile(t, t.TempDir(), "copy", headTailMin)
st := syncTree(t, db, filepath.Dir(b)) st := syncTree(t, db, filepath.Dir(b))
if st != (scanStats{added: 1}) { if st != (scanStats{walked: 1, added: 1}) {
t.Errorf("stats = %+v, want 1 added and nothing skipped", st) t.Errorf("stats = %+v, want 1 added and nothing skipped", st)
} }
@@ -599,7 +601,7 @@ func TestScanContentReadFailure(t *testing.T) {
b := sparseFile(t, dirB, "b", headTailMin) b := sparseFile(t, dirB, "b", headTailMin)
st := syncTree(t, db, dirB) st := syncTree(t, db, dirB)
if st != (scanStats{added: 1, skipped: 1}) { if st != (scanStats{walked: 1, added: 1, skipped: 1}) {
t.Fatalf("stats = %+v, want 1 added 1 skipped", st) t.Fatalf("stats = %+v, want 1 added 1 skipped", st)
} }
@@ -613,7 +615,7 @@ func TestScanContentReadFailure(t *testing.T) {
} }
st = syncTree(t, db, dirB) st = syncTree(t, db, dirB)
if st != (scanStats{unchanged: 1}) { if st != (scanStats{walked: 1, unchanged: 1}) {
t.Fatalf("rescan stats = %+v, want 1 unchanged", st) t.Fatalf("rescan stats = %+v, want 1 unchanged", st)
} }
@@ -659,7 +661,7 @@ func TestScanContentCheckError(t *testing.T) {
b := sparseFile(t, t.TempDir(), "b", headTailMin) b := sparseFile(t, t.TempDir(), "b", headTailMin)
st := syncTree(t, db, filepath.Dir(b)) st := syncTree(t, db, filepath.Dir(b))
if st != (scanStats{added: 1, skipped: 1}) { if st != (scanStats{walked: 1, added: 1, skipped: 1}) {
t.Fatalf("stats = %+v, want 1 added 1 skipped", st) t.Fatalf("stats = %+v, want 1 added 1 skipped", st)
} }
@@ -687,7 +689,7 @@ func TestScanContentHardlinks(t *testing.T) {
c := sparseFile(t, dir, "copy", headTailMin) c := sparseFile(t, dir, "copy", headTailMin)
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: 3}) { if st != (scanStats{walked: 3, added: 3}) {
t.Fatalf("stats = %+v, want 3 added", st) t.Fatalf("stats = %+v, want 3 added", st)
} }
@@ -1058,7 +1060,7 @@ func TestScanPipeline(t *testing.T) {
db := openTestDB(t) db := openTestDB(t)
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: smokeTreeFiles}) { if st != (scanStats{walked: smokeTreeFiles, added: smokeTreeFiles}) {
t.Fatalf("stats = %+v, want %d added only", st, smokeTreeFiles) t.Fatalf("stats = %+v, want %d added only", st, smokeTreeFiles)
} }
@@ -1081,7 +1083,7 @@ func TestSyncScanUnchangedReuse(t *testing.T) {
writeFile(t, dir, "b.bin", pattern(2, 600)) writeFile(t, dir, "b.bin", pattern(2, 600))
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: 2}) { if st != (scanStats{walked: 2, added: 2}) {
t.Fatalf("first scan stats = %+v, want 2 added", st) t.Fatalf("first scan stats = %+v, want 2 added", st)
} }
@@ -1095,7 +1097,7 @@ func TestSyncScanUnchangedReuse(t *testing.T) {
} }
st = syncTree(t, db, dir) st = syncTree(t, db, dir)
if st != (scanStats{unchanged: 2}) { if st != (scanStats{walked: 2, unchanged: 2}) {
t.Fatalf("rescan stats = %+v, want 2 unchanged", st) t.Fatalf("rescan stats = %+v, want 2 unchanged", st)
} }
@@ -1124,7 +1126,7 @@ func TestSyncScanMtimeBump(t *testing.T) {
} }
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{updated: 1}) { if st != (scanStats{walked: 1, updated: 1}) {
t.Fatalf("mtime-bump stats = %+v, want 1 updated", st) t.Fatalf("mtime-bump stats = %+v, want 1 updated", st)
} }
@@ -1152,7 +1154,7 @@ func TestSyncScanAddRemove(t *testing.T) {
} }
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: 1, removed: 1, unchanged: 1}) { if st != (scanStats{walked: 2, added: 1, removed: 1, unchanged: 1}) {
t.Fatalf("add/remove stats = %+v, want 1 added 1 removed 1 unchanged", t.Fatalf("add/remove stats = %+v, want 1 added 1 removed 1 unchanged",
st) st)
} }
@@ -1269,7 +1271,7 @@ func TestSyncScanOverlappingRoots(t *testing.T) {
// A file reachable via two overlapping operands is deduplicated // A file reachable via two overlapping operands is deduplicated
// by path in the shared walk and processed once. // 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 != (scanStats{added: 1}) { if st != (scanStats{walked: 1, added: 1}) {
t.Fatalf("stats = %+v, want 1 added", st) t.Fatalf("stats = %+v, want 1 added", st)
} }
@@ -1290,7 +1292,7 @@ func TestScanSkipsUniqueSizes(t *testing.T) {
// Neither size is shared, so neither file is read: both records // Neither size is shared, so neither file is read: both records
// are written without hashes and no duplicates are reported. // are written without hashes and no duplicates are reported.
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: 2}) { if st != (scanStats{walked: 2, added: 2}) {
t.Fatalf("stats = %+v, want 2 added", st) t.Fatalf("stats = %+v, want 2 added", st)
} }
@@ -1312,7 +1314,7 @@ func TestScanSkipsUniqueSizes(t *testing.T) {
c := writeFile(t, dir, "c.bin", pattern(1, 500)) c := writeFile(t, dir, "c.bin", pattern(1, 500))
st = syncTree(t, db, dir) st = syncTree(t, db, dir)
if st != (scanStats{added: 1, updated: 1, unchanged: 1}) { if st != (scanStats{walked: 3, added: 1, updated: 1, unchanged: 1}) {
t.Fatalf("rescan stats = %+v, want 1 added 1 updated 1 unchanged", t.Fatalf("rescan stats = %+v, want 1 added 1 updated 1 unchanged",
st) st)
} }
@@ -1376,7 +1378,7 @@ func TestScanHardlinksReadOnce(t *testing.T) {
} }
st := syncTree(t, db, dir) st := syncTree(t, db, dir)
if st != (scanStats{added: 2}) { if st != (scanStats{walked: 2, added: 2}) {
t.Fatalf("stats = %+v, want 2 added", st) t.Fatalf("stats = %+v, want 2 added", st)
} }
@@ -1498,6 +1500,13 @@ func injectWriteFailure(t *testing.T, path string) {
func baselineGoroutines(t *testing.T) int { func baselineGoroutines(t *testing.T) int {
t.Helper() t.Helper()
// The first scan command in a process starts os/signal's goroutine,
// which never exits. Start it now, so the baseline counts it
// instead of the scan seeming to leave it behind.
ch := make(chan os.Signal, 1)
signal.Notify(ch, syscall.SIGINT)
signal.Stop(ch)
deadline := time.Now().Add(goroutineSettle) deadline := time.Now().Add(goroutineSettle)
last := runtime.NumGoroutine() last := runtime.NumGoroutine()