Compare commits
3
Commits
2f339fc53a
...
6d1cf717f2
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6d1cf717f2 | ||
|
|
7ff0dbb6e1 | ||
|
|
bccc14ffef |
@@ -329,7 +329,12 @@ All three subcommands operate on a single SQLite database file:
|
||||
`-shm` files beside it, which SQLite creates with the database
|
||||
file's permissions.
|
||||
- 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. `scan` creates
|
||||
the schema and sets the version in one transaction, so a first scan
|
||||
stopped while doing so leaves an empty database the next scan sets
|
||||
up. A database at version 0 that already has a `files` table was
|
||||
therefore not made by sfdupes; every subcommand refuses it with an
|
||||
error telling the user to remove the file and rescan):
|
||||
|
||||
```sql
|
||||
CREATE TABLE files (
|
||||
|
||||
@@ -29,6 +29,17 @@
|
||||
|
||||
# Completed Steps
|
||||
|
||||
- correct four inaccurate comments in `cancel_test.go` and rename
|
||||
`walkCancelInFlightDirs` to `walkCancelInFlightFiles` (2026-10-04,
|
||||
https://git.eeqj.de/sneak/sfdupes/issues/33)
|
||||
|
||||
- test the `-x` filesystem-boundary rules in `subdirJob` (2026-10-04,
|
||||
https://git.eeqj.de/sneak/sfdupes/issues/17)
|
||||
|
||||
- `scan` creates the schema in one transaction; a version-0 database with a
|
||||
`files` table is refused with a clear schema-version error (2026-10-04,
|
||||
https://git.eeqj.de/sneak/sfdupes/issues/11)
|
||||
|
||||
- README documents install, Docker, a daily cron scan and how to read
|
||||
and check the reports (2026-10-04,
|
||||
https://git.eeqj.de/sneak/sfdupes/issues/54)
|
||||
|
||||
+56
-15
@@ -18,11 +18,21 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// poolUnwind bounds how long a goroutine is given to leave a pool
|
||||
// after its context is cancelled. Only a failing run ever waits this
|
||||
// long: a pool that ignored its cancellation parks forever, and this
|
||||
// is what turns that into a failed assertion instead of a suite that
|
||||
// hangs until the test binary's own timeout.
|
||||
// This file gathers the tests for scan cancellation and worker-pool
|
||||
// unwinding. Everything it exercises lives in scan.go, so by the repo's
|
||||
// convention of one test file per source file it would belong in
|
||||
// scan_test.go. It is kept separate on purpose: cancellation behaviour
|
||||
// cuts across both the walk pool and the hash pool as a single concern,
|
||||
// and scan_test.go is already over 1,600 lines. That is the deliberate
|
||||
// exception the convention otherwise expects to be stated.
|
||||
|
||||
// poolUnwind bounds how long a test waits for a cancellation to take
|
||||
// effect: for a goroutine to return or a channel to close once its
|
||||
// context is cancelled, or for a signal to cancel the scan's context.
|
||||
// Only a failing run waits this long, and the bound is what makes that
|
||||
// failure an assertion instead of a hang. A call made without it, as
|
||||
// most of this file's scans are, has no bound: a regression that parks
|
||||
// it is caught only as the test binary's own timeout.
|
||||
const poolUnwind = 2 * time.Second
|
||||
|
||||
// walkClock is a context whose cancellation is driven by the scan's
|
||||
@@ -33,9 +43,14 @@ const poolUnwind = 2 * time.Second
|
||||
//
|
||||
// The accounting behind the n chosen by each test: every blocking
|
||||
// channel operation in the walk selects on Done, so the walk spends
|
||||
// one consultation per file event plus a couple per directory, while
|
||||
// the index load that runs ahead of it spends a small fixed number
|
||||
// (three) whatever the record count.
|
||||
// one consultation per file event plus a couple per directory. The
|
||||
// index load that runs ahead of it also consults Done, but a bounded
|
||||
// number of times that does not grow with the record count. The tests
|
||||
// depend on that property, not on the bound's exact value: each test
|
||||
// sets n from the consultations of the walk, plus those of the hash
|
||||
// phase when it cancels mid-hash, far from both ends of the phase it
|
||||
// interrupts, so the cancellation lands inside that phase whatever the
|
||||
// record count.
|
||||
type walkClock struct {
|
||||
n int64
|
||||
seen atomic.Int64
|
||||
@@ -95,7 +110,10 @@ const (
|
||||
walkCancelFilesPerDir = 20
|
||||
walkCancelFiles = walkCancelDirs * walkCancelFilesPerDir
|
||||
walkCancelWorkers = 4
|
||||
walkCancelInFlightDirs = walkCancelWorkers * walkCancelFilesPerDir
|
||||
// The most files the walkCancelWorkers directories already in
|
||||
// flight when the scan is cancelled can still emit, at
|
||||
// walkCancelFilesPerDir each. A file count, not a directory count.
|
||||
walkCancelInFlightFiles = walkCancelWorkers * walkCancelFilesPerDir
|
||||
)
|
||||
|
||||
// walkCancelAtDone is the consultation on which the fixture's context
|
||||
@@ -158,6 +176,11 @@ func assertRecordsIntact(t *testing.T, db *sql.DB, before []string) {
|
||||
// unreachable, makes the scan carry its truncated view into the update
|
||||
// phase, which counts every record the walk never reached for removal.
|
||||
//
|
||||
// The syncScan call here is not bounded by poolUnwind: a regression
|
||||
// that left a worker pool parked would hang it, and that regression is
|
||||
// caught only by the test binary's own timeout, not by a quick
|
||||
// assertion.
|
||||
//
|
||||
//nolint:paralleltest // counts goroutines: must not run beside others
|
||||
func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) {
|
||||
dir := buildWalkCancelTree(t)
|
||||
@@ -209,10 +232,11 @@ func assertWalkGuardAborted(t *testing.T, st scanStats, err error) {
|
||||
}
|
||||
|
||||
// The workers drop every directory still queued once the scan is
|
||||
// cancelled, so only the directories already in flight can add to
|
||||
// the census after the fact. A census beyond that bound would mean
|
||||
// the cancellation was not observed where it should have been.
|
||||
limit := walkCancelAtDone + walkCancelInFlightDirs
|
||||
// cancelled, so only the files in the directories already in flight
|
||||
// can add to the census after the fact. A census beyond that bound
|
||||
// would mean the cancellation was not observed where it should have
|
||||
// been.
|
||||
limit := walkCancelAtDone + walkCancelInFlightFiles
|
||||
if st.unchanged > limit {
|
||||
t.Errorf("census covers %d files, want at most %d: the walk kept "+
|
||||
"taking directories off the queue after cancellation",
|
||||
@@ -646,7 +670,11 @@ func TestDispatchDirsClosesJobsWhenCancelled(t *testing.T) {
|
||||
|
||||
// TestFeedHashJobsClosesJobsWhenCancelled checks that the hash feeder
|
||||
// abandons the runs it has not queued yet and still closes the job
|
||||
// channel, which is what lets the workers' range terminate.
|
||||
// channel, which is what lets the workers' range terminate. The
|
||||
// receive on jobs below is not bounded: a feeder that returned without
|
||||
// closing jobs would leave that receive with no sender and no close, so
|
||||
// this regression is caught by the test binary's timeout rather than by
|
||||
// a bounded assertion.
|
||||
func TestFeedHashJobsClosesJobsWhenCancelled(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -674,6 +702,16 @@ func TestFeedHashJobsClosesJobsWhenCancelled(t *testing.T) {
|
||||
// nobody wants the hashes of — while still letting the range run out
|
||||
// so the pool tears down. The queued run names a file that does not
|
||||
// exist, so a worker that hashed it anyway would produce a result.
|
||||
//
|
||||
// hashWorker's other cancellation exit, abandoning the send of a
|
||||
// result, is reachable from the scan: stop cancels the pool before it
|
||||
// drains results, so a worker waiting on that send can leave through
|
||||
// it. TestScanHashWriteFailureUnwindsPool reaches it in some runs only,
|
||||
// depending on whether the workers have filled the results channel when
|
||||
// stop runs, and no test fails without it, since stop's drain frees a
|
||||
// waiting worker anyway. This test, for its part, catches a removed
|
||||
// drop check in some runs only: a worker that hashes the run anyway
|
||||
// then picks at random between sending the result and leaving.
|
||||
func TestHashWorkerDropsQueuedRuns(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -706,7 +744,10 @@ func TestHashWorkerDropsQueuedRuns(t *testing.T) {
|
||||
// TestHashPhaseCancelledReturnsContextError checks the result loop's
|
||||
// own exit: with the pool cancelled, no result will ever arrive, and
|
||||
// the loop must leave through the cancellation rather than wait for a
|
||||
// receive that cannot happen.
|
||||
// receive that cannot happen. This call is not bounded by poolUnwind: a
|
||||
// loop that dropped its cancellation case would block on that receive,
|
||||
// so the regression surfaces as the test binary's timeout rather than
|
||||
// as a bounded assertion.
|
||||
func TestHashPhaseCancelledReturnsContextError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -219,6 +219,12 @@ func openReportDatabase(ctx context.Context,
|
||||
}
|
||||
|
||||
v, err := userVersion(ctx, db)
|
||||
if err == nil && v == 0 {
|
||||
// An empty database passes this check and fails the version
|
||||
// check below.
|
||||
err = checkUnversioned(ctx, db)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
_ = db.Close()
|
||||
|
||||
@@ -245,6 +251,11 @@ func initSchema(ctx context.Context, db *sql.DB) error {
|
||||
|
||||
switch v {
|
||||
case 0:
|
||||
err = checkUnversioned(ctx, db)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return createSchema(ctx, db)
|
||||
case schemaVersion:
|
||||
return nil
|
||||
@@ -254,25 +265,65 @@ func initSchema(ctx context.Context, db *sql.DB) error {
|
||||
}
|
||||
}
|
||||
|
||||
// checkUnversioned checks a database at user_version 0 before it is
|
||||
// taken for an empty one. createSchema creates the files table and
|
||||
// sets the version together, so a files table at version 0 was made by
|
||||
// something else. Adopting it could corrupt unrelated data, so that is
|
||||
// a schema-version error telling the operator to remove the file and
|
||||
// rescan.
|
||||
func checkUnversioned(ctx context.Context, db *sql.DB) error {
|
||||
var name string
|
||||
|
||||
err := db.QueryRowContext(ctx,
|
||||
"SELECT name FROM sqlite_master "+
|
||||
"WHERE type = 'table' AND name = 'files'").Scan(&name)
|
||||
|
||||
switch {
|
||||
case err == nil:
|
||||
return fmt.Errorf(
|
||||
"has a files table but no schema version; "+
|
||||
"remove the file and rescan: %w", errSchemaVersion)
|
||||
case errors.Is(err, sql.ErrNoRows):
|
||||
return nil
|
||||
default:
|
||||
return fmt.Errorf("check for files table: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
// createSchema applies the schema to a fresh database and stamps the
|
||||
// schema version.
|
||||
// schema version in one transaction, so a creation stopped partway, by
|
||||
// an interrupt or an error, leaves an empty database the next scan
|
||||
// sets up, never a files table at version 0, which checkUnversioned
|
||||
// refuses.
|
||||
func createSchema(ctx context.Context, db *sql.DB) error {
|
||||
_, err := db.ExecContext(ctx, createTableSQL)
|
||||
tx, err := db.BeginTx(ctx, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create schema: %w", err)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(ctx, createIndexSQL)
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
|
||||
_, err = tx.ExecContext(ctx, createTableSQL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create schema: %w", err)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(ctx,
|
||||
_, err = tx.ExecContext(ctx, createIndexSQL)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create schema: %w", err)
|
||||
}
|
||||
|
||||
_, err = tx.ExecContext(ctx,
|
||||
"PRAGMA user_version = "+strconv.Itoa(schemaVersion))
|
||||
if err != nil {
|
||||
return fmt.Errorf("set schema version: %w", err)
|
||||
}
|
||||
|
||||
err = tx.Commit()
|
||||
if err != nil {
|
||||
return fmt.Errorf("create schema: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
+71
@@ -8,6 +8,7 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
@@ -77,6 +78,76 @@ func TestOpenScanDatabaseCreates(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenDatabaseUnversionedForeign(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// A database that has a files table but user_version 0, written by
|
||||
// some other tool. report, trees and scan must refuse it with the
|
||||
// schema-version error, not adopt it and not emit a raw SQLite
|
||||
// "table files already exists".
|
||||
path := testDBPath(t)
|
||||
|
||||
db, err := sql.Open("sqlite", path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_, err = db.ExecContext(t.Context(), "CREATE TABLE files (x INTEGER)")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
_ = db.Close()
|
||||
|
||||
_, err = openReportDatabase(t.Context(), path)
|
||||
if !errors.Is(err, errSchemaVersion) ||
|
||||
!strings.Contains(err.Error(), "remove the file and rescan") {
|
||||
t.Fatalf("report: err = %v, want errSchemaVersion telling the "+
|
||||
"operator to remove the file and rescan", err)
|
||||
}
|
||||
|
||||
_, err = openScanDatabase(t.Context(), path)
|
||||
if !errors.Is(err, errSchemaVersion) ||
|
||||
!strings.Contains(err.Error(), "remove the file and rescan") {
|
||||
t.Fatalf("scan: err = %v, want errSchemaVersion telling the "+
|
||||
"operator to remove the file and rescan", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchemaCreationStoppedPartway(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
// A first scan stopped while creating the schema must leave a
|
||||
// database the next scan accepts. max_page_count(2) leaves room for
|
||||
// the files table but not its index, so schema creation fails right
|
||||
// after CREATE TABLE, a point an interrupt could also stop it at.
|
||||
path := testDBPath(t)
|
||||
|
||||
db, err := openDB(path, scanParams+"&_pragma=max_page_count(2)")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = initSchema(t.Context(), db)
|
||||
_ = db.Close()
|
||||
|
||||
if err == nil {
|
||||
t.Fatal("initSchema with no room for the index succeeded")
|
||||
}
|
||||
|
||||
db, err = openScanDatabase(t.Context(), path)
|
||||
if err != nil {
|
||||
t.Fatalf("next scan: %v", err)
|
||||
}
|
||||
|
||||
defer func() { _ = db.Close() }()
|
||||
|
||||
v, err := userVersion(t.Context(), db)
|
||||
if err != nil || v != schemaVersion {
|
||||
t.Fatalf("userVersion = %d, %v; want %d, nil", v, err, schemaVersion)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOpenReportDatabaseMissing(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
+155
@@ -6,8 +6,10 @@ import (
|
||||
"crypto/sha256"
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/fs"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
@@ -912,6 +914,159 @@ func TestDeviceOfInfo(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// dirEntryFor returns the fs.DirEntry for name within dir, obtained via
|
||||
// the same os.ReadDir the walk uses, so it carries a real Info().
|
||||
func dirEntryFor(t *testing.T, dir, name string) fs.DirEntry {
|
||||
t.Helper()
|
||||
|
||||
entries, err := os.ReadDir(dir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
for _, e := range entries {
|
||||
if e.Name() == name {
|
||||
return e
|
||||
}
|
||||
}
|
||||
|
||||
t.Fatalf("entry %q not found in %q", name, dir)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// callSubdirJob runs subdirJob against e under parent, collecting any
|
||||
// warning events it emits (subdirJob emits at most one).
|
||||
func callSubdirJob(t *testing.T, p string, e fs.DirEntry,
|
||||
parent dirJob, oneFS bool,
|
||||
) (dirJob, bool, []walkEvent) {
|
||||
t.Helper()
|
||||
|
||||
events := make(chan walkEvent, 1)
|
||||
job, ok := subdirJob(t.Context(), p, e, parent, oneFS, events)
|
||||
close(events)
|
||||
|
||||
var evs []walkEvent
|
||||
for ev := range events {
|
||||
evs = append(evs, ev)
|
||||
}
|
||||
|
||||
return job, ok, evs
|
||||
}
|
||||
|
||||
// TestSubdirJobOneFilesystem exercises the -x boundary check in
|
||||
// subdirJob directly, so no second real filesystem is needed. The
|
||||
// subdirectory's real device is compared against a fabricated operand
|
||||
// device.
|
||||
func TestSubdirJobOneFilesystem(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
dir := t.TempDir()
|
||||
|
||||
sub := filepath.Join(dir, "sub")
|
||||
|
||||
err := os.Mkdir(sub, 0o750)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
info, err := os.Lstat(sub)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
dev, ok := deviceOfInfo(info)
|
||||
if !ok {
|
||||
t.Skip("platform exposes no device id")
|
||||
}
|
||||
|
||||
// A device the subdirectory is not on, standing in for an operand
|
||||
// rooted on a different filesystem.
|
||||
otherDev := dev + 1
|
||||
e := dirEntryFor(t, dir, "sub")
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
oneFS bool
|
||||
parent dirJob
|
||||
wantOK bool
|
||||
}{
|
||||
// -x on, subdirectory on a different device than its operand:
|
||||
// descent is refused.
|
||||
{"reject across boundary", true,
|
||||
dirJob{rootDev: otherDev, rootDevOK: true}, false},
|
||||
// -x on but the operand's own device is unknown: the boundary
|
||||
// check is bypassed and descent proceeds.
|
||||
{"bypass when root device unknown", true,
|
||||
dirJob{rootDev: otherDev, rootDevOK: false}, true},
|
||||
// Default (no -x): boundaries are crossed even onto a different
|
||||
// device.
|
||||
{"cross by default", false,
|
||||
dirJob{rootDev: otherDev, rootDevOK: true}, true},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
job, ok, evs := callSubdirJob(t, sub, e, tc.parent, tc.oneFS)
|
||||
if ok != tc.wantOK {
|
||||
t.Fatalf("accepted = %v, want %v", ok, tc.wantOK)
|
||||
}
|
||||
|
||||
if len(evs) != 0 {
|
||||
t.Fatalf("unexpected events: %+v", evs)
|
||||
}
|
||||
|
||||
// The accepted job must carry the operand's device down, or -x
|
||||
// stops checking below the first level.
|
||||
want := dirJob{
|
||||
path: sub,
|
||||
rootDev: tc.parent.rootDev,
|
||||
rootDevOK: tc.parent.rootDevOK,
|
||||
}
|
||||
if ok && job != want {
|
||||
t.Fatalf("job = %+v, want %+v", job, want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// errInfoUnavailable is returned by errDirEntry.Info().
|
||||
var errInfoUnavailable = errors.New("info unavailable")
|
||||
|
||||
// errDirEntry is a directory entry whose Info() always fails, driving
|
||||
// subdirJob's stat-error branch deterministically.
|
||||
type errDirEntry struct{ name string }
|
||||
|
||||
func (e errDirEntry) Name() string { return e.name }
|
||||
func (errDirEntry) IsDir() bool { return true }
|
||||
func (errDirEntry) Type() fs.FileMode {
|
||||
return fs.ModeDir
|
||||
}
|
||||
|
||||
func (errDirEntry) Info() (fs.FileInfo, error) {
|
||||
return nil, errInfoUnavailable
|
||||
}
|
||||
|
||||
// TestSubdirJobStatError asserts that when a subdirectory's Info()
|
||||
// fails under -x, subdirJob warns and refuses descent.
|
||||
func TestSubdirJobStatError(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
p := "/does/not/matter/sub"
|
||||
|
||||
_, ok, evs := callSubdirJob(t, p, errDirEntry{name: "sub"},
|
||||
dirJob{rootDev: 1, rootDevOK: true}, true)
|
||||
if ok {
|
||||
t.Fatal("descent accepted after stat error, want refused")
|
||||
}
|
||||
|
||||
if len(evs) != 1 || !evs[0].fail || !strings.Contains(evs[0].warn, p) {
|
||||
t.Fatalf("want one warning naming the path, got %+v", evs)
|
||||
}
|
||||
}
|
||||
|
||||
// buildSmokeTree recreates the README smoke-test filesystem layout
|
||||
// with deterministic content and returns the tree root.
|
||||
func buildSmokeTree(t *testing.T) string {
|
||||
|
||||
Reference in New Issue
Block a user