sfdupes/cancel_test.go

package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"syscall"
	"testing"
	"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.
const poolUnwind = 2 * time.Second

// walkClock is a context whose cancellation is driven by the scan's
// own progress rather than by the wall clock: it cancels itself the
// moment its Done method has been consulted n times. That is what
// makes "cancel in the middle of the walk" reproducible instead of a
// race against a timer.
//
// 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.
type walkClock struct {
	n    int64
	seen atomic.Int64
	once sync.Once
	done chan struct{}
}

// newWalkClock returns a context that cancels itself on the nth
// consultation of its Done method.
func newWalkClock(n int64) *walkClock {
	return &walkClock{n: n, done: make(chan struct{})}
}

// Done returns the cancellation channel, cancelling the context on the
// nth call and on every call after it. The same channel is returned
// throughout, so a caller that took it before the cancellation still
// observes the close.
func (c *walkClock) Done() <-chan struct{} {
	if c.seen.Add(1) >= c.n {
		c.once.Do(func() { close(c.done) })
	}

	return c.done
}

// Err reports the cancellation without consuming a consultation, which
// is what lets the post-walk guard read it without disturbing the
// count.
func (c *walkClock) Err() error {
	select {
	case <-c.done:
		return context.Canceled
	default:
		return nil
	}
}

// Deadline reports no deadline: this context is cancelled by progress,
// never by time.
func (c *walkClock) Deadline() (time.Time, bool) {
	return time.Time{}, false
}

// Value carries nothing.
func (c *walkClock) Value(_ any) any {
	return nil
}

// walkCancelDirs and walkCancelFilesPerDir shape the fixture for the
// mid-walk cancellation test. Spreading the files over directories is
// load-bearing: it is what bounds how much of the tree can still be
// walked after the cancellation, since the workers drop every
// directory still queued and only the handful already in flight can
// emit anything more.
const (
	walkCancelDirs         = 100
	walkCancelFilesPerDir  = 20
	walkCancelFiles        = walkCancelDirs * walkCancelFilesPerDir
	walkCancelWorkers      = 4
	walkCancelInFlightDirs = walkCancelWorkers * walkCancelFilesPerDir
)

// walkCancelAtDone is the consultation on which the fixture's context
// cancels itself. A quarter of the file count is far past the index
// load's fixed handful and far short of the walk's total, so the
// cancellation lands deep inside the walk and nowhere near either end
// of it.
const walkCancelAtDone = walkCancelFiles / 4

// buildWalkCancelTree writes walkCancelFiles empty files spread over
// walkCancelDirs subdirectories. Zero-length files are never opened by
// the hasher, so the fixture costs directory entries and no read I/O
// while still giving the walk thousands of events to emit.
func buildWalkCancelTree(t *testing.T) string {
	t.Helper()

	dir := t.TempDir()

	for i := range walkCancelDirs {
		sub := filepath.Join(dir, "d"+strconv.Itoa(i))

		err := os.Mkdir(sub, 0o750)
		if err != nil {
			t.Fatal(err)
		}

		writeEmptyFiles(t, sub, walkCancelFilesPerDir)
	}

	return dir
}

// assertRecordsIntact fails when the database no longer holds exactly
// the records it held before, reporting the first difference rather
// than dumping thousands of paths.
func assertRecordsIntact(t *testing.T, db *sql.DB, before []string) {
	t.Helper()

	got := recordPaths(dbRecords(t, db))
	if len(got) != len(before) {
		t.Fatalf("%d records after the cancelled scan, want %d",
			len(got), len(before))
	}

	for i := range got {
		if got[i] != before[i] {
			t.Fatalf("record %d = %q after the cancelled scan, want %q",
				i, got[i], before[i])
		}
	}
}

// TestSyncScanCancelledMidWalkKeepsRecords is the regression net under
// the post-walk guard. The scan is cancelled part-way through the
// walk, so it reaches the guard holding a genuinely partial size
// census and a still-populated index of records the walk never got to.
// 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
// notices if it stops doing so: deleting the guard, or making it
// unreachable, makes the scan carry its truncated view into the update
// phase, which counts every record the walk never reached for removal.
//
//nolint:paralleltest // counts goroutines: must not run beside others
func TestSyncScanCancelledMidWalkKeepsRecords(t *testing.T) {
	dir := buildWalkCancelTree(t)
	db := openTestDB(t)

	st := syncTree(t, db, dir)
	if st.added != walkCancelFiles {
		t.Fatalf("setup scan added %d records, want %d",
			st.added, walkCancelFiles)
	}

	before := recordPaths(dbRecords(t, db))
	base := baselineGoroutines(t)

	st, err := syncScan(newWalkClock(walkCancelAtDone), db,
		[]string{dir}, walkCancelWorkers, false)

	assertWalkGuardAborted(t, st, err)
	assertRecordsIntact(t, db, before)

	if got := settledGoroutines(t, base); got > base {
		t.Errorf("goroutines = %d after the cancelled scan, want %d back",
			got, base)
	}
}

