progress_test.go

package progress_test

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	"github.com/jackc/pgx/v5"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/block/pg-sprite/pkg/dbconn"
	"github.com/block/pg-sprite/pkg/progress"
)

type fakeClock struct{ now time.Time }

func (c *fakeClock) Now() time.Time { return c.now }

// fakeRow satisfies pgx.Row with a caller-supplied Scan.
type fakeRow struct{ scan func(dest ...any) error }

func (r fakeRow) Scan(dest ...any) error { return r.scan(dest...) }

// fakeSession satisfies dbconn.RowQuerier with a caller-supplied query,
// standing in for the executor's reserved verdict session.
type fakeSession struct {
	query func(ctx context.Context, sql string, args ...any) pgx.Row
}

func (s fakeSession) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
	return s.query(ctx, sql, args...)
}

// runningTrackerWithBuild returns a tracker mid concurrent index build,
// polling against session.
func runningTrackerWithBuild(t *testing.T, session fakeSession) *progress.Tracker {
	t.Helper()
	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
	require.NoError(t, err)
	tracker.Start(1, progress.OperationConcurrentIndex)
	tracker.StartStep(1, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx ON public.t (id)")
	tracker.SetConcurrentBuild(session, 4242)
	return tracker
}

func TestTrackerReportsSequencePositionAndInjectedElapsed(t *testing.T) {
	clock := &fakeClock{now: time.Unix(100, 0)}
	tracker, err := progress.NewTracker(clock)
	require.NoError(t, err)

	tracker.Start(3, progress.OperationBrief)
	clock.now = clock.now.Add(2 * time.Second)
	tracker.StartStep(2, progress.OperationValidate, "ALTER TABLE public.t VALIDATE CONSTRAINT c")
	tracker.SetAttempt(2)
	clock.now = clock.now.Add(750 * time.Millisecond)

	snapshot, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Equal(t, progress.PhaseRunning, snapshot.Phase)
	assert.Equal(t, 2, snapshot.Step)
	assert.Equal(t, 3, snapshot.TotalSteps)
	assert.Equal(t, 2750*time.Millisecond, snapshot.Elapsed)
	assert.Equal(t, 750*time.Millisecond, snapshot.StepElapsed)
	assert.Equal(t, progress.OperationValidate, snapshot.Detail.Operation)
	assert.Equal(t, 2, snapshot.Detail.Attempt)
	assert.Nil(t, snapshot.Detail.Work, "native progress must not fabricate copy counters")
}

func TestTrackerSequenceStepsAdvanceMonotonically(t *testing.T) {
	clock := &fakeClock{now: time.Unix(100, 0)}
	tracker, err := progress.NewTracker(clock)
	require.NoError(t, err)
	tracker.Start(3, progress.OperationBrief)

	for step := 1; step <= 3; step++ {
		tracker.StartStep(step, progress.OperationBrief, "ALTER TABLE public.t ADD COLUMN c int")
		snapshot, progressErr := tracker.Progress(t.Context())
		require.NoError(t, progressErr)
		assert.Equal(t, step, snapshot.Step)
		assert.Equal(t, 3, snapshot.TotalSteps)
	}
	tracker.Finish(nil)
	terminal, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Equal(t, progress.PhaseFinished, terminal.Phase)
	assert.Equal(t, 3, terminal.Step)
}

func TestTrackerReportsTerminalState(t *testing.T) {
	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
	require.NoError(t, err)
	tracker.Start(1, progress.OperationOptimistic)
	tracker.Finish(nil)

	snapshot, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Equal(t, progress.PhaseFinished, snapshot.Phase)
	assert.False(t, snapshot.Detail.Active)
}

func TestNewTrackerRequiresClock(t *testing.T) {
	_, err := progress.NewTracker(nil)
	require.Error(t, err)
}

// A terminal snapshot is terminal: elapsed values freeze at the instant
// Finish recorded and do not grow with the clock, for both outcomes.
func TestTerminalSnapshotFreezesElapsed(t *testing.T) {
	const sql = "ALTER TABLE public.t ADD COLUMN c int"
	cases := []struct {
		name    string
		outcome error
		phase   progress.Phase
	}{
		{name: "finished", outcome: nil, phase: progress.PhaseFinished},
		{name: "failed", outcome: errors.New("build failed"), phase: progress.PhaseFailed},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			clock := &fakeClock{now: time.Unix(100, 0)}
			tracker, err := progress.NewTracker(clock)
			require.NoError(t, err)
			tracker.Start(1, progress.OperationOptimistic)
			tracker.StartStep(1, progress.OperationOptimistic, sql)
			clock.now = clock.now.Add(3 * time.Second)
			tracker.Finish(tc.outcome)

			clock.now = clock.now.Add(time.Hour)
			snapshot, err := tracker.Progress(t.Context())
			require.NoError(t, err)
			assert.Equal(t, tc.phase, snapshot.Phase)
			assert.Equal(t, 3*time.Second, snapshot.Elapsed, "elapsed must freeze at Finish")
			assert.Equal(t, 3*time.Second, snapshot.StepElapsed, "step elapsed must freeze at Finish")
			assert.Equal(t, sql, snapshot.Detail.Statement, "terminal snapshot must retain its statement")

			clock.now = clock.now.Add(time.Hour)
			again, err := tracker.Progress(t.Context())
			require.NoError(t, err)
			assert.Equal(t, snapshot, again, "a terminal snapshot must not change between polls")
		})
	}
}

