progress: detach and bound the build progress read from its observer

SAFETY.md

@@ -95,12 +95,13 @@ The short version — the full rules live in [docs/tcb-model.md](docs/tcb-model.
   caller's classification; the grammar is load-bearing expertise, not copyable mechanics),
   `pkg/progress` (the executors' progress-observation seam: they write state into a
   caller-owned tracker whose mutators take only a memory lock, and its polling reads ride
-  the reserved verdict session behind a separate one-slot poll gate, which `CancelBuild`
-  alone waits on only as long as its caller's context lasts — the executor's own state
+  the reserved verdict session behind a separate one-slot poll gate, which `Progress` and
+  `CancelBuild` wait on only as long as their caller's context lasts — the executor's own state
   updates never wait for a database read, but the handoffs that end a poll target's
   ownership *are* observer-gated: `StopConcurrentBuild` and `SetWorkSource` drain an
   in-flight poll before the executor reclaims the build's session, a wait bounded by the
-  poller's context and the session's `statement_timeout`, and `StopWorkSource` and
+  read's own deadline (the build's read runs detached from its observer's context, so an
+  observer giving up never closes the session) and the session's `statement_timeout`, and `StopWorkSource` and
   `SetConcurrentBuild` drain one before an engine step releases the state its `WorkSource`
   reads, a wait the `WorkSource` contract requires `Work` to bound itself — memory reads or
   catalog reads under a session `statement_timeout`, never the observer's context alone),

docs/invariants.md

@@ -507,6 +507,15 @@ pool — chunk copy, resume clear, progress measurement, chunk cut, digest, repa
 transaction that sets its own `lock_timeout` and `statement_timeout` before reading, so a pool
 built without `pkg/dbconn`'s session defaults still bounds every lock wait (a cut queued behind
 `ACCESS EXCLUSIVE` on a raw pool ends at the copier's lock timeout).
+The concurrent build's failure verdict runs on a reserved session that observers must not
+disturb: `Tracker.Progress` reads the build's progress on that session detached from the
+observer's context and under its own deadline, set above `pkg/dbconn`'s default
+`statement_timeout` so a slow read ends server-side with the session intact, because a context
+that ends mid-query closes the connection; `Tracker.CancelBuild` signals the same way.
+*Enforced:* `pkg/progress` (`Tracker.Progress`, `progressReadTimeout`;
+`TestProgressRunsTheBuildReadDetachedFromTheObserver`,
+`TestProgressBuildReadDeadlineOutlastsTheServerStatementTimeout`, and
+`TestProgressReadCancelledMidQueryLeavesTheVerdictSessionUsable` in `pkg/executor`).
 *Source:* [design-principles](design-principles.md#correctness-and-safety), [mysql-vs-postgresql](mysql-vs-postgresql.md#why-ddl-is-dangerous-the-lock-queue);
 CIC exception from the validation review.
 

docs/low-level-design.md

@@ -849,7 +849,8 @@ operator-accepted blocking change that committed ends in an `executed-without-on
 verdict (exit 3), the one case that is committed but not vouched for as online-safe
 ([the ladder](cli-output-examples.md#exit-codes)). Native execution
 exposes a caller-owned `progress.Tracker`. Embedders run a blocking executor
-call in their own bounded task and poll `Tracker.Progress(ctx)`: sequence position and elapsed
+call in their own bounded task and poll `Tracker.Progress(ctx)` (a build read can outlast `ctx`
+by up to a minute; see [progress-report.md](progress-report.md)): sequence position and elapsed
 time come from in-process state, while an active concurrent index build is read on demand from
 `pg_stat_progress_create_index` by the build's backend PID. There is no background poller to
 own or stop, and a missing progress-view row is represented as an inactive observation rather

docs/progress-report.md

@@ -195,14 +195,33 @@ returns the identical snapshot, elapsed values included.
 
 ## Polling semantics
 
-The tracker is caller-owned and has no goroutines or timers: polling lifetime is exactly the
-caller's context. A poll during an active concurrent index build performs one read of the
-server's progress view over the executor's reserved session; a poll during a `copy`,
-`checksum` or `catch-up` step asks the engine's work source once; every other poll is pure memory. On a query or source
+The tracker is caller-owned and has no goroutines or timers. A poll during an active
+concurrent index build performs one read of the server's progress view over the executor's
+reserved session; a poll during a `copy`, `checksum` or `catch-up` step asks the engine's
+work source once; every other poll is pure memory. On a query or source
 error the returned snapshot still carries the last-known tracker state — `phase` is never
 empty — with the error returned alongside for the caller to classify. Pollers serialize
 against each other, so the reserved session and the work source each see one observation
-at a time.
+at a time. A poller waits its turn for as long as its context lasts: one that gives up first,
+or whose context has already ended when its turn comes, gets the last-known snapshot and
+`progress.ErrProgressNotRead` wrapped with its context's error, having read nothing, so a
+stalled poll never holds every other observer behind it. Test for `ErrProgressNotRead` before
+testing for a context error: a context error without it came from the read itself.
+
+The progress read itself never runs under the poller's context. It runs detached from it, under
+its own deadline: a context that ends mid-read — an HTTP request timing out — closes the
+connection the read rides on, and that connection is the reserved session the build's failure
+verdict needs. A read that is in flight when its poller gives up therefore finishes on its own
+terms and returns its result; the work source, which is not that session, is read under the
+poller's context. Once the read is issued, `Progress` is not bounded by its context: it returns
+when the server answers — within the pool's `statement_timeout` on a `pkg/dbconn` pool (30 s by
+default) — and within twice `dbconn.DefaultStatementTimeout` (one minute) at the latest, when
+the network has stopped answering. That margin holds for a pool with the default or a lower
+`statement_timeout`; on a pool configured above it, or one whose sessions carry none, a read
+that stalls that long closes the session. A caller that needs a hard return at its own deadline
+calls `Progress` from its own goroutine. A deadline error without `ErrProgressNotRead` means the
+reserved session stopped answering, not that the caller timed out — the case to surface, since
+the build's verdict runs on that session.
 
 The handoffs that end a poll target's ownership of the step — stopping a build or a work
 source, or replacing one with the other — wait for an observation in flight before they

pkg/executor/cic_progress_poll_integration_test.go

@@ -0,0 +1,78 @@
+package executor_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/progress"
+)
+
+// heldProgressRead stands in for the reserved verdict session, running the
+// tracker's own progress query on the real connection after the server has
+// slept for two seconds, so the read is provably in flight — executing on
+// the server, its client waiting for the answer — for a window a test can
+// act in.
+type heldProgressRead struct{ conn *pgxpool.Conn }
+
+func (s heldProgressRead) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
+	return s.conn.QueryRow(ctx, `
+		SELECT progress.*
+		  FROM (SELECT pg_catalog.pg_sleep(2)) AS held,
+		       LATERAL (`+sql+`) AS progress`, args...)
+}
+
+// An observer whose context ends while its progress read is in flight must
+// not take the verdict session with it: pgx closes a connection whose query's
+// context ends mid-read, and the build's failure verdict — the invalid-index
+// own-leftover classification that unblocks automated recovery — runs on this
+// session after the build fails.
+func TestProgressReadCancelledMidQueryLeavesTheVerdictSessionUsable(t *testing.T) {
+	pool, _ := newPool(t)
+	build, err := pool.Acquire(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(build.Release)
+	verdict, err := pool.Acquire(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(verdict.Release)
+	verdictPID := verdict.Conn().PgConn().PID()
+
+	tracker, err := progress.NewTracker(progress.WallClock{})
+	require.NoError(t, err)
+	tracker.Start(1, progress.OperationConcurrentIndex)
+	tracker.SetConcurrentBuild(heldProgressRead{conn: verdict}, build.Conn().PgConn().PID())
+
+	observer, giveUp := context.WithCancel(t.Context())
+	defer giveUp()
+	polled := make(chan error, 1)
+	go func() {
+		_, pollErr := tracker.Progress(observer)
+		polled <- pollErr
+	}()
+	require.Eventually(t, func() bool {
+		var held bool
+		err := pool.QueryRow(t.Context(),
+			`SELECT EXISTS (SELECT FROM pg_catalog.pg_stat_activity
+			                 WHERE pid = $1 AND state = 'active' AND wait_event = 'PgSleep')`,
+			int32(verdictPID)).Scan(&held)
+		return err == nil && held
+	}, 30*time.Second, 20*time.Millisecond, "the progress read must be executing on the verdict session")
+
+	giveUp()
+
+	select {
+	case pollErr := <-polled:
+		require.NoError(t, pollErr, "the read runs to its own end; the observer giving up does not abort it")
+	case <-time.After(30 * time.Second):
+		t.Fatal("the progress read did not return")
+	}
+	var pid int32
+	require.NoError(t, verdict.QueryRow(t.Context(), "SELECT pg_catalog.pg_backend_pid()").Scan(&pid),
+		"the verdict session must still run queries after an observer gave up mid-read")
+	assert.Equal(t, int32(verdictPID), pid, "the session must be the same backend, not a replacement")
+}

pkg/progress/progress.go

@@ -138,17 +138,19 @@ type Snapshot struct {
 
 // Tracker is a concurrency-safe progress source. The caller owns it; it has
 // no goroutines. Progress performs the one read needed for an active index
-// build, or the one WorkSource call for an engine-measured step, making
-// polling lifetime identical to the caller's context.
+// build, or the one WorkSource call for an engine-measured step. A
+// WorkSource call lives as long as the caller's context; an index build's
+// read is bounded on its own, so a caller's context ending never reaches the
+// reserved session.
 //
 // Two guards split the tracker's concerns: mu guards the state fields and
 // is held only for memory access, so the executor's own updates never wait
 // for a database read; the poll gate serializes observers, so the reserved
 // session — a single pgx connection that is not safe for concurrent use —
 // only ever carries one progress query at a time, and a WorkSource sees one
 // poll at a time. The gate is a one-slot channel rather than a mutex so
-// that CancelBuild, the one taker with a caller waiting on an answer, can
-// stop waiting when that caller's context ends.
+// that Progress and CancelBuild, the takers with a caller waiting on an
+// answer, can stop waiting when that caller's context ends.
 //
 // Four state changes take the poll gate, because each ends a poll target's
 // ownership of the step's work and must not do so under a poll still in
@@ -192,9 +194,11 @@ func NewTracker(clock Clock) (*Tracker, error) {
 // Now returns the tracker's injected time for executor duration accounting.
 func (t *Tracker) Now() time.Time { return t.clock.Now() }
 
-// takePollGate waits, without limit, for the poll gate. The observers and
-// the handoff fences take it this way: a poll is bounded by its own
-// context once it holds the gate, and a fence waits only for that poll.
+// takePollGate waits, without limit, for the poll gate. The handoff fences
+// take it this way: a poll is bounded once it holds the gate — a build read
+// by its own timeout, a WorkSource read by its contract — so a fence waits
+// only for that poll, and the executor's verdict never depends on an
+// observer's patience.
 func (t *Tracker) takePollGate() {
 	t.pollGate <- struct{}{}
 }
@@ -293,6 +297,16 @@ var (
 	// other sentinels CancelBuild returns also mean nothing was delivered;
 	// see CancelBuild for the full outcome table.
 	ErrCancelNotDispatched = errors.New("the cancel signal was not sent")
+	// ErrProgressNotRead is returned by Progress when the caller's context
+	// ended before the read was issued — while it waited for another poll
+	// or a handoff fence to release the poll gate, or in the instant it took
+	// the gate. It is returned wrapped together with the caller's context
+	// error, so errors.Is matches both; test for this sentinel before
+	// testing for the context error. A context error without it came from
+	// the read itself: a concurrent build's read runs detached from the
+	// caller's context, so its deadline error means the reserved verdict
+	// session stopped answering, not that the caller timed out.
+	ErrProgressNotRead = errors.New("the progress read was not issued")
 )
 
 // cancelSignalTimeout bounds the cancel signal's round trip. The signal
@@ -302,6 +316,19 @@ var (
 // the outcome the reserved session exists to prevent.
 const cancelSignalTimeout = 5 * time.Second
 
+// progressReadTimeout bounds one progress read on the reserved verdict
+// session, which runs detached from the observer's context for the same
+// reason the cancel signal does. A client-side deadline that fires mid-query
+// closes the connection, so the bound is a backstop, set above the pool's
+// baseline statement_timeout: a slow read ends at the server with a query
+// cancellation and leaves the session intact, and only a network that has
+// stopped answering — a session lost either way — reaches this deadline.
+// The margin holds for a pool built by pkg/dbconn with the default or a lower
+// statement_timeout; a pool configured above this bound, or one whose
+// sessions carry no statement_timeout, reaches the deadline first, and a
+// read that stalls that long closes the session.
+const progressReadTimeout = 2 * dbconn.DefaultStatementTimeout
+
 // cancelBuildSQL reads the build backend's state and signals it only when
 // the server reports it active, in one statement so the read and the
 // signal cannot straddle a state change the tracker then misreports.
@@ -499,17 +526,11 @@ func (t *Tracker) Finish(err error) {
 	t.session, t.buildPID, t.source = nil, 0, nil
 }
 
-// Progress returns a snapshot and, for an active concurrent index build,
-// queries PostgreSQL's progress view by the executor-owned backend PID; for
-// a step with a WorkSource it asks the source for the step's counters. On a
-// query or source error the snapshot still carries the last-known tracker
-// state. The state lock is released before the query, so concurrent pollers
-// serialize only against each other (and the four handoff fences), never
-// against the executor's own state updates.
-func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
-	t.takePollGate()
-	defer t.releasePollGate()
+// readState snapshots the tracker under one lock acquisition: the snapshot,
+// and the poll target it was taken against.
+func (t *Tracker) readState() (Snapshot, dbconn.RowQuerier, uint32, WorkSource) {
 	t.mu.RLock()
+	defer t.mu.RUnlock()
 	now := t.clock.Now()
 	if !t.ended.IsZero() {
 		now = t.ended
@@ -519,8 +540,43 @@ func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
 		s.Elapsed = now.Sub(t.started)
 		s.StepElapsed = now.Sub(t.stepStart)
 	}
-	session, pid, source := t.session, t.buildPID, t.source
-	t.mu.RUnlock()
+	return s, t.session, t.buildPID, t.source
+}
+
+// Progress returns a snapshot and, for an active concurrent index build,
+// queries PostgreSQL's progress view by the executor-owned backend PID; for
+// a step with a WorkSource it asks the source for the step's counters. On a
+// query or source error the snapshot still carries the last-known tracker
+// state. The state lock is released before the query, so concurrent pollers
+// serialize only against each other (and the four handoff fences), never
+// against the executor's own state updates.
+//
+// A poller waits for the poll gate for as long as its ctx lasts; one that
+// gives up first, or whose ctx has already ended when it takes the gate,
+// gets the last-known snapshot and ErrProgressNotRead wrapped with ctx's
+// error, having read nothing. Once the gate is held, a concurrent build's
+// progress read runs on the executor's reserved verdict session bounded and
+// detached from ctx, so an observer whose context ends mid-read — an HTTP
+// request timing out — cannot tear down the session the build's failure
+// verdict needs. Progress is therefore not bounded by ctx once the read is
+// issued: it returns when the server answers, which on a pkg/dbconn pool is
+// within its statement_timeout, and within twice dbconn's default
+// statement_timeout (one minute) at the latest. A caller that needs a hard
+// return at its own deadline calls Progress from its own goroutine. A
+// WorkSource is read under ctx, as its contract requires.
+func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
+	if err := t.takePollGateOrGiveUp(ctx); err != nil {
+		s, _, _, _ := t.readState()
+		return s, fmt.Errorf("%w: %w", ErrProgressNotRead, err)
+	}
+	defer t.releasePollGate()
+	if err := ctx.Err(); err != nil {
+		// The caller gave up in the instant the gate came free; the gate is
+		// held, so this is the last point at which nothing has been read.
+		s, _, _, _ := t.readState()
+		return s, fmt.Errorf("%w: %w", ErrProgressNotRead, err)
+	}
+	s, session, pid, source := t.readState()
 	if s.Phase != PhaseRunning {
 		return s, nil
 	}
@@ -530,7 +586,13 @@ func (t *Tracker) Progress(ctx context.Context) (Snapshot, error) {
 	if pid == 0 || session == nil {
 		return s, nil
 	}
-	p, active, err := dbconn.ConcurrentIndexProgress(ctx, session, pid)
+	// INV: LK-2 — the poll is bounded without being cancelled by its
+	// observer: the reserved session's statement_timeout ends a slow read
+	// server-side and leaves the session usable, and progressReadTimeout
+	// backstops a network that has stopped answering.
+	pollCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), progressReadTimeout)
+	defer cancel()
+	p, active, err := dbconn.ConcurrentIndexProgress(pollCtx, session, pid)
 	if err != nil {
 		return s, err
 	}