sfdupes/cancel_test.go

package main

import (
	"context"
	"database/sql"
	"errors"
	"os"
	"path/filepath"
	"strconv"
	"sync"
	"sync/atomic"
	"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 a later
// phase and fail there instead, with a wrapped error rather than the
// bare cancellation.
//
//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 the guard's own
// bare cancellation as the error. A wrapped error means the partial
// census was carried past the guard into the hash or 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 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 {
		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)
}

// 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)
	}()

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