// Start resets everything a prior run left behind: a reused tracker must
// never report the previous run's step, terminal instant, or build session.
func TestStartResetsPriorRunState(t *testing.T) {
	clock := &fakeClock{now: time.Unix(100, 0)}
	tracker, err := progress.NewTracker(clock)
	require.NoError(t, err)
	tracker.Start(3, progress.OperationBrief)
	tracker.StartStep(2, progress.OperationValidate, "ALTER TABLE public.t VALIDATE CONSTRAINT c")
	tracker.SetConcurrentBuild(fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(...any) error {
			t.Fatal("a new run must not poll the prior run's session")
			return nil
		}}
	}}, 4242)
	tracker.Finish(errors.New("first run failed"))

	clock.now = clock.now.Add(time.Minute)
	tracker.Start(1, progress.OperationConcurrentIndex)
	clock.now = clock.now.Add(time.Second)

	snapshot, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Equal(t, progress.PhaseRunning, snapshot.Phase)
	assert.Zero(t, snapshot.Step, "the prior run's step must not leak into the new run")
	assert.Equal(t, 1, snapshot.TotalSteps)
	assert.Equal(t, time.Second, snapshot.Elapsed, "elapsed must restart, not resume from the prior terminal instant")
	assert.Equal(t, progress.OperationConcurrentIndex, snapshot.Detail.Operation)
}

// The JSON shape is the adapter-facing contract: exact keys, exact
// omissions, driven through a real poll so the test pins what a consumer
// actually receives; a consumer pins the current format_version against it.
func TestSnapshotJSONShape(t *testing.T) {
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(dest ...any) error {
			*(dest[0].(*string)) = "building index"
			*(dest[1].(*uint64)) = 11 // blocks_done
			*(dest[2].(*uint64)) = 40 // blocks_total
			*(dest[3].(*uint64)) = 7  // tuples_done
			*(dest[4].(*uint64)) = 21 // tuples_total
			*(dest[5].(*uint64)) = 3  // lockers_total
			*(dest[6].(*uint64)) = 1  // lockers_done
			pid := int32(31337)
			*(dest[7].(**int32)) = &pid
			return nil
		}}
	}}
	clock := &fakeClock{now: time.Unix(100, 0)}
	tracker, err := progress.NewTracker(clock)
	require.NoError(t, err)
	tracker.Start(3, progress.OperationAdmitting)
	clock.now = clock.now.Add(2 * time.Second)
	tracker.StartStep(2, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx ON public.t (id)")
	tracker.SetAttempt(2)
	tracker.SetConcurrentBuild(session, 4242)
	clock.now = clock.now.Add(750 * time.Millisecond)

	snapshot, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	raw, err := json.Marshal(snapshot)
	require.NoError(t, err)
	assert.JSONEq(t, `{
		"format_version": 6,
		"phase": "running",
		"step": 2,
		"total_steps": 3,
		"elapsed_ns": 2750000000,
		"step_elapsed_ns": 750000000,
		"detail": {
			"operation": "concurrent-index-build",
			"statement": "CREATE INDEX CONCURRENTLY idx ON public.t (id)",
			"server_phase": "building index",
			"active": true,
			"attempt": 2,
			"current_locker_pid": 31337,
			"work": {
				"rows_copied": 0,
				"rows_total": 0,
				"bytes_copied": 0,
				"bytes_total": 0,
				"chunks_compared": 0,
				"rows_hashed": 0,
				"chunks_mismatched": 0,
				"chunks_repaired": 0,
				"chunks_reread": 0,
				"changes_applied": 0,
				"changes_buffered": 0,
				"lag_bytes": 0,
				"blocks_done": 11,
				"blocks_total": 40,
				"tuples_done": 7,
				"tuples_total": 21,
				"lockers_total": 3,
				"lockers_done": 1
			}
		}
	}`, string(raw))
}

func TestTrackerReportsCurrentStatementAndRetainsItOnFinish(t *testing.T) {
	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
	require.NoError(t, err)
	tracker.Start(3, progress.OperationAdmitting)

	const sql = "CREATE INDEX idx ON public.t (id)"
	tracker.StartStep(2, progress.OperationBrief, sql)
	snapshot, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Equal(t, sql, snapshot.Detail.Statement)

	tracker.Finish(nil)
	finished, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Equal(t, sql, finished.Detail.Statement)
}

// Optional fields are omitted, not emitted as zero values — but the always-on
// keys (format_version, phase, both elapsed counters, active) are present
// even on an idle tracker, so a consumer never guesses whether zero means
// "unset" or "omitted".
func TestSnapshotJSONOmitsUnsetOptionalFields(t *testing.T) {
	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
	require.NoError(t, err)

	snapshot, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	raw, err := json.Marshal(snapshot)
	require.NoError(t, err)
	assert.JSONEq(t, `{
		"format_version": 6,
		"phase": "pending",
		"elapsed_ns": 0,
		"step_elapsed_ns": 0,
		"detail": {"active": false}
	}`, string(raw))
}

