feat(executor): run accepted blocking statements under both budgets

docs/execution-model.md

@@ -286,6 +286,9 @@ concurrently is.
 | --- | --- | --- |
 | `budget-lock-exceeded` | no | The lock was not granted within `lock_timeout`; nothing executed |
 | `budget-statement-exceeded` | no | The statement ran past `statement_timeout` and was cancelled |
+| `blocking-outcome-unknown` | no | The accepted blocking transaction reached an ambiguous client boundary; inspect the catalog before retrying |
+| `invalid-blocking-budget` | yes | An accepted blocking bound is disabled or cannot be represented by PostgreSQL |
+| `unsupported-accepted-blocking` | yes | The statement is outside the accepted blocking executor's narrow index-maintenance set, or the server will not run it inside the engine-owned transaction (`REINDEX` on a partitioned relation, SQLSTATE `25001`) |
 | `cancelled-by-caller` | no | The caller's own context ended while the statement ran and the budget had not elapsed; in caller-owned mode this is the build's ordinary exit |
 | `cancelled-externally` | no | The statement was cancelled from outside the executor — not by its caller and not by its budget; an operator's `pg_cancel_backend` or `Tracker.CancelBuild` |
 | `invalid-index-own-leftover` | no | The failed build's own INVALID index remains; `RebuildAbandonedIndex` removes it under proof ([recovery runbook](invalid-index-recovery.md)) |

docs/invariants.md