// assertWalkGuardAborted checks that the scan stopped at the post-walk
// guard: with a census that is neither empty (the walk really ran)
// nor complete (it really was cut short), and with no record counted
// for removal. A removal count means the partial census was carried
// past the guard into the update phase, which is the failure this test
// exists to catch.
func assertWalkGuardAborted(t *testing.T, st scanStats, err error) {
	t.Helper()

	if !errors.Is(err, context.Canceled) {
		t.Fatalf("syncScan cancelled mid-walk = %v, want %v",
			err, context.Canceled)
	}

	if st.unchanged == 0 {
		t.Fatalf("stats = %+v: the census is empty, so the walk never "+
			"ran and the guard was reached for the wrong reason", st)
	}

	if st.unchanged >= walkCancelFiles {
		t.Fatalf("stats = %+v: the census covers the whole tree, so the "+
			"walk was not cut short", st)
	}

	// 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
	if st.unchanged > limit {
		t.Errorf("census covers %d files, want at most %d: the walk kept "+
			"taking directories off the queue after cancellation",
			st.unchanged, limit)
	}

	if st.removed != 0 {
		t.Errorf("stats = %+v: the scan counted records for removal from "+
			"a partial census", st)
	}
}

// TestSyncScanCancelledBeforeLoadIndex covers the trivial end of the
// cancellation path: a scan handed a context that is already cancelled
// fails in the index load, before the walk pool is ever started. It
// says nothing about the post-walk guard — nothing downstream of
// loadIndex runs at all — only that the failure surfaces as a
// cancellation and that no record is touched on the way out.
func TestSyncScanCancelledBeforeLoadIndex(t *testing.T) {
	t.Parallel()

	dir := buildSmokeTree(t)
	db := openTestDB(t)

	syncTree(t, db, dir)

	before := recordPaths(dbRecords(t, db))

	ctx, cancel := context.WithCancel(t.Context())
	cancel()

	st, err := syncScan(ctx, db, []string{dir}, walkCancelWorkers, false)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("syncScan on a cancelled context = %v, want %v",
			err, context.Canceled)
	}

	if st != (scanStats{}) {
		t.Errorf("stats = %+v, want none: the scan gave up in the index "+
			"load, before any phase ran", st)
	}

	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,
// failing the test if it does not close within poolUnwind. A pool that
// ignored its cancellation leaves its channel open with its goroutines
// parked, and this is what reports that as an assertion.
func drainClosed[T any](t *testing.T, ch <-chan T, what string) int {
	t.Helper()

	counted := make(chan int, 1)

	go func() {
		n := 0
		for range ch {
			n++
		}

		counted <- n
	}()

	select {
	case n := <-counted:
		return n
	case <-time.After(poolUnwind):
		t.Fatalf("%s stayed open after cancellation", what)

		return 0
	}
}

// awaitReturn fails the test if done is not closed within poolUnwind.
func awaitReturn(t *testing.T, done <-chan struct{}, what string) {
	t.Helper()

	select {
	case <-done:
	case <-time.After(poolUnwind):
		t.Fatalf("%s did not return after cancellation", what)
	}
}

// cancelledContext returns a context that is already cancelled.
func cancelledContext(t *testing.T) context.Context {
	t.Helper()

	ctx, cancel := context.WithCancel(t.Context())
	cancel()

	return ctx
}

// TestSendEventAbandonsBlockedSend checks that a walk goroutine with an
// event to deliver and nobody to deliver it to leaves on cancellation
// instead of holding the pool open. The channel here is unbuffered and
// unread, so the send can never complete.
func TestSendEventAbandonsBlockedSend(t *testing.T) {
	t.Parallel()

	done := make(chan struct{})
	events := make(chan walkEvent)

	go func() {
		defer close(done)

		sendEvent(cancelledContext(t), events, walkEvent{})
	}()

	awaitReturn(t, done, "sendEvent")
}