// The server row's columns must land on the right Work fields — distinct
// values for every column so a swapped pair cannot pass.
func TestProgressMergesServerIndexBuildWork(t *testing.T) {
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(dest ...any) error {
			*(dest[0].(*string)) = "building index"
			*(dest[1].(*uint64)) = 11 // blocks_done
			*(dest[2].(*uint64)) = 40 // blocks_total
			*(dest[3].(*uint64)) = 7  // tuples_done
			*(dest[4].(*uint64)) = 21 // tuples_total
			*(dest[5].(*uint64)) = 3  // lockers_total
			*(dest[6].(*uint64)) = 1  // lockers_done
			pid := int32(31337)
			*(dest[7].(**int32)) = &pid
			return nil
		}}
	}}
	tracker := runningTrackerWithBuild(t, session)

	s, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.True(t, s.Detail.Active)
	assert.Equal(t, "building index", s.Detail.ServerPhase)
	require.NotNil(t, s.Detail.Work)
	assert.Equal(t, uint64(11), s.Detail.Work.BlocksDone)
	assert.Equal(t, uint64(40), s.Detail.Work.BlocksTotal)
	assert.Equal(t, uint64(7), s.Detail.Work.TuplesDone)
	assert.Equal(t, uint64(21), s.Detail.Work.TuplesTotal)
	assert.Equal(t, uint64(3), s.Detail.Work.LockersTotal)
	assert.Equal(t, uint64(1), s.Detail.Work.LockersDone)
	assert.Equal(t, uint32(31337), s.Detail.CurrentLockerPID)
	assert.Zero(t, s.Detail.Work.RowsCopied, "native progress must not fabricate copy counters")
}

// backendObservation is what the server reports for the build's backend:
// its pg_stat_activity state (nil for a hidden or untracked backend) and
// whether the cancel signal was delivered.
type backendObservation struct {
	state     *string
	cancelled bool
}

func observed(state string, cancelled bool) backendObservation {
	return backendObservation{state: &state, cancelled: cancelled}
}

// cancelSession is a verdict session that answers the cancel query with the
// supplied observation and records the backend PID it was asked to signal.
func cancelSession(t *testing.T, obs backendObservation, scanErr error, signalled *uint32) fakeSession {
	t.Helper()
	return fakeSession{query: func(ctx context.Context, sql string, args ...any) pgx.Row {
		assert.Contains(t, sql, "pg_catalog.pg_cancel_backend",
			"the signal must resolve in pg_catalog, never through search_path")
		require.Len(t, args, 1)
		pid, ok := args[0].(uint32)
		require.True(t, ok, "the backend PID must be passed as a bind parameter")
		*signalled = pid
		return fakeRow{scan: func(dest ...any) error {
			if scanErr != nil {
				return scanErr
			}
			*(dest[0].(**string)) = obs.state
			*(dest[1].(*bool)) = obs.cancelled
			return nil
		}}
	}}
}

// neverQueried is a session no build may reach: any query is a test failure.
func neverQueried(t *testing.T) fakeSession {
	t.Helper()
	return fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		t.Fatal("no build is active, so the session must not be queried")
		return nil
	}}
}

func TestCancelBuildSignalsTheActiveBuild(t *testing.T) {
	var signalled uint32
	tracker := runningTrackerWithBuild(t, cancelSession(t, observed("active", true), nil, &signalled))
	require.NoError(t, tracker.CancelBuild(t.Context()))
	assert.Equal(t, uint32(4242), signalled, "the cancel must target the build's own backend")
}

// The signal rides on the executor's reserved verdict session, so it never
// runs under the caller's own context: a caller deadline expiring mid-query
// would tear down the session the build's failure verdict needs. The
// caller's context still decides whether a signal is attempted at all.
func TestCancelBuildRunsTheSignalDetachedFromTheCaller(t *testing.T) {
	t.Run("the caller giving up mid-query does not end the signal's context", func(t *testing.T) {
		callerCtx, cancel := context.WithCancel(t.Context())
		var signalled uint32
		session := cancelSession(t, observed("active", true), nil, &signalled)
		inner := session.query
		session.query = func(ctx context.Context, sql string, args ...any) pgx.Row {
			cancel()
			assert.NoError(t, ctx.Err(), "cancelling the caller's context must not cancel the signal's")
			_, bounded := ctx.Deadline()
			assert.True(t, bounded, "a detached signal must carry its own deadline")
			return inner(ctx, sql, args...)
		}
		tracker := runningTrackerWithBuild(t, session)
		require.NoError(t, tracker.CancelBuild(callerCtx))
		assert.Equal(t, uint32(4242), signalled)
	})
	t.Run("a caller that already gave up sends nothing", func(t *testing.T) {
		tracker := runningTrackerWithBuild(t, neverQueried(t))
		ended, cancel := context.WithCancel(t.Context())
		cancel()
		err := tracker.CancelBuild(ended)
		require.ErrorIs(t, err, context.Canceled)
		assert.ErrorIs(t, err, progress.ErrCancelNotDispatched)
		assert.NotErrorIs(t, err, progress.ErrNoActiveBuild)
	})
}