@@ -21,6 +21,7 @@ several of these unrepresentable, and the in-TCB engineering rules live in
 
 - [Correctness (CO)](#correctness-co)
 - [Locking and concurrency (LK)](#locking-and-concurrency-lk)
+- [Accepted blocking execution (AB)](#accepted-blocking-execution-ab)
 - [State, checkpoint, and resume (ST)](#state-checkpoint-and-resume-st)
 - [Refusals and preflight (RF)](#refusals-and-preflight-rf)
 - [Orchestration / control-plane (OC)](#orchestration--control-plane-oc)
@@ -295,6 +296,24 @@ stale-observation tests that alter the catalog between observation and lock on a
 database. *Source:* PostgreSQL's session-level `ShareUpdateExclusiveLock` on the heap for
 every `CONCURRENTLY` index command; [invalid-index-recovery](invalid-index-recovery.md).
 
+## Accepted blocking execution (AB)
+
+### AB-1 — Every accepted blocking statement runs under both engine-owned bounds
+
+Every accepted blocking statement runs in one engine-owned session and transaction with an
+explicit, non-zero `lock_timeout` and `statement_timeout`. Transaction-local settings override
+ambient defaults, and an absent, sub-millisecond, or server-unrepresentable bound is refused
+before a session is acquired. *Enforced:* `pkg/executor` (`ExecuteAcceptedBlocking`). *Source:*
+[lock-budgeted passthrough](lock-budgeted-passthrough.md#engine-owned-session-and-budgets).
+
+### AB-2 — Lock-budget exhaustion executes nothing
+
+An accepted blocking statement that cannot acquire its lock within `lock_timeout` is not
+retried: PostgreSQL aborts that transaction before the DDL executes, and the executor returns
+the typed lock-budget outcome. *Enforced:* `pkg/executor` (`ExecuteAcceptedBlocking`, SQLSTATE
+`55P03`). *Source:*
+[lock-budgeted passthrough](lock-budgeted-passthrough.md#failure-and-interruption-semantics).
+
 ## State, checkpoint, and resume (ST)
 
 ### ST-1 — The checkpoint is one row per target, written atomically
@@ -515,6 +534,7 @@ about **how we write and review the code**.
 | CO-9 | 3 onward | shadowing-search_path tests per read site and per pooled session |
 | LK-3 | 4–6 | cancellation/claim race test |
 | LK-5 | 3 (native recovery) | stale-observation fail-closed tests, never-drops-valid, not-droppable skip, shared-budget test |
+| AB-1, AB-2 | accepted-blocking rollout step 2 | exact session bounds + lock exhaustion leaves catalog unchanged |
 | LK-4, ST-5 | 7 | dropped-connection cutover, fidelity checklist |
 | ST-1, ST-2, ST-3, ST-4 | 8 | kill/resume, cross-version refuse, orphan-slot reap, failover reconcile |
 | ST-6 | 1 onward, complete by 8 | preflight matrix |

docs/lock-budgeted-passthrough.md

@@ -184,7 +184,7 @@ The v1 set is:
 
 | Class | Typed refusal shape | v1 | Rationale |
 | --- | --- | --- | --- |
-| `by-design` | `index-statement` at the plain single-relation `DROP INDEX`, `REINDEX INDEX`, or `REINDEX TABLE` site, with the concurrent safer idiom | eligible | This is the core maintenance-window case: the submitted form is understood, the safer idiom is known, one table's lock is being accepted, and bounding `ACCESS EXCLUSIVE` acquisition prevents lock-queue pile-up. The multi-relation sites (`DROP INDEX a, b`, `REINDEX SCHEMA`, `DATABASE`, `SYSTEM`) stay ineligible. |
+| `by-design` | `index-statement` at the plain single-relation `DROP INDEX`, `REINDEX INDEX`, or `REINDEX TABLE` site, with the concurrent safer idiom | eligible | This is the core maintenance-window case: the submitted form is understood, the safer idiom is known, one table's lock is being accepted, and bounding `ACCESS EXCLUSIVE` acquisition prevents lock-queue pile-up. The multi-relation sites (`DROP INDEX a, b`, `REINDEX SCHEMA`, `DATABASE`, `SYSTEM`) stay ineligible. `REINDEX` on a partitioned table or index is eligible by shape but cannot run inside the engine-owned transaction; the executor reports it as `unsupported-accepted-blocking` (see [Engine-owned session and budgets](#engine-owned-session-and-budgets)). |
 | `capability-boundary` | `unsupported-partitioned-parent` with cause `parent-blocking-index-build` | eligible | The missing partition-aware flow does not make the plain PostgreSQL statement unknown. Its principal pre-execution hazard is acquiring the parent lock, which `lock_timeout` bounds. |
 | `capability-boundary` | `not-native-safe-rewrite-required` | ineligible | A rewrite holds its strong lock for the full rewrite. Bounding acquisition alone does not bound the outage, and v1 must not imply otherwise. |
 | `capability-boundary` | `backend-unavailable` and all other missing routes or backends | ineligible | A missing execution backend is not permission to substitute an unrelated blocking implementation. |
@@ -208,11 +208,25 @@ copy-and-swap is not a reason to add passthrough eligibility; it removes the ref
 
 ## Engine-owned session and budgets
 
-An eligible statement runs as one statement in one engine-owned session and transaction,
-using the same bounded runner and retry policy as other brief native execution. The runner
-sets `lock_timeout` and `statement_timeout` in the transaction before the statement. It does
-not hand SQL to a shell, inherit an unbounded caller session, or use the caller-owned
-concurrent-index exception.
+An eligible statement runs as one statement in one engine-owned session and transaction, in
+exactly one attempt. The runner sets `lock_timeout` and `statement_timeout` in the transaction
+before the statement. It does not hand SQL to a shell, inherit an unbounded caller session, or
+use the caller-owned concurrent-index exception.
+
+The single attempt is deliberate and differs from brief native execution, which retries a
+lock-budget miss up to three times. The operator accepted one bounded `ACCESS EXCLUSIVE`
+acquisition; a second attempt would re-queue behind the same holder, stack another
+`lock_timeout` of blocked readers and writers behind the engine's request, and turn the stated
+budget into a multiple of itself. An exhausted lock budget is therefore the final typed outcome
+([AB-2](invariants.md#ab-2--lock-budget-exhaustion-executes-nothing)); the operator decides
+whether to run the command again.
+
+The engine-owned transaction is also the path's boundary. `REINDEX TABLE` and `REINDEX INDEX`
+on a partitioned relation are single-relation by shape, but PostgreSQL reindexes each partition
+in its own transaction and refuses to start inside a transaction block (SQLSTATE `25001`)
+before touching any partition. The executor reports that server refusal as
+`unsupported-accepted-blocking`, a permanent outcome, rather than as an execution failure an
+adapter might retry.
 
 Both bounds are required and non-zero. `--accept-blocking` rejects an omitted, zero, or
 disabled `statement_timeout`; it does not silently inherit an unbounded value. The ordinary
@@ -403,7 +417,9 @@ Sequence implementation as follows:
    refusal identity retained". This design establishes no invariant on its own and amends none
    until the behavior ships; the registry describes shipped behavior only, matching the
    sequencing rule in [refusal-classes.md](refusal-classes.md), whose own rollout record
-   checked RF-5 and RF-6 and recorded them unchanged.
+   checked RF-5 and RF-6 and recorded them unchanged. *(done: the executor-owned AB-1 and AB-2
+   invariants ship here; the front-door invariants, refusal-identity invariant, and RF-5/RF-6
+   amendments move to step 4 with the flag.)*
 3. Add `executed-without-online-safety`, retained reason/class/cause, budget fields, exit code
    3, and dry-run eligibility to the verdict and plan-report contracts. Update
    [cli-output-examples.md](cli-output-examples.md) with generated examples and pin the JSON

pkg/executor/accepted_blocking.go

@@ -0,0 +1,164 @@
+package executor
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/block/pg-sprite/pkg/progress"
+	"github.com/block/pg-sprite/pkg/statement"
+)
+
+var (
+	// ErrInvalidBlockingBudget means an accepted-blocking bound would be
+	// disabled or cannot be represented by PostgreSQL.
+	ErrInvalidBlockingBudget = errors.New("invalid accepted-blocking budget")
+	// ErrUnsupportedAcceptedBlocking means the statement is not one of the
+	// single-relation blocking index forms this executor admits, or the
+	// server will not run it inside the engine-owned transaction. Retrying
+	// the same statement reproduces it; the outcome is permanent.
+	ErrUnsupportedAcceptedBlocking = errors.New("statement is not an accepted blocking index statement")
+)
+
+// sqlstateActiveSQLTransaction is raised when a statement that must own its
+// transaction is submitted inside a transaction block. In the admitted set
+// only REINDEX on a partitioned table or index does this: PostgreSQL
+// reindexes each partition in its own transaction and refuses before
+// touching any of them.
+const sqlstateActiveSQLTransaction = "25001"
+
+// BlockingBudget bounds one operator-accepted blocking statement.
+type BlockingBudget struct {
+	LockTimeout      time.Duration
+	StatementTimeout time.Duration
+}
+
+func (b BlockingBudget) validate() error {
+	// INV: AB-1 — both limits are non-zero and representable before a
+	// session is acquired; whole milliseconds avoid PostgreSQL's zero/off
+	// truncation.
+	if b.LockTimeout < minBudget || b.LockTimeout > maxOverallBudget {
+		return fmt.Errorf("%w: lock timeout must be between %s and %s, got %s", ErrInvalidBlockingBudget, minBudget, maxOverallBudget, b.LockTimeout)
+	}
+	if b.StatementTimeout < minBudget || b.StatementTimeout > maxOverallBudget {
+		return fmt.Errorf("%w: statement timeout must be between %s and %s, got %s", ErrInvalidBlockingBudget, minBudget, maxOverallBudget, b.StatementTimeout)
+	}
+	return nil
+}
+
+// BlockingReport records a committed accepted-blocking statement and its
+// engine-owned bounds.
+type BlockingReport struct {
+	SQL              string        `json:"sql"`
+	LockTimeout      time.Duration `json:"lock_timeout_ns"`
+	StatementTimeout time.Duration `json:"statement_timeout_ns"`
+	Duration         time.Duration `json:"duration_ns"`
+}
+
+// BlockingExecutionError reports a PostgreSQL failure after the accepted
+// statement was submitted. Err retains the server SQLSTATE.
+type BlockingExecutionError struct{ Err error }
+
+func (e *BlockingExecutionError) Error() string {
+	return fmt.Sprintf("accepted blocking statement failed: %v", e.Err)
+}
+func (e *BlockingExecutionError) Unwrap() error { return e.Err }
+
+// BlockingOutcomeUnknownError means the client cannot establish whether
+// the transaction committed; the catalog must be inspected before retrying.
+type BlockingOutcomeUnknownError struct{ Err error }
+
+func (e *BlockingOutcomeUnknownError) Error() string {
+	return fmt.Sprintf("accepted blocking statement outcome is unknown until the catalog is inspected: %v", e.Err)
+}
+func (e *BlockingOutcomeUnknownError) Unwrap() error { return e.Err }
+
+// ExecuteAcceptedBlocking runs one admitted blocking index statement in one
+// engine-owned transaction with explicit lock and statement bounds.
+func ExecuteAcceptedBlocking(ctx context.Context, pool *pgxpool.Pool, sql string, b BlockingBudget) (BlockingReport, error) {
+	return executeAcceptedBlocking(ctx, pool, sql, b, progress.WallClock{})
+}
+
+func executeAcceptedBlocking(ctx context.Context, pool *pgxpool.Pool, sql string, b BlockingBudget, clock progress.Clock) (rep BlockingReport, err error) {
+	if err := b.validate(); err != nil {
+		return rep, err
+	}
+	st, err := statement.ParseOne(sql)
+	if err != nil {
+		return rep, fmt.Errorf("%w: %w", ErrUnsupportedAcceptedBlocking, err)
+	}
+	if !acceptedBlockingShape(st) {
+		return rep, ErrUnsupportedAcceptedBlocking
+	}
+	rep = BlockingReport{SQL: sql, LockTimeout: b.LockTimeout, StatementTimeout: b.StatementTimeout}
+	start := clock.Now()
+	tx, err := pool.Begin(ctx)
+	if err != nil {
+		return BlockingReport{}, fmt.Errorf("begin accepted blocking statement: %w", err)
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+
+	// INV: AB-1 — transaction-local values override every caller/session
+	// default on the same engine-owned session that executes the statement.
+	settings := "SET LOCAL lock_timeout = " + strconv.FormatInt(b.LockTimeout.Milliseconds(), 10) +
+		"; SET LOCAL statement_timeout = " + strconv.FormatInt(b.StatementTimeout.Milliseconds(), 10)
+	if _, err := tx.Exec(ctx, settings); err != nil {
+		return BlockingReport{}, fmt.Errorf("set accepted blocking budgets: %w", err)
+	}
+	if _, err := tx.Exec(ctx, sql); err != nil {
+		rep.Duration = clock.Now().Sub(start)
+		return rep, acceptedBlockingStatementError(ctx, err, b)
+	}
+	if err := tx.Commit(ctx); err != nil {
+		rep.Duration = clock.Now().Sub(start)
+		return rep, &BlockingOutcomeUnknownError{Err: err}
+	}
+	rep.Duration = clock.Now().Sub(start)
+	return rep, nil
+}
+
+func acceptedBlockingShape(st statement.Statement) bool {
+	if st.Concurrent() {
+		return false
+	}
+	switch st.Kind() {
+	case statement.KindCreateIndex:
+		return true
+	case statement.KindDropIndex, statement.KindReindex:
+		return st.IndexTarget() == statement.IndexTargetSingleRelation
+	default:
+		return false
+	}
+}
+
+func acceptedBlockingStatementError(ctx context.Context, err error, b BlockingBudget) error {
+	if ctx.Err() != nil {
+		return &BlockingOutcomeUnknownError{Err: err}
+	}
+	var pgErr *pgconn.PgError
+	if errors.As(err, &pgErr) {
+		switch pgErr.Code {
+		case sqlstateLockNotAvailable:
+			// INV: AB-2 — PostgreSQL rejected lock acquisition and aborted the
+			// transaction before the submitted DDL could execute.
+			return &BudgetError{Cause: CauseLock, Budget: b.LockTimeout, Attempts: 1, cause: err}
+		case sqlstateQueryCanceled:
+			return &BudgetError{Cause: CauseStatement, Budget: b.StatementTimeout, Attempts: 1, cause: err}
+		case sqlstateActiveSQLTransaction:
+			// The server refused the statement before it executed, and it
+			// will refuse it every time: the engine-owned transaction is the
+			// only way this executor runs anything. Report the statement as
+			// outside the path rather than as a retryable execution failure.
+			return fmt.Errorf("%w: the server will not run it inside the engine-owned transaction: %w",
+				ErrUnsupportedAcceptedBlocking, err)
+		default:
+			return &BlockingExecutionError{Err: err}
+		}
+	}
+	return &BlockingExecutionError{Err: err}
+}

pkg/executor/accepted_blocking_integration_test.go

@@ -0,0 +1,109 @@
+package executor_test
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/pg-sprite/pkg/executor"
+)
+
+func TestExecuteAcceptedBlockingDropsIndex(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int); CREATE INDEX i ON %s.t (id)", schema, schema))
+	require.NoError(t, err)
+	b := executor.BlockingBudget{LockTimeout: time.Second, StatementTimeout: 10 * time.Second}
+	sql := fmt.Sprintf("DROP INDEX %s.i", schema)
+
+	rep, err := executor.ExecuteAcceptedBlocking(t.Context(), pool, sql, b)
+	require.NoError(t, err)
+	assert.Equal(t, sql, rep.SQL)
+	assert.Equal(t, b.LockTimeout, rep.LockTimeout)
+	assert.Equal(t, b.StatementTimeout, rep.StatementTimeout)
+	assert.Positive(t, rep.Duration)
+	exists, _ := indexState(t, pool, schema, "i")
+	assert.False(t, exists)
+}
+
+func TestExecuteAcceptedBlockingLockBudgetExecutesNothing(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int); CREATE INDEX i ON %s.t (id)", schema, schema))
+	require.NoError(t, err)
+	holder, err := pool.Begin(t.Context())
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = holder.Rollback(t.Context()) })
+	_, err = holder.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema))
+	require.NoError(t, err)
+	b := executor.BlockingBudget{LockTimeout: 100 * time.Millisecond, StatementTimeout: 5 * time.Second}
+	start := time.Now()
+
+	_, err = executor.ExecuteAcceptedBlocking(t.Context(), pool, fmt.Sprintf("DROP INDEX %s.i", schema), b)
+	var budgetErr *executor.BudgetError
+	require.ErrorAs(t, err, &budgetErr)
+	assert.Equal(t, executor.CauseLock, budgetErr.Cause)
+	assert.Less(t, time.Since(start), 2*time.Second)
+	exists, valid := indexState(t, pool, schema, "i")
+	assert.True(t, exists)
+	assert.True(t, valid)
+}
+
+// REINDEX on a partitioned relation is admitted by shape but PostgreSQL
+// refuses to run it inside a transaction block, which is the only way this
+// executor runs anything. The failure is reported as a permanent outcome
+// outside the path, not as a retryable execution failure, and nothing is
+// changed: the parent and its partitions keep their indexes.
+func TestExecuteAcceptedBlockingRefusesReindexOnPartitionedRelation(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf(`
+		CREATE TABLE %[1]s.parent (id int) PARTITION BY RANGE (id);
+		CREATE TABLE %[1]s.leaf PARTITION OF %[1]s.parent FOR VALUES FROM (0) TO (10);
+		CREATE INDEX parent_i ON %[1]s.parent (id)`, schema))
+	require.NoError(t, err)
+	b := executor.BlockingBudget{LockTimeout: time.Second, StatementTimeout: 10 * time.Second}
+
+	for _, sql := range []string{
+		fmt.Sprintf("REINDEX TABLE %s.parent", schema),
+		fmt.Sprintf("REINDEX INDEX %s.parent_i", schema),
+	} {
+		t.Run(sql, func(t *testing.T) {
+			_, err := executor.ExecuteAcceptedBlocking(t.Context(), pool, sql, b)
+			require.ErrorIs(t, err, executor.ErrUnsupportedAcceptedBlocking)
+			var pgErr *pgconn.PgError
+			require.ErrorAs(t, err, &pgErr)
+			assert.Equal(t, "25001", pgErr.Code)
+			assert.True(t, executor.OutcomeCode(err).Permanent())
+		})
+	}
+	exists, valid := indexState(t, pool, schema, "parent_i")
+	assert.True(t, exists)
+	assert.True(t, valid)
+}
+
+func TestExecuteAcceptedBlockingAppliesBothBoundsOnExecutingSession(t *testing.T) {
+	pool, schema := newPool(t)
+	_, err := pool.Exec(t.Context(), fmt.Sprintf(`
+		CREATE TABLE %s.t (id int);
+		INSERT INTO %s.t VALUES (1);
+		CREATE FUNCTION %s.observe_bounds(int) RETURNS int LANGUAGE plpgsql IMMUTABLE AS $$
+		BEGIN
+			IF current_setting('lock_timeout') = '137ms' AND current_setting('statement_timeout') = '2468ms' THEN
+				RAISE EXCEPTION USING ERRCODE = 'P0001';
+			END IF;
+			RAISE EXCEPTION USING ERRCODE = 'P0002';
+		END $$`, schema, schema, schema))
+	require.NoError(t, err)
+	b := executor.BlockingBudget{LockTimeout: 137 * time.Millisecond, StatementTimeout: 2468 * time.Millisecond}
+
+	_, err = executor.ExecuteAcceptedBlocking(t.Context(), pool,
+		fmt.Sprintf("CREATE INDEX observed_i ON %s.t (%s.observe_bounds(id))", schema, schema), b)
+	var pgErr *pgconn.PgError
+	require.ErrorAs(t, err, &pgErr)
+	assert.Equal(t, "P0001", pgErr.Code, "the executing transaction must observe both exact bounds")
+	var executionErr *executor.BlockingExecutionError
+	assert.True(t, errors.As(err, &executionErr))
+}