diff --git a/docs/execution-model.md b/docs/execution-model.md index 2227c24..764905c 100644 --- a/docs/execution-model.md +++ b/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)) | diff --git a/docs/invariants.md b/docs/invariants.md index 17a82c5..c1c2410 100644 --- a/docs/invariants.md +++ b/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 | diff --git a/docs/lock-budgeted-passthrough.md b/docs/lock-budgeted-passthrough.md index 9a694e0..357e023 100644 --- a/docs/lock-budgeted-passthrough.md +++ b/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 diff --git a/pkg/executor/accepted_blocking.go b/pkg/executor/accepted_blocking.go new file mode 100644 index 0000000..f37d363 --- /dev/null +++ b/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} +} diff --git a/pkg/executor/accepted_blocking_integration_test.go b/pkg/executor/accepted_blocking_integration_test.go new file mode 100644 index 0000000..788795c --- /dev/null +++ b/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)) +} diff --git a/pkg/executor/accepted_blocking_internal_test.go b/pkg/executor/accepted_blocking_internal_test.go new file mode 100644 index 0000000..0837807 --- /dev/null +++ b/pkg/executor/accepted_blocking_internal_test.go @@ -0,0 +1,103 @@ +package executor + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/statement" +) + +func TestBlockingBudgetValidation(t *testing.T) { + valid := BlockingBudget{LockTimeout: time.Second, StatementTimeout: time.Minute} + tests := []struct { + name string + budget BlockingBudget + ok bool + }{ + {name: "zero lock", budget: BlockingBudget{StatementTimeout: time.Second}}, + {name: "negative lock", budget: BlockingBudget{LockTimeout: -time.Second, StatementTimeout: time.Second}}, + {name: "lock over ceiling", budget: BlockingBudget{LockTimeout: maxOverallBudget + time.Millisecond, StatementTimeout: time.Second}}, + {name: "zero statement", budget: BlockingBudget{LockTimeout: time.Second}}, + {name: "negative statement", budget: BlockingBudget{LockTimeout: time.Second, StatementTimeout: -time.Second}}, + {name: "statement over ceiling", budget: BlockingBudget{LockTimeout: time.Second, StatementTimeout: maxOverallBudget + time.Millisecond}}, + {name: "valid", budget: valid, ok: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.budget.validate() + if tt.ok { + require.NoError(t, err) + return + } + require.ErrorIs(t, err, ErrInvalidBlockingBudget) + assert.Equal(t, CodeInvalidBlockingBudget, OutcomeCode(err)) + }) + } +} + +func TestAcceptedBlockingStatementErrorMapsSQLSTATE(t *testing.T) { + b := BlockingBudget{LockTimeout: time.Second, StatementTimeout: time.Minute} + tests := []struct { + name string + code string + want Code + }{ + {name: "lock", code: sqlstateLockNotAvailable, want: CodeBudgetLockExceeded}, + {name: "statement", code: sqlstateQueryCanceled, want: CodeBudgetStatementExceeded}, + {name: "cannot run in transaction block", code: sqlstateActiveSQLTransaction, want: CodeUnsupportedAcceptedBlocking}, + {name: "other postgres", code: "23505", want: CodeExecutionFailed}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := acceptedBlockingStatementError(t.Context(), &pgconn.PgError{Code: tt.code}, b) + assert.Equal(t, tt.want, OutcomeCode(err)) + }) + } + + // A statement the server will not run inside a transaction block fails + // the same way on every attempt, so its outcome must be permanent and + // must keep the server's SQLSTATE for the operator. + serverErr := &pgconn.PgError{Code: sqlstateActiveSQLTransaction, Message: "REINDEX TABLE cannot run inside a transaction block"} + err := acceptedBlockingStatementError(t.Context(), serverErr, b) + require.ErrorIs(t, err, ErrUnsupportedAcceptedBlocking) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr) + assert.Equal(t, sqlstateActiveSQLTransaction, pgErr.Code) + assert.True(t, OutcomeCode(err).Permanent()) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + err = acceptedBlockingStatementError(ctx, errors.New("connection lost"), b) + var unknown *BlockingOutcomeUnknownError + require.ErrorAs(t, err, &unknown) + assert.Equal(t, CodeBlockingOutcomeUnknown, OutcomeCode(err)) +} + +func TestAcceptedBlockingAdmission(t *testing.T) { + tests := []struct { + sql string + want bool + }{ + {sql: "DROP INDEX app.i", want: true}, + {sql: "REINDEX INDEX app.i", want: true}, + {sql: "REINDEX TABLE app.t", want: true}, + {sql: "CREATE INDEX i ON app.t (id)", want: true}, + {sql: "DROP INDEX CONCURRENTLY app.i"}, + {sql: "DROP INDEX app.i, app.j"}, + {sql: "REINDEX SCHEMA app"}, + {sql: "DELETE FROM app.t"}, + } + for _, tt := range tests { + t.Run(tt.sql, func(t *testing.T) { + st, err := statement.ParseOne(tt.sql) + require.NoError(t, err) + assert.Equal(t, tt.want, acceptedBlockingShape(st)) + }) + } +} diff --git a/pkg/executor/code.go b/pkg/executor/code.go index 80f808c..bac8f89 100644 --- a/pkg/executor/code.go +++ b/pkg/executor/code.go @@ -23,6 +23,12 @@ const ( // CodeBudgetStatementExceeded: the statement ran past // statement_timeout and was cancelled; the change does real work. CodeBudgetStatementExceeded Code = "budget-statement-exceeded" + // CodeBlockingOutcomeUnknown requires catalog inspection before retry. + CodeBlockingOutcomeUnknown Code = "blocking-outcome-unknown" + // CodeInvalidBlockingBudget identifies an unrepresentable or disabled bound. + CodeInvalidBlockingBudget Code = "invalid-blocking-budget" + // CodeUnsupportedAcceptedBlocking identifies a statement outside this narrow path. + CodeUnsupportedAcceptedBlocking Code = "unsupported-accepted-blocking" // CodeCancelledByCaller: the build's statement was cancelled because // the caller's own context ended while it ran. CodeCancelledByCaller Code = "cancelled-by-caller" @@ -130,6 +136,9 @@ func Codes() []Code { return []Code{ CodeBudgetLockExceeded, CodeBudgetStatementExceeded, + CodeBlockingOutcomeUnknown, + CodeInvalidBlockingBudget, + CodeUnsupportedAcceptedBlocking, CodeCancelledByCaller, CodeCancelledExternally, CodeInvalidIndexOwnLeftover, @@ -175,6 +184,8 @@ func (c Code) Permanent() bool { switch c { case CodeInvalidIndexOtherTable, CodeInvalidIndexNotDroppable, + CodeInvalidBlockingBudget, + CodeUnsupportedAcceptedBlocking, CodeEmptySequence, CodeUnsupportedSequenceStep, CodeUnsupportedPartitionedParent, @@ -217,6 +228,10 @@ func OutcomeCode(err error) Code { if errors.As(err, &budgetErr) { return budgetErr.Code() } + var unknownErr *BlockingOutcomeUnknownError + if errors.As(err, &unknownErr) { + return CodeBlockingOutcomeUnknown + } return sentinelCode(err) } @@ -234,6 +249,10 @@ func sentinelCode(err error) Code { // The read's cause rides inside the wrap; the code names what the // step left — a standing table whose names are unproven. return CodeCreateNamesUnverified + case errors.Is(err, ErrInvalidBlockingBudget): + return CodeInvalidBlockingBudget + case errors.Is(err, ErrUnsupportedAcceptedBlocking): + return CodeUnsupportedAcceptedBlocking case errors.Is(err, ErrCancelledByCaller): return CodeCancelledByCaller case errors.Is(err, ErrCancelledExternally): diff --git a/pkg/executor/code_test.go b/pkg/executor/code_test.go index d7c9169..4ee3037 100644 --- a/pkg/executor/code_test.go +++ b/pkg/executor/code_test.go @@ -66,6 +66,9 @@ func TestOutcomeCodeMapsTypedOutcomes(t *testing.T) { want: executor.CodeInvalidIndexUnproven, }, {name: "cancelled by caller", err: executor.ErrCancelledByCaller, want: executor.CodeCancelledByCaller}, + {name: "blocking outcome unknown", err: &executor.BlockingOutcomeUnknownError{Err: errors.New("lost")}, want: executor.CodeBlockingOutcomeUnknown}, + {name: "invalid blocking budget", err: executor.ErrInvalidBlockingBudget, want: executor.CodeInvalidBlockingBudget}, + {name: "unsupported accepted blocking", err: executor.ErrUnsupportedAcceptedBlocking, want: executor.CodeUnsupportedAcceptedBlocking}, { name: "abandonment unproven through to removal", err: &executor.InvalidIndexError{Schema: "s", Index: "i", Cleanup: executor.ErrAbandonmentUnproven}, @@ -138,6 +141,9 @@ func TestCodePermanentClassifiesEveryCode(t *testing.T) { want := map[executor.Code]bool{ executor.CodeBudgetLockExceeded: false, executor.CodeBudgetStatementExceeded: false, + executor.CodeBlockingOutcomeUnknown: false, + executor.CodeInvalidBlockingBudget: true, + executor.CodeUnsupportedAcceptedBlocking: true, executor.CodeCancelledByCaller: false, executor.CodeCancelledExternally: false, executor.CodeInvalidIndexOwnLeftover: false,