// cancelOutcomeDeadline bounds how long a test waits for a CancelBuild call
// to return once nothing holds the reserved session any more.
const cancelOutcomeDeadline = 5 * time.Second

// awaitCancelOutcome returns the error a CancelBuild call delivered on
// outcome, failing the test if it does not arrive in time.
func awaitCancelOutcome(t *testing.T, outcome <-chan error) error {
	t.Helper()
	deadline := time.After(cancelOutcomeDeadline)
	select {
	case err := <-outcome:
		return err
	case <-deadline:
		t.Fatal("CancelBuild did not return after the reserved session was released")
		return nil
	}
}

// blockingSession is a verdict session whose first query of each kind can
// be held: the progress read parks until releasePoll is closed, and the
// first cancel signal parks until releaseSignal is closed. Every cancel
// signal that reaches the session is counted, so a test can prove that a
// caller who gave up sent nothing.
type blockingSession struct {
	pollEntered   chan struct{}
	releasePoll   chan struct{}
	signalEntered chan struct{}
	releaseSignal chan struct{}
	signals       atomic.Int32
}

func newBlockingSession() *blockingSession {
	return &blockingSession{
		pollEntered:   make(chan struct{}),
		releasePoll:   make(chan struct{}),
		signalEntered: make(chan struct{}),
		releaseSignal: make(chan struct{}),
	}
}

func (s *blockingSession) session() fakeSession {
	return fakeSession{query: func(_ context.Context, sql string, _ ...any) pgx.Row {
		if !strings.Contains(sql, "pg_catalog.pg_cancel_backend") {
			return fakeRow{scan: func(...any) error {
				close(s.pollEntered)
				<-s.releasePoll
				return pgx.ErrNoRows
			}}
		}
		if s.signals.Add(1) == 1 {
			close(s.signalEntered)
			<-s.releaseSignal
		}
		return fakeRow{scan: func(dest ...any) error {
			active := "active"
			*(dest[0].(**string)) = &active
			*(dest[1].(*bool)) = true
			return nil
		}}
	}}
}

// A caller that gives up while CancelBuild waits for the reserved session
// — held by an in-flight progress poll or by an overlapping cancel — sends
// nothing, and says so with ErrCancelNotDispatched alongside its own
// context error, so a consumer can tell "never sent" from a context error
// that came back after the signal was sent.
func TestCancelBuildReportsNotDispatchedWhenTheCallerGivesUpWaitingForTheSession(t *testing.T) {
	t.Run("cancelled behind an in-flight progress poll", func(t *testing.T) {
		blocking := newBlockingSession()
		close(blocking.releaseSignal)
		tracker := runningTrackerWithBuild(t, blocking.session())

		var workers sync.WaitGroup
		defer workers.Wait()
		workers.Go(func() {
			_, err := tracker.Progress(t.Context())
			assert.NoError(t, err)
		})
		<-blocking.pollEntered

		callerCtx, giveUp := context.WithCancel(t.Context())
		defer giveUp()
		outcome := make(chan error, 1)
		// The poll holds the session until releasePoll closes, so the caller
		// gives up before CancelBuild can take it, whether or not CancelBuild
		// has started waiting yet.
		workers.Go(func() { outcome <- tracker.CancelBuild(callerCtx) })
		giveUp()
		close(blocking.releasePoll)

		err := awaitCancelOutcome(t, outcome)
		require.ErrorIs(t, err, progress.ErrCancelNotDispatched)
		assert.ErrorIs(t, err, context.Canceled)
		assert.NotErrorIs(t, err, progress.ErrNoActiveBuild)
		assert.Equal(t, int32(0), blocking.signals.Load(), "a caller that gave up must send no signal")

		// The build is still signallable by a caller that has not given up.
		require.NoError(t, tracker.CancelBuild(t.Context()))
		assert.Equal(t, int32(1), blocking.signals.Load())
	})
	t.Run("deadline passed behind an in-flight progress poll", func(t *testing.T) {
		blocking := newBlockingSession()
		close(blocking.releaseSignal)
		tracker := runningTrackerWithBuild(t, blocking.session())

		var workers sync.WaitGroup
		defer workers.Wait()
		workers.Go(func() {
			_, err := tracker.Progress(t.Context())
			assert.NoError(t, err)
		})
		<-blocking.pollEntered

		expired, cancel := context.WithDeadline(t.Context(), time.Unix(0, 0))
		defer cancel()
		outcome := make(chan error, 1)
		workers.Go(func() { outcome <- tracker.CancelBuild(expired) })
		close(blocking.releasePoll)

		err := awaitCancelOutcome(t, outcome)
		require.ErrorIs(t, err, progress.ErrCancelNotDispatched)
		assert.ErrorIs(t, err, context.DeadlineExceeded)
		assert.Equal(t, int32(0), blocking.signals.Load(), "a caller past its deadline must send no signal")
	})
	t.Run("cancelled behind an overlapping cancel signal", func(t *testing.T) {
		blocking := newBlockingSession()
		tracker := runningTrackerWithBuild(t, blocking.session())

		var workers sync.WaitGroup
		defer workers.Wait()
		first := make(chan error, 1)
		workers.Go(func() { first <- tracker.CancelBuild(t.Context()) })
		<-blocking.signalEntered

		callerCtx, giveUp := context.WithCancel(t.Context())
		defer giveUp()
		second := make(chan error, 1)
		workers.Go(func() { second <- tracker.CancelBuild(callerCtx) })
		giveUp()
		close(blocking.releaseSignal)

		require.NoError(t, awaitCancelOutcome(t, first), "the overlapping signal was sent and accepted")
		err := awaitCancelOutcome(t, second)
		require.ErrorIs(t, err, progress.ErrCancelNotDispatched)
		assert.ErrorIs(t, err, context.Canceled)
		assert.Equal(t, int32(1), blocking.signals.Load(), "only the overlapping cancel may reach the session")
	})
}