// TestWalkWorkersDropQueuedDirs checks that cancelled walk workers keep
// reading jobs and drop the directories rather than stopping their
// read: the range over jobs has to run out for the pool to tear down
// and close its event stream.
func TestWalkWorkersDropQueuedDirs(t *testing.T) {
	t.Parallel()

	dir := t.TempDir()
	writeEmptyFiles(t, dir, walkCancelFilesPerDir)

	jobs, _, events := startWalkWorkers(cancelledContext(t), 2, false)

	for range 4 {
		jobs <- dirJob{path: dir}
	}

	close(jobs)

	if n := drainClosed(t, events, "the walk event stream"); n != 0 {
		t.Errorf("cancelled walk workers emitted %d events, want none", n)
	}
}

// TestWalkWorkerAbandonsSubdirHandoff checks the other blocking send a
// walk worker makes: handing discovered subdirectories back to the
// dispatcher. Once the dispatcher has left, nothing drains that
// channel, and a worker parked on it would hold the pool open forever.
func TestWalkWorkerAbandonsSubdirHandoff(t *testing.T) {
	t.Parallel()

	dir := t.TempDir()
	writeEmptyFiles(t, dir, 1)

	err := os.Mkdir(filepath.Join(dir, "sub"), 0o750)
	if err != nil {
		t.Fatal(err)
	}

	ctx, cancel := context.WithCancel(t.Context())
	defer cancel()

	jobs, subdirs, events := startWalkWorkers(ctx, 1, false)

	// Fill the hand-back channel to its capacity — one slot per worker
	// — so the worker's own hand-back is certain to block.
	subdirs <- nil

	jobs <- dirJob{path: dir}

	// The file event proves the worker has read the directory and has
	// nothing left to do but the blocked hand-back.
	ev := <-events
	if ev.fail {
		t.Fatalf("walk event = %+v, want the fixture file", ev)
	}

	cancel()
	close(jobs)
	drainClosed(t, events, "the walk event stream")
}

// TestDispatchDirsClosesJobsWhenCancelled checks that a dispatcher
// leaving on cancellation closes the job channel on its way out. The
// workers range over that channel; a dispatcher that returned without
// closing it would strand every one of them.
func TestDispatchDirsClosesJobsWhenCancelled(t *testing.T) {
	t.Parallel()

	// Unbuffered and unread: with no worker pool behind it, the
	// dispatcher can only leave through its cancellation case.
	jobs := make(chan dirJob)
	subdirs := make(chan []dirJob)
	initial := []dirJob{{path: "/a"}, {path: "/b"}}

	dispatchDirs(cancelledContext(t), initial, jobs, subdirs)

	if n := drainClosed(t, jobs, "the walk job queue"); n > len(initial) {
		t.Errorf("dispatcher queued %d jobs, want at most %d",
			n, len(initial))
	}
}

// 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.
func TestFeedHashJobsClosesJobsWhenCancelled(t *testing.T) {
	t.Parallel()

	done := make(chan struct{})
	// Unbuffered and unread until the feeder has returned, so the only
	// way out of the feeder is its cancellation case.
	jobs := make(chan []fileRec)
	runs := [][]fileRec{{{path: "a"}}, {{path: "b"}}}

	go func() {
		defer close(done)

		feedHashJobs(cancelledContext(t), runs, jobs)
	}()

	awaitReturn(t, done, "feedHashJobs")

	if _, ok := <-jobs; ok {
		t.Error("the hash job channel was left open after cancellation")
	}
}

// TestHashWorkerDropsQueuedRuns checks that a cancelled hash worker
// keeps reading jobs and drops the runs rather than reading files
// 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.
func TestHashWorkerDropsQueuedRuns(t *testing.T) {
	t.Parallel()

	done := make(chan struct{})
	jobs := make(chan []fileRec, 1)
	results := make(chan hashResult, 1)

	run := []fileRec{{path: filepath.Join(t.TempDir(), "missing"), size: 1}}

	jobs <- run

	close(jobs)

	go func() {
		defer close(done)

		hashWorker(cancelledContext(t), jobs, results, hashSignature)
	}()

	awaitReturn(t, done, "hashWorker")

	select {
	case r := <-results:
		t.Errorf("cancelled hash worker produced %+v, want the run dropped",
			r)
	default:
	}
}

// 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.
func TestHashPhaseCancelledReturnsContextError(t *testing.T) {
	t.Parallel()

	s := &scanState{
		db:     openTestDB(t),
		toHash: []fileRec{{path: "a", size: 1, dev: 1, ino: 1}},
	}

	err := s.hashPhase(cancelledContext(t), 2)
	if !errors.Is(err, context.Canceled) {
		t.Fatalf("hashPhase on a cancelled context = %v, want %v",
			err, context.Canceled)
	}
}