diff --git a/SAFETY.md b/SAFETY.md index 6c66f3ea..597cf919 100644 --- a/SAFETY.md +++ b/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), diff --git a/docs/invariants.md b/docs/invariants.md index ac559033..93514677 100644 --- a/docs/invariants.md +++ b/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. diff --git a/docs/low-level-design.md b/docs/low-level-design.md index 69812677..c438d6b8 100644 --- a/docs/low-level-design.md +++ b/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 diff --git a/docs/progress-report.md b/docs/progress-report.md index 048fe1b7..6d2f35a0 100644 --- a/docs/progress-report.md +++ b/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 diff --git a/pkg/executor/cic_progress_poll_integration_test.go b/pkg/executor/cic_progress_poll_integration_test.go new file mode 100644 index 00000000..467b2730 --- /dev/null +++ b/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") +} diff --git a/pkg/progress/progress.go b/pkg/progress/progress.go index 88a89c41..54fd7f36 100644 --- a/pkg/progress/progress.go +++ b/pkg/progress/progress.go @@ -138,8 +138,10 @@ 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 @@ -147,8 +149,8 @@ type Snapshot struct { // 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 } diff --git a/pkg/progress/progress_test.go b/pkg/progress/progress_test.go index 7b10e7dd..1a16905e 100644 --- a/pkg/progress/progress_test.go +++ b/pkg/progress/progress_test.go @@ -15,6 +15,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/progress" ) @@ -826,6 +827,108 @@ func TestProgressReturnsSnapshotAlongsideQueryError(t *testing.T) { 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) { diff --git a/pkg/progress/work_source_test.go b/pkg/progress/work_source_test.go index ca8934dd..3d3c4682 100644 --- a/pkg/progress/work_source_test.go +++ b/pkg/progress/work_source_test.go @@ -497,3 +497,20 @@ func TestStateMutatorsDoNotWaitForInFlightWorkSourcePoll(t *testing.T) { t.Fatal("a state mutator waited behind an in-flight work source poll") } } + +// A WorkSource is the engine's own read, not the reserved session, so it is +// read under the observer's context: an observer that gives up must not keep +// a source poll, and the poll gate with it, alive for the source's whole read. +func TestProgressReadsTheWorkSourceUnderTheObserversContext(t *testing.T) { + observer, giveUp := context.WithCancel(t.Context()) + var sawEndedContext atomic.Bool + tracker := runningTrackerWithSource(t, fakeSource{work: func(ctx context.Context) (progress.Work, error) { + giveUp() + sawEndedContext.Store(ctx.Err() != nil) + return progress.Work{}, nil + }}) + + _, err := tracker.Progress(observer) + require.NoError(t, err) + assert.True(t, sawEndedContext.Load(), "the work source must be read under the observer's context") +}