// A caller that gives up while a poll holds the reserved session gets its
// answer then, not when the poll lets go: the wait for the session is
// bounded by the caller's context, so an orchestrator's stop verb never
// hangs behind an observer. The poll is deliberately never released until
// the outcome is in hand; a wait that ignored the context would only return
// at the outcome deadline.
func TestCancelBuildReturnsWhenTheCallerGivesUpWhileWaiting(t *testing.T) {
	blocking := newBlockingSession()
	close(blocking.releaseSignal)
	tracker := runningTrackerWithBuild(t, blocking.session())

	var workers sync.WaitGroup
	defer workers.Wait()
	workers.Go(func() {
		_, err := tracker.Progress(t.Context())
		assert.NoError(t, err)
	})
	<-blocking.pollEntered
	defer close(blocking.releasePoll)

	callerCtx, giveUp := context.WithCancel(t.Context())
	defer giveUp()
	outcome := make(chan error, 1)
	workers.Go(func() { outcome <- tracker.CancelBuild(callerCtx) })
	giveUp()

	err := awaitCancelOutcome(t, outcome)
	require.ErrorIs(t, err, progress.ErrCancelNotDispatched)
	assert.ErrorIs(t, err, context.Canceled)
	assert.Equal(t, int32(0), blocking.signals.Load(), "a caller that gave up must send no signal")
}

// A caller that gives up while waiting for a build that ends during the
// wait learns that the build is gone, not that its cancel went unsent: there
// is nothing left to retry against.
func TestCancelBuildReportsNoActiveBuildWhenTheBuildEndsWhileTheCallerWaits(t *testing.T) {
	blocking := newBlockingSession()
	tracker := runningTrackerWithBuild(t, blocking.session())

	var workers sync.WaitGroup
	defer workers.Wait()
	workers.Go(func() {
		_, err := tracker.Progress(t.Context())
		assert.NoError(t, err)
	})
	<-blocking.pollEntered
	defer close(blocking.releasePoll)

	// The executor's own resets clear the build under the state lock alone,
	// so the build can end while the poll still holds the session.
	tracker.Finish(nil)

	callerCtx, giveUp := context.WithCancel(t.Context())
	defer giveUp()
	outcome := make(chan error, 1)
	workers.Go(func() { outcome <- tracker.CancelBuild(callerCtx) })
	giveUp()

	err := awaitCancelOutcome(t, outcome)
	require.ErrorIs(t, err, progress.ErrNoActiveBuild)
	assert.NotErrorIs(t, err, progress.ErrCancelNotDispatched)
	assert.Equal(t, int32(0), blocking.signals.Load())
}

// probeCtx is a caller context — no deadline, no values — that records how
// CancelBuild consults it. Its Done channel never fires, and reports that
// CancelBuild has started waiting on it; Err reports an ended context only
// once the poll has released the session, and records whether it was read
// before that.
type probeCtx struct {
	waiting   chan struct{}
	waitOnce  sync.Once
	released  atomic.Bool
	readEarly atomic.Bool
}

func newProbeCtx() *probeCtx {
	return &probeCtx{waiting: make(chan struct{})}
}

func (c *probeCtx) Deadline() (time.Time, bool) { return time.Time{}, false }

func (c *probeCtx) Done() <-chan struct{} {
	c.waitOnce.Do(func() { close(c.waiting) })
	return nil
}

func (c *probeCtx) Err() error {
	if !c.released.Load() {
		c.readEarly.Store(true)
		return nil
	}
	return context.Canceled
}

func (c *probeCtx) Value(any) any { return nil }

// CancelBuild reads whether the caller has given up only once it holds the
// reserved session, so a caller still live when the session comes free
// sends its signal, and one that gave up in the meantime is told nothing
// was sent. A check hoisted above the wait would find the context live,
// then send after the caller had gone.
func TestCancelBuildReadsTheCallersContextOnlyOnceItHoldsTheSession(t *testing.T) {
	blocking := newBlockingSession()
	close(blocking.releaseSignal)
	tracker := runningTrackerWithBuild(t, blocking.session())

	var workers sync.WaitGroup
	defer workers.Wait()
	workers.Go(func() {
		_, err := tracker.Progress(t.Context())
		assert.NoError(t, err)
	})
	<-blocking.pollEntered
	releasePoll := sync.OnceFunc(func() { close(blocking.releasePoll) })
	defer releasePoll()

	caller := newProbeCtx()
	outcome := make(chan error, 1)
	workers.Go(func() { outcome <- tracker.CancelBuild(caller) })

	// CancelBuild is parked on the session; the context "ends" only now,
	// as the poll lets go, so any earlier read of it was premature.
	select {
	case <-caller.waiting:
	case <-time.After(cancelOutcomeDeadline):
		t.Fatal("CancelBuild never waited on the caller's context while the poll held the session")
	}
	caller.released.Store(true)
	releasePoll()

	err := awaitCancelOutcome(t, outcome)
	require.ErrorIs(t, err, progress.ErrCancelNotDispatched)
	assert.False(t, caller.readEarly.Load(), "the caller's context must be read only once the session is held")
	assert.Equal(t, int32(0), blocking.signals.Load())
}

// Once the signal is sent, a context error that comes back — the signal's
// own deadline expiring while the caller has also given up — may follow a
// signal the server acted on, so it is never reported as not dispatched.
func TestCancelBuildDoesNotReportNotDispatchedOnceTheSignalIsSent(t *testing.T) {
	callerCtx, giveUp := context.WithCancel(t.Context())
	defer giveUp()
	var signalled uint32
	timedOut := fmt.Errorf("read cancel result: %w", context.DeadlineExceeded)
	session := cancelSession(t, backendObservation{}, timedOut, &signalled)
	inner := session.query
	session.query = func(ctx context.Context, sql string, args ...any) pgx.Row {
		giveUp()
		return inner(ctx, sql, args...)
	}
	tracker := runningTrackerWithBuild(t, session)

	err := tracker.CancelBuild(callerCtx)
	require.ErrorIs(t, err, context.DeadlineExceeded)
	assert.NotErrorIs(t, err, progress.ErrCancelNotDispatched)
	assert.Equal(t, uint32(4242), signalled, "the signal reached the session before the error came back")
}

// Only a backend the server positively reports active is signalled; an
// idle backend would drop the signal, so it is reported as not running
// rather than as a delivered cancel.
func TestCancelBuildReportsAStatementNotRunning(t *testing.T) {
	var signalled uint32
	t.Run("no row for the backend", func(t *testing.T) {
		tracker := runningTrackerWithBuild(t, cancelSession(t, backendObservation{}, pgx.ErrNoRows, &signalled))
		require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrBuildNotRunning)
	})
	t.Run("backend gone between the read and the signal", func(t *testing.T) {
		tracker := runningTrackerWithBuild(t, cancelSession(t, observed("active", false), nil, &signalled))
		require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrBuildNotRunning)
	})
	for _, idle := range []string{"idle", "idle in transaction", "idle in transaction (aborted)"} {
		t.Run(idle, func(t *testing.T) {
			// The session answers true, as a mutant that signalled an
			// idle backend would see the server answer; the state must win.
			tracker := runningTrackerWithBuild(t, cancelSession(t, observed(idle, true), nil, &signalled))
			err := tracker.CancelBuild(t.Context())
			require.ErrorIs(t, err, progress.ErrBuildNotRunning)
			assert.NotErrorIs(t, err, progress.ErrBuildUnobservable)
		})
	}
}

// A state the server does not expose is neither running nor idle: the
// tracker refuses rather than signal blind, and says so distinctly, because
// "not running" would tell the operator to wait for a build that is in
// fact running.
func TestCancelBuildReportsAnUnobservableBackend(t *testing.T) {
	var signalled uint32
	cases := map[string]backendObservation{
		"hidden backend (NULL state)":  {state: nil},
		"activity tracking off":        observed("disabled", false),
		"state outside the vocabulary": observed("fastpath function call", false),
	}
	for name, obs := range cases {
		t.Run(name, func(t *testing.T) {
			tracker := runningTrackerWithBuild(t, cancelSession(t, obs, nil, &signalled))
			err := tracker.CancelBuild(t.Context())
			require.ErrorIs(t, err, progress.ErrBuildUnobservable)
			assert.NotErrorIs(t, err, progress.ErrBuildNotRunning)
			assert.NotErrorIs(t, err, progress.ErrNoActiveBuild)
		})
	}
}

func TestCancelBuildWrapsASessionFailure(t *testing.T) {
	var signalled uint32
	boom := errors.New("connection reset")
	tracker := runningTrackerWithBuild(t, cancelSession(t, backendObservation{}, boom, &signalled))
	err := tracker.CancelBuild(t.Context())
	require.ErrorIs(t, err, boom)
	assert.NotErrorIs(t, err, progress.ErrBuildNotRunning)
	assert.NotErrorIs(t, err, progress.ErrBuildUnobservable)
	assert.NotErrorIs(t, err, progress.ErrNoActiveBuild)
	assert.NotErrorIs(t, err, progress.ErrCancelNotDispatched)
}

// Every transition that ends a build's ownership of its backend also ends
// the tracker's ability to signal it: after each one CancelBuild refuses
// without touching the session, so a stale PID can never reach
// pg_cancel_backend on a backend the pool may have handed to someone else.
func TestCancelBuildRefusesOnceTheBuildIsNotActive(t *testing.T) {
	tracker, err := progress.NewTracker(&fakeClock{now: time.Unix(100, 0)})
	require.NoError(t, err)
	require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrNoActiveBuild, "idle tracker")

	tracker.Start(1, progress.OperationConcurrentIndex)
	tracker.SetConcurrentBuild(neverQueried(t), 4242)
	tracker.StopConcurrentBuild()
	require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrNoActiveBuild, "after StopConcurrentBuild")

	tracker.SetConcurrentBuild(neverQueried(t), 4242)
	tracker.Finish(nil)
	require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrNoActiveBuild, "after Finish")

	tracker.SetConcurrentBuild(neverQueried(t), 4242)
	tracker.Start(1, progress.OperationConcurrentIndex)
	require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrNoActiveBuild, "after Start")

	tracker.SetConcurrentBuild(neverQueried(t), 4242)
	tracker.StartStep(2, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx2 ON public.t (c)")
	require.ErrorIs(t, tracker.CancelBuild(t.Context()), progress.ErrNoActiveBuild, "after StartStep")
}

// A build that has left the progress view is reported inactive, with no
// stale server detail attached.
func TestProgressClearsActiveWhenServerRowIsGone(t *testing.T) {
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(...any) error { return pgx.ErrNoRows }}
	}}
	tracker := runningTrackerWithBuild(t, session)

	s, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.False(t, s.Detail.Active, "a vanished progress row means the build is no longer active")
	assert.Empty(t, s.Detail.ServerPhase)
	assert.Nil(t, s.Detail.Work)
}

// A transient query failure surfaces the error alongside the last-known
// tracker state, not a zero-valued snapshot.
func TestProgressReturnsSnapshotAlongsideQueryError(t *testing.T) {
	queryErr := errors.New("connection severed")
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(...any) error { return queryErr }}
	}}
	tracker := runningTrackerWithBuild(t, session)

	s, err := tracker.Progress(t.Context())
	require.ErrorIs(t, err, queryErr)
	assert.Equal(t, progress.PhaseRunning, s.Phase, "the snapshot must keep the last-known state on error")
	assert.Equal(t, 1, s.Step)
	assert.Equal(t, 1, s.TotalSteps)
}

// The progress read rides on the executor's reserved verdict session, so it
// never runs under the observer's own context: pgx closes a connection whose
// query's context ends mid-read, and the build's failure verdict needs that
// session intact. The read carries its own deadline instead.
func TestProgressRunsTheBuildReadDetachedFromTheObserver(t *testing.T) {
	observer, giveUp := context.WithCancel(t.Context())
	session := fakeSession{query: func(ctx context.Context, _ string, _ ...any) pgx.Row {
		giveUp()
		assert.NoError(t, ctx.Err(), "an observer giving up must not cancel the read's context")
		_, bounded := ctx.Deadline()
		assert.True(t, bounded, "a detached read must carry its own deadline")
		return fakeRow{scan: func(...any) error { return pgx.ErrNoRows }}
	}}
	tracker := runningTrackerWithBuild(t, session)

	_, err := tracker.Progress(observer)
	require.NoError(t, err)
}

// A caller whose context ended before it called Progress has given up: it
// reads nothing, like a caller that gives up while waiting for the gate, and
// it can tell from the error that no read was issued. The gate is free, so
// without the check after taking it the select would pick the read about
// half the time.
func TestProgressWithAnEndedContextReadsNothing(t *testing.T) {
	var reads atomic.Int32
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		reads.Add(1)
		return fakeRow{scan: func(...any) error { return pgx.ErrNoRows }}
	}}
	tracker := runningTrackerWithBuild(t, session)
	ended, cancel := context.WithCancel(t.Context())
	cancel()

	for range 200 {
		snapshot, err := tracker.Progress(ended)
		require.ErrorIs(t, err, progress.ErrProgressNotRead)
		require.ErrorIs(t, err, context.Canceled)
		require.Equal(t, progress.PhaseRunning, snapshot.Phase)
	}
	assert.Zero(t, reads.Load(), "reads issued for a caller whose context had already ended")
}

// The detached read's deadline outlasts the statement_timeout a pkg/dbconn
// pool gives the reserved session, so the server ends a slow read with a
// query cancellation and the client deadline never closes the connection
// first.
func TestProgressBuildReadDeadlineOutlastsTheServerStatementTimeout(t *testing.T) {
	var remaining time.Duration
	session := fakeSession{query: func(ctx context.Context, _ string, _ ...any) pgx.Row {
		deadline, ok := ctx.Deadline()
		require.True(t, ok)
		remaining = time.Until(deadline)
		return fakeRow{scan: func(...any) error { return pgx.ErrNoRows }}
	}}
	tracker := runningTrackerWithBuild(t, session)

	_, err := tracker.Progress(t.Context())
	require.NoError(t, err)
	assert.Greater(t, remaining, dbconn.DefaultStatementTimeout)
}

// An observer waits for the poll gate only as long as its context lasts: a
// poll stalled on the reserved session — a half-open connection, say — must
// not hang every other observer behind it. The one that gives up gets the
// last-known snapshot and its context's error, and never touches the session.
func TestProgressGivesUpWaitingForThePollGateWhenTheObserverEnds(t *testing.T) {
	blocking := newBlockingSession()
	tracker := runningTrackerWithBuild(t, blocking.session())

	var workers sync.WaitGroup
	defer workers.Wait()
	defer close(blocking.releasePoll)
	workers.Go(func() {
		_, err := tracker.Progress(t.Context())
		assert.NoError(t, err)
	})
	<-blocking.pollEntered

	type outcome struct {
		snapshot progress.Snapshot
		err      error
	}
	impatient, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond)
	defer cancel()
	waited := make(chan outcome, 1)
	workers.Go(func() {
		snapshot, err := tracker.Progress(impatient)
		waited <- outcome{snapshot: snapshot, err: err}
	})

	select {
	case got := <-waited:
		require.ErrorIs(t, got.err, context.DeadlineExceeded)
		assert.ErrorIs(t, got.err, progress.ErrProgressNotRead, "a caller can tell it gave up before any read from a read that timed out")
		assert.Equal(t, progress.PhaseRunning, got.snapshot.Phase, "the snapshot must keep the last-known state")
		assert.Equal(t, 1, got.snapshot.Step)
	case <-time.After(cancelOutcomeDeadline):
		t.Fatal("an observer whose context ended kept waiting for a stalled poll")
	}
}

// Two concurrent pollers must never drive the reserved session at the same
// time: a single pgx connection is not safe for concurrent use.
func TestProgressSerializesConcurrentPollers(t *testing.T) {
	firstEntered := make(chan struct{})
	overlap := make(chan struct{})
	release := make(chan struct{})
	var entries atomic.Int32
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(...any) error {
			switch entries.Add(1) {
			case 1:
				close(firstEntered)
			case 2:
				close(overlap)
			}
			<-release
			return pgx.ErrNoRows
		}}
	}}
	tracker := runningTrackerWithBuild(t, session)

	var pollers sync.WaitGroup
	for range 2 {
		pollers.Go(func() {
			_, err := tracker.Progress(t.Context())
			assert.NoError(t, err)
		})
	}
	<-firstEntered
	secondPollerMustStillWait := time.After(100 * time.Millisecond)
	select {
	case <-overlap:
		t.Fatal("two pollers reached the session concurrently")
	case <-secondPollerMustStillWait:
	}
	close(release)
	pollers.Wait()
	assert.Equal(t, int32(2), entries.Load(), "both pollers must complete, one after the other")
}

// StopConcurrentBuild must drain an in-flight observation before returning:
// the executor reuses the reserved session for its catalog verdict as soon
// as StopConcurrentBuild returns.
func TestStopConcurrentBuildDrainsInFlightObservation(t *testing.T) {
	entered := make(chan struct{})
	release := make(chan struct{})
	var queryFinished atomic.Bool
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(...any) error {
			close(entered)
			<-release
			queryFinished.Store(true)
			return pgx.ErrNoRows
		}}
	}}
	tracker := runningTrackerWithBuild(t, session)

	var workers sync.WaitGroup
	workers.Go(func() {
		_, err := tracker.Progress(t.Context())
		assert.NoError(t, err)
	})
	<-entered

	stopReturned := make(chan struct{})
	workers.Go(func() {
		tracker.StopConcurrentBuild()
		assert.True(t, queryFinished.Load(),
			"StopConcurrentBuild must not return while an observation still holds the session")
		close(stopReturned)
	})
	stopMustStillBlock := time.After(100 * time.Millisecond)
	select {
	case <-stopReturned:
		t.Fatal("StopConcurrentBuild returned while an observation was in flight")
	case <-stopMustStillBlock:
	}
	close(release)
	workers.Wait()
}

// The executor's own state updates must never wait behind a slow
// observation: polling is observability, not a gate on execution.
func TestStateMutatorsDoNotWaitForInFlightObservation(t *testing.T) {
	entered := make(chan struct{})
	release := make(chan struct{})
	session := fakeSession{query: func(context.Context, string, ...any) pgx.Row {
		return fakeRow{scan: func(...any) error {
			close(entered)
			<-release
			return pgx.ErrNoRows
		}}
	}}
	tracker := runningTrackerWithBuild(t, session)

	var workers sync.WaitGroup
	defer workers.Wait()
	defer close(release)
	workers.Go(func() {
		_, err := tracker.Progress(t.Context())
		assert.NoError(t, err)
	})
	<-entered

	mutated := make(chan struct{})
	workers.Go(func() {
		tracker.SetAttempt(2)
		tracker.StartStep(1, progress.OperationConcurrentIndex, "CREATE INDEX CONCURRENTLY idx ON public.t (id)")
		close(mutated)
	})
	mutatorDeadline := time.After(5 * time.Second)
	select {
	case <-mutated:
	case <-mutatorDeadline:
		t.Fatal("a state mutator waited behind an in-flight observation")
	}
}