From 65f819c5d0cd8689124dccf73ced12e102302cc0 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 12 Aug 2026 11:27:02 +1000 Subject: [PATCH 1/2] executor: run planner-produced safer sequences natively (Phase 3.2) Adds the sequence executor for the remaining native idioms: NOT VALID + VALIDATE CONSTRAINT, ADD PRIMARY KEY USING INDEX over a concurrent build, the four-step SET NOT NULL, fast-default, and metadata-only changes. Every step is re-admitted by the real grammar and bound to the preflighted table before anything executes; each step class carries its own budget, and a failed step surfaces as a typed *SequenceStepError naming the committed prefix. --- SAFETY.md | 2 +- docs/low-level-design.md | 9 +- pkg/executor/sequence.go | 332 ++++++++++++++++++++++ pkg/executor/sequence_integration_test.go | 213 ++++++++++++++ pkg/executor/sequence_internal_test.go | 145 ++++++++++ pkg/executor/sequence_test.go | 60 ++++ 6 files changed, 757 insertions(+), 4 deletions(-) create mode 100644 pkg/executor/sequence.go create mode 100644 pkg/executor/sequence_integration_test.go create mode 100644 pkg/executor/sequence_internal_test.go create mode 100644 pkg/executor/sequence_test.go diff --git a/SAFETY.md b/SAFETY.md index 4f44051..479c8af 100644 --- a/SAFETY.md +++ b/SAFETY.md @@ -19,7 +19,7 @@ The invariant registry (invariant IDs referenced below) lives in | --- | --- | --- | --- | | `pkg/dbconn` — pool defaults, terminate-blockers, retries, RDS TLS; advisory table lock planned | ✅ core | exists; advisory table lock planned | LK-2 primitives; LK-1 planned | | `pkg/preflight` — precondition verifier, refusals | ✅ core | exists (Phase 1: table-size guard); grows through Phase 2 | ST-6, RF-1..RF-5 | -| `pkg/executor` — bounded optimistic attempt; native concurrent index build with invalid-index recovery; remaining native idioms at Phase 3 | ✅ core | exists (Phase 1: attempt-under-budget; Phase 3.1: concurrent index build) | LK-2 (attempt bound + the CONCURRENTLY wait-policy exception) | +| `pkg/executor` — bounded optimistic attempt; native concurrent index build with invalid-index recovery; native sequence executor for the safer idioms | ✅ core | exists (Phase 1: attempt-under-budget; Phase 3.1: concurrent index build; Phase 3.2: sequence executor) | LK-2 (attempt bound + the CONCURRENTLY wait-policy exception) | | `pkg/checksum` — chunk verifier, continuous checker, repair | ✅ core | planned (Phase 5) | CO-1, CO-2, CO-3 | | `pkg/copier` — shadow-table chunked copy | ✅ core | planned (Phase 4) | CO-4, LK-3 | | `pkg/applier` — change apply, buffer, flush scheduling | ✅ core | planned (Phase 6) | CO-4, CO-5, CO-6, LK-3 | diff --git a/docs/low-level-design.md b/docs/low-level-design.md index ebf032d..8e6c777 100644 --- a/docs/low-level-design.md +++ b/docs/low-level-design.md @@ -340,8 +340,12 @@ Classification belongs to `pkg/planner`; `pkg/statement` supplies typed operatio | `migrate` (default) | Run the Phase 1 statement gate, preflight, and bounded optimistic native attempt. It does not yet execute classifier-produced safer SQL. | | `migrate --force` (planned Phase 3) | Run each statement **exactly as submitted**, bypassing the safe rewrite. Gated — see below. | -The classifier constructs `CREATE INDEX CONCURRENTLY` and other safer sequences today, but only -`diff` and `migrate --dry-run` render them. Phase 3 makes the classified route drive execution. +The classifier constructs `CREATE INDEX CONCURRENTLY` and other safer sequences today, and the +library can execute them — `pkg/executor`'s sequence executor runs a safer sequence under the +autocommit-each-step contract (brief steps bounded like an optimistic attempt, the validation +scan and concurrent builds under their own budgets). The CLI front door does not yet route to +it: `diff` and `migrate --dry-run` render the sequences, and Phase 3's substitution work wires +the classified route into execution. ### The `--force` gate @@ -800,7 +804,6 @@ roughly in order: each executor outcome gaining a stable string code in the report contracts, the same treatment `pkg/lint` gave its findings, so orchestrators branch on one vocabulary, - execute classifier-produced safer sequences through the routed native path, -- the remaining native idioms (`NOT VALID`+`VALIDATE`, `ADD PK USING INDEX`, fast-default), - bound lock acquisition with timeout and retry for the blocking idioms, - substitution by default, the guarded `--force` escape hatch, and progress reporting (`pg_stat_progress_create_index` by the build's backend PID, which the executor already diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go new file mode 100644 index 0000000..90db964 --- /dev/null +++ b/pkg/executor/sequence.go @@ -0,0 +1,332 @@ +// This file is the native sequence executor: it runs a planner-produced +// safer native sequence under the autocommit-each-step contract +// (planner.ExecutionAutocommit) — one statement at a time, in order, each +// in its own implicit or bounded transaction, never inside one enclosing +// block. It is what executes the Phase 3 idioms whose safety comes from +// their sequencing: NOT VALID plus an online VALIDATE, ADD PRIMARY KEY / +// UNIQUE USING INDEX over a concurrent build, the four-step SET NOT NULL, +// and the single-step fast-default and metadata-only changes. +// +// The executor never trusts the caller's classification (see SAFETY.md): +// every step is re-parsed by the real grammar and admitted by shape before +// anything executes, and every step runs bounded — brief catalog steps +// under the optimistic attempt's budgets, the constraint-validation scan +// under its own generous budget, and concurrent index builds under the +// dedicated CONCURRENTLY executor. A step that turns out to do rewrite +// work is cancelled cleanly by its budget, exactly like a blind optimistic +// attempt. +// +// A failed step ends the run immediately: the steps before it committed +// and their partial state remains, by design — each planner sequence +// constructor documents what a failed step leaves behind and how a retry +// resumes. The typed *SequenceStepError names the failed step so an +// operator or orchestrator can apply that contract. + +package executor + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/block/pg-sprite/pkg/preflight" + "github.com/block/pg-sprite/pkg/statement" +) + +// Typed admission refusals for the sequence executor. Admission covers the +// whole sequence before the first step executes, so a sequence that cannot +// be finished is never started. +var ( + // ErrEmptySequence is returned for a sequence with no steps: reporting + // success over nothing executed would be a false proof. + ErrEmptySequence = errors.New("sequence has no steps") + // ErrUnsupportedSequenceStep is returned when a step is not one of the + // shapes this executor can run safely: an ALTER TABLE step, or a + // CREATE INDEX CONCURRENTLY step. The concurrent forms of other + // statements are refused deliberately — a DROP INDEX CONCURRENTLY or + // REINDEX CONCURRENTLY is not driven yet, and a cancelled + // DETACH PARTITION CONCURRENTLY leaves a detach-pending partition + // state this executor does not own detecting or recovering. + ErrUnsupportedSequenceStep = errors.New("step is not a shape the sequence executor can run safely") +) + +// StepKind is the typed execution class a step was admitted under; +// automation branches on it, never on the step's SQL text. +type StepKind string + +// The execution classes a step can be admitted under. +const ( + // StepBrief: a bounded transactional run under the brief budgets — a + // catalog change that must prove itself effectively instant, exactly + // like an optimistic attempt. + StepBrief StepKind = "brief" + // StepConcurrentIndexBuild: a CREATE INDEX CONCURRENTLY delegated to + // the dedicated concurrent build executor, with its wait policy and + // invalid-index verdict. + StepConcurrentIndexBuild StepKind = "concurrent-index-build" + // StepValidateConstraint: an ALTER TABLE ... VALIDATE CONSTRAINT — a + // long online scan under SHARE UPDATE EXCLUSIVE, bounded by the + // validate budget rather than the brief one. + StepValidateConstraint StepKind = "validate-constraint" +) + +// ValidateBudget bounds one VALIDATE CONSTRAINT step. The validation scan +// is long by design — it is the online half of the NOT VALID pattern — so +// it gets its own overall bound instead of the brief statement budget, +// while lock acquisition stays tightly bounded: the SHARE UPDATE EXCLUSIVE +// it takes conflicts with other DDL, and queueing behind one must not +// stall the sequence for the whole scan budget. +type ValidateBudget struct { + // LockTimeout bounds how long the step may wait in the lock queue. + LockTimeout time.Duration + // Overall bounds the whole validation scan via statement_timeout; + // expect large tables to need a generous value. + Overall time.Duration +} + +// validate rejects budgets that would leave the validation unbounded. +func (b ValidateBudget) validate() error { + // INV: LK-2 — the validation is bounded by construction; below one + // millisecond a setting truncates to zero, which disables the + // corresponding PostgreSQL limit entirely. + if b.LockTimeout < minBudget { + return fmt.Errorf("validate lock budget must be at least %s, got %s", minBudget, b.LockTimeout) + } + if b.Overall < minBudget { + return fmt.Errorf("validate overall budget must be at least %s, got %s", minBudget, b.Overall) + } + if b.Overall > maxOverallBudget { + return fmt.Errorf("validate overall budget must be at most %s, got %s", maxOverallBudget, b.Overall) + } + return nil +} + +// SequenceBudget bounds one sequence run: each admitted step class carries +// its own budget, because the classes have opposite needs — a brief +// catalog step must be cancelled fast, a validation scan and a concurrent +// build must be allowed to run long. +type SequenceBudget struct { + // Brief bounds every brief catalog step. + Brief Budget + // Concurrent bounds every concurrent index build step. + Concurrent ConcurrentBudget + // Validate bounds every VALIDATE CONSTRAINT step. + Validate ValidateBudget +} + +// validate rejects budget sets with any unbounded member. All three are +// validated regardless of which step classes the sequence contains: a +// budget set is a unit, and admitting a partially-invalid one would make +// the same SequenceBudget pass or fail depending on the SQL next to it. +func (b SequenceBudget) validate() error { + if err := b.Brief.validate(); err != nil { + return err + } + if err := b.Concurrent.validate(); err != nil { + return err + } + return b.Validate.validate() +} + +// StepReport says what one committed step did, machine-readably. +type StepReport struct { + // SQL is the step's statement as submitted. + SQL string + // Kind is the execution class the step ran under. + Kind StepKind + // Duration is the wall-clock time of the step, session setup and + // verification included. + Duration time.Duration + // Index carries the concurrent build's verified report; nil for every + // other step kind. + Index *IndexBuildReport +} + +// SequenceReport is the record of a completed sequence run: one report per +// step, in execution order. It is returned only when every step committed. +type SequenceReport struct { + // Steps are the per-step reports, in execution order. + Steps []StepReport +} + +// SequenceStepError reports that a step failed and the run stopped there. +// The steps before it committed and their partial state remains — the +// planner's sequence constructors document what each failed step leaves +// behind and how a retry resumes. Err carries the step's own typed failure +// (*BudgetError, *InvalidIndexError, a server error) for errors.Is/As. +type SequenceStepError struct { + // Step is the failed step's 1-based position, matching the numbering + // the planner's partial-failure contracts use. + Step int + // Total is the number of steps the sequence was admitted with. + Total int + // Kind is the execution class the failed step ran under. + Kind StepKind + // SQL is the failed step's statement. + SQL string + // Err is the step's underlying failure. + Err error +} + +// Error implements the error interface. It names the failed step and the +// committed prefix, because "what already happened" is the first triage +// question a partial sequence raises. +func (e *SequenceStepError) Error() string { + return fmt.Sprintf("sequence step %d of %d (%s) failed; steps before it committed and their state remains: %v", + e.Step, e.Total, e.Kind, e.Err) +} + +// Unwrap exposes the step's failure to errors.Is/As. +func (e *SequenceStepError) Unwrap() error { return e.Err } + +// sequenceStep is one admitted step: its statement, re-parsed by the real +// grammar, and the execution class admission proved for it. +type sequenceStep struct { + st statement.Statement + kind StepKind +} + +// RunSequence runs steps in order against the preflighted table, each step +// in its own implicit or bounded transaction — the autocommit-each-step +// contract a planner-produced safer sequence carries. The pool must come +// from pkg/dbconn and, when the sequence contains a concurrent index build, +// must allow at least two connections (see BuildIndexConcurrently). The +// whole sequence is admitted before the first step executes: every step is +// re-parsed, its shape classified, and its target verified against the +// preflight proof, so a sequence this executor cannot finish is never +// started. On success every step committed and the report says what each +// did. On failure the run stops at the failing step and returns a typed +// *SequenceStepError; the committed prefix remains, per the planner's +// documented partial-failure contracts. +// +// Like the concurrent build — and unlike a blind optimistic attempt — no +// size-guard proof is required beyond the preflight itself: long scans on +// large tables are the sequence pattern's purpose, and every brief step is +// still individually bounded by the brief budgets. +func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.PreflightedTable, steps []string, b SequenceBudget) (SequenceReport, error) { + var rep SequenceReport + if err := b.validate(); err != nil { + return rep, err + } + admitted, err := admitSequence(pt.Schema(), pt.Table(), steps) + if err != nil { + return rep, err + } + for i, step := range admitted { + start := time.Now() + var indexReport *IndexBuildReport + switch step.kind { + case StepConcurrentIndexBuild: + r, buildErr := BuildIndexConcurrently(ctx, pool, step.st.SQL(), b.Concurrent) + err = buildErr + if buildErr == nil { + indexReport = &r + } + case StepValidateConstraint: + err = AttemptNative(ctx, pool, pt, step.st, Budget{ + LockTimeout: b.Validate.LockTimeout, + StatementTimeout: b.Validate.Overall, + }) + case StepBrief: + err = AttemptNative(ctx, pool, pt, step.st, b.Brief) + default: + // Admission produces only the three kinds above; an unknown + // kind here is a programming error and aborts fail-closed. + err = fmt.Errorf("%w: unhandled step kind %q", ErrInvariantViolation, step.kind) + } + if err != nil { + return rep, &SequenceStepError{Step: i + 1, Total: len(admitted), Kind: step.kind, SQL: step.st.SQL(), Err: err} + } + rep.Steps = append(rep.Steps, StepReport{ + SQL: step.st.SQL(), + Kind: step.kind, + Duration: time.Since(start), + Index: indexReport, + }) + } + return rep, nil +} + +// admitSequence re-parses and classifies every step and verifies each +// targets the preflighted table, before anything executes. A refusal names +// the offending step by 1-based position. +func admitSequence(schema, table string, steps []string) ([]sequenceStep, error) { + if len(steps) == 0 { + return nil, ErrEmptySequence + } + admitted := make([]sequenceStep, 0, len(steps)) + for i, sql := range steps { + step, err := admitStep(schema, table, sql) + if err != nil { + return nil, fmt.Errorf("sequence step %d of %d: %w", i+1, len(steps), err) + } + admitted = append(admitted, step) + } + return admitted, nil +} + +// admitStep classifies one step by its parsed shape, then verifies its +// target. Only two statement kinds are admissible: a CREATE INDEX +// CONCURRENTLY (delegated to the concurrent build executor) and an ALTER +// TABLE, split into the long-running VALIDATE CONSTRAINT class and the +// brief class everything else runs under. Anything else — including every +// other CONCURRENTLY form — is refused typed. Shape comes before the +// target check so an unsupported statement is reported as the shape +// refusal it is, not as a target mismatch. +func admitStep(schema, table, sql string) (sequenceStep, error) { + st, err := statement.ParseOne(sql) + if err != nil { + return sequenceStep{}, err + } + var step sequenceStep + switch st.Kind() { + case statement.KindCreateIndex: + if !st.Concurrent() { + // A blocking CREATE INDEX never belongs in a safer sequence; + // the planner emits only the concurrent form. + return sequenceStep{}, fmt.Errorf("blocking CREATE INDEX: %w", ErrUnsupportedSequenceStep) + } + step = sequenceStep{st: st, kind: StepConcurrentIndexBuild} + case statement.KindAlterTable: + if step, err = admitAlterTableStep(st, sql); err != nil { + return sequenceStep{}, err + } + default: + return sequenceStep{}, fmt.Errorf("statement kind %q: %w", st.Kind(), ErrUnsupportedSequenceStep) + } + // INV: ST-7 — every step runs only against the table the preflight + // proof verified; checked here for the whole sequence so a mixed-target + // sequence is refused before its first step executes, and re-checked by + // the brief runner per statement. + if st.Table() == "" || st.Schema() != schema || st.Table() != table { + return sequenceStep{}, fmt.Errorf("%w: ST-7: step targets %q but preflight verified %q", + ErrInvariantViolation, qualifiedName(st.Schema(), st.Table()), qualifiedName(schema, table)) + } + return step, nil +} + +// admitAlterTableStep splits ALTER TABLE steps into their execution +// classes. A VALIDATE CONSTRAINT — and only a lone one — gets the validate +// class: in a multi-operation ALTER the validation shares its transaction +// with the other subcommands, so the whole statement must prove itself +// under the brief budgets instead. Any CONCURRENTLY subcommand (DETACH +// PARTITION CONCURRENTLY) is refused: cancelling its wait leaves a +// detach-pending partition state this executor does not own recovering. +func admitAlterTableStep(st statement.Statement, sql string) (sequenceStep, error) { + ops, err := statement.ParseOps(sql) + if err != nil { + return sequenceStep{}, err + } + for _, op := range ops { + if op.Concurrent { + return sequenceStep{}, fmt.Errorf("%s: %w", op.Describe(), ErrUnsupportedSequenceStep) + } + } + if len(ops) == 1 && ops[0].Kind == statement.OpValidateConstraint { + return sequenceStep{st: st, kind: StepValidateConstraint}, nil + } + return sequenceStep{st: st, kind: StepBrief}, nil +} diff --git a/pkg/executor/sequence_integration_test.go b/pkg/executor/sequence_integration_test.go new file mode 100644 index 0000000..a456d92 --- /dev/null +++ b/pkg/executor/sequence_integration_test.go @@ -0,0 +1,213 @@ +package executor_test + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/executor" + "github.com/block/pg-sprite/pkg/planner" +) + +// sqlstateCheckViolation is the typed outcome a failed VALIDATE surfaces. +const sqlstateCheckViolation = "23514" + +// runBudget bounds integration sequences: brief steps must prove themselves +// fast, the long classes get room to finish on tiny test tables. +var runBudget = executor.SequenceBudget{ + Brief: executor.Budget{LockTimeout: 500 * time.Millisecond, StatementTimeout: 2 * time.Second}, + Concurrent: executor.ConcurrentBudget{Overall: time.Minute}, + Validate: executor.ValidateBudget{LockTimeout: 500 * time.Millisecond, Overall: time.Minute}, +} + +// constraintState reports whether the named constraint exists on the table +// and whether it is validated — the catalog oracle for the NOT VALID +// pattern. +func constraintState(t *testing.T, pool *pgxpool.Pool, schema, table, constraint string) (exists, validated bool) { + t.Helper() + err := pool.QueryRow(t.Context(), + `SELECT con.convalidated + FROM pg_constraint con + JOIN pg_class c ON c.oid = con.conrelid + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = $1 AND c.relname = $2 AND con.conname = $3`, + schema, table, constraint).Scan(&validated) + if errors.Is(err, pgx.ErrNoRows) { + return false, false + } + require.NoError(t, err) + return true, validated +} + +// saferSequence classifies sql through the planner and returns the safer +// sequence it constructed: the integration tests run exactly what the +// planner produces, proving the planner-to-executor seam end to end. +func saferSequence(t *testing.T, sql string) []string { + t.Helper() + plan, err := planner.Classify(sql, planner.Facts{}) + require.NoError(t, err) + require.Len(t, plan.Decisions, 1) + d := plan.Decisions[0] + require.Equal(t, planner.ReasonSaferIdiom, d.Reason, "the test premise is a safer-idiom decision") + require.NotEmpty(t, d.SaferSQL, "the planner must have constructed the sequence") + require.Equal(t, planner.ExecutionAutocommit, d.SaferSQLExecution) + return d.SaferSQL +} + +func TestRunSequenceValidatesCheckConstraintOnline(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int PRIMARY KEY, v int); INSERT INTO %s.t SELECT g, g FROM generate_series(1, 100) g", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0)", schema)) + rep, err := executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + require.NoError(t, err) + + require.Len(t, rep.Steps, 2) + assert.Equal(t, executor.StepBrief, rep.Steps[0].Kind, "the NOT VALID add is a brief catalog step") + assert.Equal(t, executor.StepValidateConstraint, rep.Steps[1].Kind, "the validation runs under its own class") + exists, validated := constraintState(t, pool, schema, "t", "v_positive") + assert.True(t, exists, "the constraint must exist") + assert.True(t, validated, "the constraint must be validated, not left NOT VALID") +} + +func TestRunSequenceSetNotNullLeavesNoScaffold(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int PRIMARY KEY, v int); INSERT INTO %s.t SELECT g, g FROM generate_series(1, 100) g", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN v SET NOT NULL", schema)) + rep, err := executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + require.NoError(t, err) + require.Len(t, rep.Steps, 4) + + var notNull bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT attnotnull FROM pg_attribute + WHERE attrelid = to_regclass($1) AND attname = 'v'`, + schema+".t").Scan(¬Null)) + assert.True(t, notNull, "the column must be NOT NULL") + var scaffolds int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) FROM pg_constraint WHERE conrelid = to_regclass($1) AND contype = 'c'`, + schema+".t").Scan(&scaffolds)) + assert.Zero(t, scaffolds, "the proving CHECK scaffold must be dropped") +} + +func TestRunSequenceAddsPrimaryKeyOverConcurrentBuild(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int NOT NULL, v int); INSERT INTO %s.t SELECT g, g FROM generate_series(1, 100) g", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD PRIMARY KEY (id)", schema)) + rep, err := executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + require.NoError(t, err) + + require.Len(t, rep.Steps, 2) + assert.Equal(t, executor.StepConcurrentIndexBuild, rep.Steps[0].Kind) + require.NotNil(t, rep.Steps[0].Index, "the build step must carry the verified index report") + assert.NotZero(t, rep.Steps[0].Index.IndexOID) + assert.Equal(t, executor.StepBrief, rep.Steps[1].Kind, "the USING INDEX attach is a brief catalog step") + + var pkIndex string + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT ci.relname + FROM pg_constraint con + JOIN pg_class ci ON ci.oid = con.conindid + WHERE con.conrelid = to_regclass($1) AND con.contype = 'p'`, + schema+".t").Scan(&pkIndex)) + assert.Equal(t, rep.Steps[0].Index.Index, pkIndex, "the primary key must own the concurrently built index") +} + +func TestRunSequenceRunsSingleStepChanges(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + // A fast-default ADD COLUMN and a metadata-only change are one-step + // sequences: the executor covers them without a dedicated path. + rep, err := executor.RunSequence(t.Context(), pool, pt, + []string{fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int NOT NULL DEFAULT 0", schema)}, runBudget) + require.NoError(t, err) + require.Len(t, rep.Steps, 1) + assert.Equal(t, executor.StepBrief, rep.Steps[0].Kind) + assert.Equal(t, "integer", columnType(t, pool, schema, "t", "age")) + + rep, err = executor.RunSequence(t.Context(), pool, pt, + []string{fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN age DROP DEFAULT", schema)}, runBudget) + require.NoError(t, err) + require.Len(t, rep.Steps, 1) +} + +func TestRunSequenceStopsAtFailingStepAndReportsPartialState(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int PRIMARY KEY, v int); INSERT INTO %s.t VALUES (1, -1)", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + // The violating row makes step 1 (the NOT VALID add) succeed and step 2 + // (the validation scan) fail: the documented partial state is the + // constraint left NOT VALID. + steps := saferSequence(t, fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0)", schema)) + _, err = executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + + var stepErr *executor.SequenceStepError + require.ErrorAs(t, err, &stepErr) + assert.Equal(t, 2, stepErr.Step, "the validation step must be the one reported failed") + assert.Equal(t, 2, stepErr.Total) + assert.Equal(t, executor.StepValidateConstraint, stepErr.Kind) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr, "the server failure must stay reachable through the step error") + assert.Equal(t, sqlstateCheckViolation, pgErr.Code) + + exists, validated := constraintState(t, pool, schema, "t", "v_positive") + assert.True(t, exists, "the committed step's constraint must remain, per the partial-failure contract") + assert.False(t, validated, "the failed validation must leave the constraint NOT VALID") +} + +func TestRunSequenceBudgetCancelsBlockedBriefStep(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v int)", schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + // A second session holds ACCESS EXCLUSIVE for the whole test, so the + // brief step can never be granted its lock and the lock budget fires. + blocker, err := pool.Begin(t.Context()) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context()))) + }) + _, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema)) + require.NoError(t, err) + + _, err = executor.RunSequence(t.Context(), pool, pt, + []string{fmt.Sprintf("ALTER TABLE %s.t DROP COLUMN v", schema)}, runBudget) + + var stepErr *executor.SequenceStepError + require.ErrorAs(t, err, &stepErr) + assert.Equal(t, 1, stepErr.Step) + var budgetErr *executor.BudgetError + require.ErrorAs(t, err, &budgetErr, "the budget outcome must stay reachable through the step error") + assert.Equal(t, executor.CauseLock, budgetErr.Cause) +} diff --git a/pkg/executor/sequence_internal_test.go b/pkg/executor/sequence_internal_test.go new file mode 100644 index 0000000..759380e --- /dev/null +++ b/pkg/executor/sequence_internal_test.go @@ -0,0 +1,145 @@ +// White-box tests for sequence admission: the shape classification and +// fail-closed refusals that decide which budget class each step runs under +// — decisions that must be provable without a database. + +package executor + +import ( + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestAdmitStepClassifiesShapes(t *testing.T) { + tests := []struct { + name string + sql string + want StepKind + }{ + { + name: "ADD CONSTRAINT NOT VALID is brief", + sql: `ALTER TABLE s.t ADD CONSTRAINT c CHECK (v > 0) NOT VALID`, + want: StepBrief, + }, + { + name: "a lone VALIDATE CONSTRAINT gets the validate class", + sql: `ALTER TABLE s.t VALIDATE CONSTRAINT c`, + want: StepValidateConstraint, + }, + { + name: "SET NOT NULL is brief", + sql: `ALTER TABLE s.t ALTER COLUMN v SET NOT NULL`, + want: StepBrief, + }, + { + name: "the scaffold DROP CONSTRAINT is brief", + sql: `ALTER TABLE s.t DROP CONSTRAINT c`, + want: StepBrief, + }, + { + name: "a fast-default ADD COLUMN is brief", + sql: `ALTER TABLE s.t ADD COLUMN age int NOT NULL DEFAULT 0`, + want: StepBrief, + }, + { + name: "a metadata-only change is brief", + sql: `ALTER TABLE s.t ALTER COLUMN v DROP DEFAULT`, + want: StepBrief, + }, + { + name: "ADD CONSTRAINT USING INDEX is brief", + sql: `ALTER TABLE s.t ADD CONSTRAINT t_pkey PRIMARY KEY USING INDEX t_pkey`, + want: StepBrief, + }, + { + name: "a concurrent index build is delegated", + sql: `CREATE UNIQUE INDEX CONCURRENTLY i ON s.t (v)`, + want: StepConcurrentIndexBuild, + }, + { + name: "a VALIDATE sharing a multi-operation ALTER runs brief", + sql: `ALTER TABLE s.t VALIDATE CONSTRAINT c, ADD COLUMN x int`, + want: StepBrief, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + step, err := admitStep("s", "t", tt.sql) + require.NoError(t, err) + assert.Equal(t, tt.want, step.kind) + }) + } +} + +func TestAdmitStepRefusals(t *testing.T) { + tests := []struct { + name string + sql string + wantErr error + }{ + { + name: "a blocking CREATE INDEX never belongs in a sequence", + sql: `CREATE INDEX i ON s.t (v)`, + wantErr: ErrUnsupportedSequenceStep, + }, + { + name: "DROP INDEX CONCURRENTLY is not driven", + sql: `DROP INDEX CONCURRENTLY s.i`, + wantErr: ErrUnsupportedSequenceStep, + }, + { + name: "REINDEX CONCURRENTLY is not driven", + sql: `REINDEX INDEX CONCURRENTLY s.i`, + wantErr: ErrUnsupportedSequenceStep, + }, + { + name: "a cancelled DETACH PARTITION CONCURRENTLY leaves detach-pending state", + sql: `ALTER TABLE s.t DETACH PARTITION p CONCURRENTLY`, + wantErr: ErrUnsupportedSequenceStep, + }, + { + name: "CREATE TABLE is not a sequence step", + sql: `CREATE TABLE s.t (id int)`, + wantErr: ErrUnsupportedSequenceStep, + }, + { + name: "a step against another table breaks the preflight binding", + sql: `ALTER TABLE s.other DROP CONSTRAINT c`, + wantErr: ErrInvariantViolation, + }, + { + name: "a step against another schema breaks the preflight binding", + sql: `ALTER TABLE elsewhere.t DROP CONSTRAINT c`, + wantErr: ErrInvariantViolation, + }, + { + name: "an unqualified index build against a qualified preflight breaks the binding", + sql: `CREATE UNIQUE INDEX CONCURRENTLY i ON t (v)`, + wantErr: ErrInvariantViolation, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := admitStep("s", "t", tt.sql) + require.ErrorIs(t, err, tt.wantErr) + }) + } +} + +func TestAdmitSequenceNamesTheOffendingStep(t *testing.T) { + steps := []string{ + `ALTER TABLE s.t ADD CONSTRAINT c CHECK (v > 0) NOT VALID`, + `CREATE TABLE s.t (id int)`, + } + _, err := admitSequence("s", "t", steps) + require.ErrorIs(t, err, ErrUnsupportedSequenceStep) + var stepErr *SequenceStepError + assert.False(t, errors.As(err, &stepErr), "an admission refusal is not a step failure: nothing executed") +} + +func TestAdmitSequenceSurfacesParseFailures(t *testing.T) { + _, err := admitSequence("s", "t", []string{`ALTER TABLE s.t THIS IS NOT SQL`}) + require.Error(t, err) +} diff --git a/pkg/executor/sequence_test.go b/pkg/executor/sequence_test.go new file mode 100644 index 0000000..847594f --- /dev/null +++ b/pkg/executor/sequence_test.go @@ -0,0 +1,60 @@ +package executor_test + +import ( + "errors" + "math" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/executor" + "github.com/block/pg-sprite/pkg/preflight" +) + +// sequenceBudget is a fully-bounded budget set for unit tests; admission +// refusals return before any database access. +var sequenceBudget = executor.SequenceBudget{ + Brief: executor.Budget{LockTimeout: 500 * time.Millisecond, StatementTimeout: 2 * time.Second}, + Concurrent: executor.ConcurrentBudget{Overall: time.Minute}, + Validate: executor.ValidateBudget{LockTimeout: 500 * time.Millisecond, Overall: time.Minute}, +} + +func TestRunSequenceRejectsUnboundedBudgets(t *testing.T) { + tests := []struct { + name string + mutate func(*executor.SequenceBudget) + }{ + {name: "zero brief lock budget", mutate: func(b *executor.SequenceBudget) { b.Brief.LockTimeout = 0 }}, + {name: "zero brief statement budget", mutate: func(b *executor.SequenceBudget) { b.Brief.StatementTimeout = 0 }}, + {name: "zero concurrent overall budget", mutate: func(b *executor.SequenceBudget) { b.Concurrent.Overall = 0 }}, + {name: "zero validate lock budget", mutate: func(b *executor.SequenceBudget) { b.Validate.LockTimeout = 0 }}, + {name: "zero validate overall budget", mutate: func(b *executor.SequenceBudget) { b.Validate.Overall = 0 }}, + {name: "negative validate overall budget", mutate: func(b *executor.SequenceBudget) { b.Validate.Overall = -time.Second }}, + { + name: "validate overall budget above the server ceiling", + mutate: func(b *executor.SequenceBudget) { + b.Validate.Overall = time.Duration(math.MaxInt32)*time.Millisecond + time.Millisecond + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b := sequenceBudget + tt.mutate(&b) + // A nil pool proves the refusal happens at admission, before + // any database access. + _, err := executor.RunSequence(t.Context(), nil, preflight.PreflightedTable{}, + []string{"ALTER TABLE s.t DROP CONSTRAINT c"}, b) + require.Error(t, err) + var stepErr *executor.SequenceStepError + assert.False(t, errors.As(err, &stepErr), "a budget refusal must precede any step execution") + }) + } +} + +func TestRunSequenceRefusesEmptySequence(t *testing.T) { + _, err := executor.RunSequence(t.Context(), nil, preflight.PreflightedTable{}, nil, sequenceBudget) + require.ErrorIs(t, err, executor.ErrEmptySequence) +} From 4acad1110bd7d8994cb28a1e397f7ab009d04474 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 12 Aug 2026 14:28:37 +1000 Subject: [PATCH 2/2] harden sequence admission, budgets, and validate-class verdicts Refusals decidable at admission must never fire mid-run after a committed prefix: the delegated build executor's static checks and pool guard now run before the first step, and over-ceiling budget values are refused as a unit. An external cancel of a long validation is corroborated by elapsed time so it never reads as budget exhaustion. Registry and design docs amended to match. --- docs/high-level-design.md | 7 +- docs/invariants.md | 17 +- docs/low-level-design.md | 13 +- pkg/executor/optimistic.go | 10 + pkg/executor/sequence.go | 57 ++++++ pkg/executor/sequence_integration_test.go | 211 +++++++++++++++++++++- pkg/executor/sequence_internal_test.go | 24 ++- pkg/executor/sequence_test.go | 18 ++ 8 files changed, 336 insertions(+), 21 deletions(-) diff --git a/docs/high-level-design.md b/docs/high-level-design.md index d3f2ba6..f17d680 100644 --- a/docs/high-level-design.md +++ b/docs/high-level-design.md @@ -239,9 +239,10 @@ Two principles govern this: an **explicit confirmation** (typed acknowledgement, not a bare `-y`), and the override is logged. Force is an escape hatch, not a convenience. -Today the classifier constructs safer sequences and `diff` / `migrate --dry-run` render them; -default `migrate` still uses the bounded optimistic Phase 1 path. Phase 3 adds substitution and -execution of classifier-produced SQL. +Today the classifier constructs safer sequences, `diff` / `migrate --dry-run` render them, and +the library's sequence executor runs them under the autocommit-each-step contract; default +`migrate` still uses the bounded optimistic Phase 1 path. Phase 3's remaining work is the +substitution wiring that routes the classified sequences into `migrate`. In non-interactive contexts (CI), advisory mode is a natural gate: the engine prints the recommended rewrites and exits non-zero if a submitted statement would need a riskier path than diff --git a/docs/invariants.md b/docs/invariants.md index ffa9591..2352e90 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -139,10 +139,15 @@ The cutover swap is the only `ACCESS EXCLUSIVE` acquisition in the happy path, a strong-lock acquisition (swap, catalog flips, trigger install in fallback mode) runs under `lock_timeout` + bounded retry/backoff so the engine never sits at the head of the lock queue (mysql-vs-postgresql § the lock queue). -**Exception policy required:** `CREATE INDEX CONCURRENTLY` (and `REINDEX CONCURRENTLY`, -`VALIDATE CONSTRAINT`) wait on other transactions via lock waits that a naive `lock_timeout` -cancels — leaving an `INVALID` index. These statements get their own wait policy rather than the -blanket timeout. *Enforced:* every DDL execution path in the native and copy-and-swap executors. +**Exception policy required:** `CREATE INDEX CONCURRENTLY` and `REINDEX CONCURRENTLY` wait on +other transactions via lock waits that a naive `lock_timeout` cancels — leaving an `INVALID` +index — so they get their own wait policy (no per-lock timeout, one overall statement deadline) +rather than the blanket timeout. `VALIDATE CONSTRAINT` is different in kind: its cancellation is +transactionally clean (the constraint simply stays `NOT VALID`; no debris), so the sequence +executor's validate class deliberately keeps a bounded per-lock timeout — queueing behind a +conflicting lock holder must not stall a sequence for the whole scan budget — while the scan +itself runs under its own generous overall budget. *Enforced:* every DDL execution path in the +native and copy-and-swap executors. *Source:* [design-principles](design-principles.md#correctness-and-safety), mysql-vs-postgresql; CIC exception from the validation review. @@ -238,7 +243,9 @@ which enforces exactly one statement through the real grammar — and refuses, b executes, any statement whose target table does not match the preflight proof it was handed. A proof for one table can never smuggle SQL against another, and a multi-statement string can never reach the database through the executor (pgx's simple protocol would happily run all of -it). *Enforced:* `pkg/executor` (`AttemptNative`), `pkg/statement` (proof construction). +it). *Enforced:* `pkg/executor` (`AttemptNative`; `RunSequence` admission re-proves every step's +target against the preflight proof before the first step executes), `pkg/statement` (proof +construction). *Source:* adversarial review of the optimistic front door. ## Refusals and preflight (RF) diff --git a/docs/low-level-design.md b/docs/low-level-design.md index 8e6c777..6a963ff 100644 --- a/docs/low-level-design.md +++ b/docs/low-level-design.md @@ -341,11 +341,14 @@ Classification belongs to `pkg/planner`; `pkg/statement` supplies typed operatio | `migrate --force` (planned Phase 3) | Run each statement **exactly as submitted**, bypassing the safe rewrite. Gated — see below. | The classifier constructs `CREATE INDEX CONCURRENTLY` and other safer sequences today, and the -library can execute them — `pkg/executor`'s sequence executor runs a safer sequence under the -autocommit-each-step contract (brief steps bounded like an optimistic attempt, the validation -scan and concurrent builds under their own budgets). The CLI front door does not yet route to -it: `diff` and `migrate --dry-run` render the sequences, and Phase 3's substitution work wires -the classified route into execution. +library executes the multi-step idiom families — `pkg/executor`'s sequence executor runs a safer +sequence under the autocommit-each-step contract (brief steps bounded like an optimistic +attempt, the validation scan and concurrent builds under their own budgets). The one-step +`CONCURRENTLY` rewrites the classifier also emits (`DROP INDEX`, `REINDEX`, +`DETACH PARTITION`) are not yet driven: the sequence executor refuses them typed, because a +cancelled wait leaves recovery states it does not yet own. The CLI front door does not yet +route to it: `diff` and `migrate --dry-run` render the sequences, and Phase 3's substitution +work wires the classified route into execution. ### The `--force` gate diff --git a/pkg/executor/optimistic.go b/pkg/executor/optimistic.go index 2096a27..3b009ff 100644 --- a/pkg/executor/optimistic.go +++ b/pkg/executor/optimistic.go @@ -108,6 +108,16 @@ func (b Budget) validate() error { if b.StatementTimeout < minBudget { return fmt.Errorf("statement budget must be at least %s, got %s", minBudget, b.StatementTimeout) } + // Both settings are int32-millisecond server GUCs sent raw via + // SET LOCAL: a value beyond the server ceiling would be rejected + // mid-attempt as an out-of-range setting — an operational error where + // a budget defect decidable here should refuse at admission. + if b.LockTimeout > maxOverallBudget { + return fmt.Errorf("lock budget must be at most %s, got %s", maxOverallBudget, b.LockTimeout) + } + if b.StatementTimeout > maxOverallBudget { + return fmt.Errorf("statement budget must be at most %s, got %s", maxOverallBudget, b.StatementTimeout) + } return nil } diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go index 90db964..4f3350b 100644 --- a/pkg/executor/sequence.go +++ b/pkg/executor/sequence.go @@ -95,6 +95,9 @@ func (b ValidateBudget) validate() error { if b.LockTimeout < minBudget { return fmt.Errorf("validate lock budget must be at least %s, got %s", minBudget, b.LockTimeout) } + if b.LockTimeout > maxOverallBudget { + return fmt.Errorf("validate lock budget must be at most %s, got %s", maxOverallBudget, b.LockTimeout) + } if b.Overall < minBudget { return fmt.Errorf("validate overall budget must be at least %s, got %s", minBudget, b.Overall) } @@ -215,6 +218,14 @@ func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight if err != nil { return rep, err } + // INV: LK-2 — the concurrent executor's pool guard is re-proven for + // the whole sequence before the first step executes: a too-small pool + // is decidable now, and letting BuildIndexConcurrently discover it + // mid-run would leave a committed prefix behind a refusal this + // executor could have made up front. + if sequenceHasConcurrentBuild(admitted) && pool.Config().MaxConns < 2 { + return rep, ErrPoolTooSmall + } for i, step := range admitted { start := time.Now() var indexReport *IndexBuildReport @@ -230,6 +241,7 @@ func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight LockTimeout: b.Validate.LockTimeout, StatementTimeout: b.Validate.Overall, }) + err = corroborateValidateCancel(err, b.Validate, time.Since(start)) case StepBrief: err = AttemptNative(ctx, pool, pt, step.st, b.Brief) default: @@ -250,6 +262,43 @@ func RunSequence(ctx context.Context, pool *pgxpool.Pool, pt preflight.Preflight return rep, nil } +// sequenceHasConcurrentBuild reports whether any admitted step is a +// concurrent index build — the class whose executor needs the two-connection +// pool guarantee. +func sequenceHasConcurrentBuild(steps []sequenceStep) bool { + for _, s := range steps { + if s.kind == StepConcurrentIndexBuild { + return true + } + } + return false +} + +// corroborateValidateCancel disambiguates a statement-cancellation verdict +// on a validate step. SQLSTATE 57014 is query_canceled generally — an +// operator's pg_cancel_backend raises the same code as statement_timeout — +// and the brief mapping reads it as statement-budget exhaustion. That +// conflation is tolerable inside a seconds-scale brief budget, but wrong +// across the validate class's generous budget: a deliberate cancel hours +// early would read as exhaustion, and exhaustion invites escalation to a +// heavier strategy when the cancel means the change should be left alone. +// As in the concurrent build executor, elapsed time corroborates: the +// executor's own statement_timeout cannot fire before the budget elapses, +// so an earlier cancellation came from outside. The original verdict is +// folded into the message, not the chain — the whole point is that this +// failure is not a *BudgetError. +func corroborateValidateCancel(err error, b ValidateBudget, elapsed time.Duration) error { + var budgetErr *BudgetError + if !errors.As(err, &budgetErr) || budgetErr.Cause != CauseStatement { + return err + } + if elapsed >= b.Overall { + return err + } + return fmt.Errorf("%w (after %s of a %s budget): %s", + ErrCancelledExternally, elapsed.Round(time.Millisecond), b.Overall, err.Error()) +} + // admitSequence re-parses and classifies every step and verifies each // targets the preflighted table, before anything executes. A refusal names // the offending step by 1-based position. @@ -289,6 +338,14 @@ func admitStep(schema, table, sql string) (sequenceStep, error) { // the planner emits only the concurrent form. return sequenceStep{}, fmt.Errorf("blocking CREATE INDEX: %w", ErrUnsupportedSequenceStep) } + // The delegated executor's statically-decidable admission + // requirements — a named index, no IF NOT EXISTS, a + // schema-qualified table — are proven here too: a refusal + // decidable before anything executes must never fire mid-run + // after earlier steps committed. + if _, err := admitConcurrentIndexBuild(sql); err != nil { + return sequenceStep{}, err + } step = sequenceStep{st: st, kind: StepConcurrentIndexBuild} case statement.KindAlterTable: if step, err = admitAlterTableStep(st, sql); err != nil { diff --git a/pkg/executor/sequence_integration_test.go b/pkg/executor/sequence_integration_test.go index a456d92..f90824d 100644 --- a/pkg/executor/sequence_integration_test.go +++ b/pkg/executor/sequence_integration_test.go @@ -13,6 +13,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/block/pg-sprite/internal/testutil" + "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/executor" "github.com/block/pg-sprite/pkg/planner" ) @@ -22,10 +24,14 @@ const sqlstateCheckViolation = "23514" // runBudget bounds integration sequences: brief steps must prove themselves // fast, the long classes get room to finish on tiny test tables. +// Validate.LockTimeout is deliberately distinct from Brief.LockTimeout: the +// budget-class tests prove which class a step ran under by the budget value +// its cancellation reports, so equal values would make a misclassification +// invisible. var runBudget = executor.SequenceBudget{ Brief: executor.Budget{LockTimeout: 500 * time.Millisecond, StatementTimeout: 2 * time.Second}, Concurrent: executor.ConcurrentBudget{Overall: time.Minute}, - Validate: executor.ValidateBudget{LockTimeout: 500 * time.Millisecond, Overall: time.Minute}, + Validate: executor.ValidateBudget{LockTimeout: time.Second, Overall: time.Minute}, } // constraintState reports whether the named constraint exists on the table @@ -183,6 +189,18 @@ func TestRunSequenceStopsAtFailingStepAndReportsPartialState(t *testing.T) { exists, validated := constraintState(t, pool, schema, "t", "v_positive") assert.True(t, exists, "the committed step's constraint must remain, per the partial-failure contract") assert.False(t, validated, "the failed validation must leave the constraint NOT VALID") + + // TM-1: the documented retry contract must actually work — fix the + // violating data and resume from the failed step, using nothing but + // the typed error's own step number. + _, err = pool.Exec(t.Context(), fmt.Sprintf("UPDATE %s.t SET v = 1 WHERE v <= 0", schema)) + require.NoError(t, err) + rep, err := executor.RunSequence(t.Context(), pool, pt, steps[stepErr.Step-1:], runBudget) + require.NoError(t, err, "resuming from the failed step must complete the sequence") + require.Len(t, rep.Steps, 1) + assert.Equal(t, executor.StepValidateConstraint, rep.Steps[0].Kind) + _, validated = constraintState(t, pool, schema, "t", "v_positive") + assert.True(t, validated, "the resumed validation must leave the constraint validated") } func TestRunSequenceBudgetCancelsBlockedBriefStep(t *testing.T) { @@ -191,18 +209,21 @@ func TestRunSequenceBudgetCancelsBlockedBriefStep(t *testing.T) { require.NoError(t, err) pt := mustPreflight(t, pool, schema, "t") - // A second session holds ACCESS EXCLUSIVE for the whole test, so the - // brief step can never be granted its lock and the lock budget fires. + // A second session holds ACCESS EXCLUSIVE while the sequence runs, so + // the brief step can never be granted its lock and the lock budget + // fires. The cleanup rollback is a redundant safety closer: the test + // body rolls the blocker back itself, and the guaranteed ErrTxClosed + // from the cleanup is discarded. blocker, err := pool.Begin(t.Context()) require.NoError(t, err) t.Cleanup(func() { - require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context()))) + _ = blocker.Rollback(context.WithoutCancel(t.Context())) }) _, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema)) require.NoError(t, err) - _, err = executor.RunSequence(t.Context(), pool, pt, - []string{fmt.Sprintf("ALTER TABLE %s.t DROP COLUMN v", schema)}, runBudget) + steps := []string{fmt.Sprintf("ALTER TABLE %s.t DROP COLUMN v", schema)} + _, err = executor.RunSequence(t.Context(), pool, pt, steps, runBudget) var stepErr *executor.SequenceStepError require.ErrorAs(t, err, &stepErr) @@ -210,4 +231,182 @@ func TestRunSequenceBudgetCancelsBlockedBriefStep(t *testing.T) { var budgetErr *executor.BudgetError require.ErrorAs(t, err, &budgetErr, "the budget outcome must stay reachable through the step error") assert.Equal(t, executor.CauseLock, budgetErr.Cause) + assert.Equal(t, runBudget.Brief.LockTimeout, budgetErr.Budget, "a brief step must be cancelled by the brief lock budget") + + // TM-3: the cancelled step must have left durable state untouched — + // the budget cancellation rolls back, it never half-commits. + assert.True(t, columnExists(t, pool, schema, "t", "v"), + "the lock-cancelled DROP COLUMN must not have committed") + + // TM-3: after the fault clears, the same sequence must proceed to + // completion. + require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context()))) + _, err = executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + require.NoError(t, err, "the sequence must complete once the lock holder is gone") + assert.False(t, columnExists(t, pool, schema, "t", "v")) +} + +// columnExists reports whether the named live column exists on the table — +// the durable-state oracle for cancelled brief steps. +func columnExists(t *testing.T, pool *pgxpool.Pool, schema, table, column string) bool { + t.Helper() + var exists bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT EXISTS ( + SELECT FROM pg_attribute + WHERE attrelid = to_regclass($1) AND attname = $2 + AND attnum > 0 AND NOT attisdropped)`, + schema+"."+table, column).Scan(&exists)) + return exists +} + +// TestRunSequenceValidateRunsUnderValidateBudget proves the headline +// behavior of the validate class: a lone VALIDATE CONSTRAINT runs under +// ValidateBudget, not the brief budgets. A second-session ACCESS EXCLUSIVE +// holder parks the validate in the lock queue, and the budget value its +// cancellation reports — distinct from Brief.LockTimeout by fixture +// construction — proves which class it ran under. +func TestRunSequenceValidateRunsUnderValidateBudget(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int PRIMARY KEY, v int); ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0) NOT VALID", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + blocker, err := pool.Begin(t.Context()) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context()))) + }) + _, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema)) + require.NoError(t, err) + + _, err = executor.RunSequence(t.Context(), pool, pt, + []string{fmt.Sprintf("ALTER TABLE %s.t VALIDATE CONSTRAINT v_positive", schema)}, runBudget) + + var stepErr *executor.SequenceStepError + require.ErrorAs(t, err, &stepErr) + assert.Equal(t, executor.StepValidateConstraint, stepErr.Kind) + var budgetErr *executor.BudgetError + require.ErrorAs(t, err, &budgetErr) + assert.Equal(t, executor.CauseLock, budgetErr.Cause) + assert.Equal(t, runBudget.Validate.LockTimeout, budgetErr.Budget, + "the validate step must be cancelled by the validate lock budget, not the brief one") +} + +// TestRunSequenceOperatorCancelOfValidateIsNotBudgetExhaustion covers the +// 57014 disambiguation for the validate class: an operator's +// pg_cancel_backend early in a generous validate budget must surface as +// ErrCancelledExternally, never as a *BudgetError — a consumer reading +// budget exhaustion would escalate to a heavier strategy when a human +// deliberately stopped the validation. +func TestRunSequenceOperatorCancelOfValidateIsNotBudgetExhaustion(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int PRIMARY KEY, v int); ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0) NOT VALID", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + // An ACCESS EXCLUSIVE holder parks the validate in the lock queue, + // giving the cancel a window; the generous budgets guarantee neither + // timeout can fire first. + blocker, err := pool.Begin(t.Context()) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, blocker.Rollback(context.WithoutCancel(t.Context()))) + }) + _, err = blocker.Exec(t.Context(), fmt.Sprintf("LOCK TABLE %s.t IN ACCESS EXCLUSIVE MODE", schema)) + require.NoError(t, err) + + b := runBudget + b.Validate = executor.ValidateBudget{LockTimeout: time.Minute, Overall: time.Minute} + done := make(chan error, 1) + go func() { + _, err := executor.RunSequence(t.Context(), pool, pt, + []string{fmt.Sprintf("ALTER TABLE %s.t VALIDATE CONSTRAINT v_positive", schema)}, b) + done <- err + }() + + // Cancel the validate's backend once it is provably executing; the + // ALTER TABLE prefix cannot match this polling query itself. + require.Eventually(t, func() bool { + var cancelled bool + err := pool.QueryRow(t.Context(), + `SELECT pg_cancel_backend(pid) FROM pg_stat_activity + WHERE query LIKE 'ALTER TABLE %' AND query LIKE '%VALIDATE CONSTRAINT%' AND state = 'active'`).Scan(&cancelled) + return err == nil && cancelled + }, 30*time.Second, 50*time.Millisecond, "the validate's backend must be found and cancelled") + + select { + case err := <-done: + require.ErrorIs(t, err, executor.ErrCancelledExternally) + var stepErr *executor.SequenceStepError + require.ErrorAs(t, err, &stepErr, "the cancel must still be attributed to its step") + assert.Equal(t, executor.StepValidateConstraint, stepErr.Kind) + var budgetErr *executor.BudgetError + assert.False(t, errors.As(err, &budgetErr), "an operator cancel must not read as budget exhaustion") + case <-time.After(time.Minute): + t.Fatal("the cancelled sequence did not return") + } +} + +// TestRunSequenceSurfacesFailedConcurrentBuildStep makes a delegated build +// step fail inside a sequence: the *InvalidIndexError verdict must stay +// reachable through the step error, and the committed prefix must remain +// per the partial-failure contract. +func TestRunSequenceSurfacesFailedConcurrentBuildStep(t *testing.T) { + pool, schema := newPool(t) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.t (id int PRIMARY KEY, v int); INSERT INTO %s.t VALUES (1, 7), (2, 7)", + schema, schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + // The duplicate rows make the unique build fail after the brief step + // committed. + steps := []string{ + fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0) NOT VALID", schema), + fmt.Sprintf("CREATE UNIQUE INDEX CONCURRENTLY i_v ON %s.t (v)", schema), + } + _, err = executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + + var stepErr *executor.SequenceStepError + require.ErrorAs(t, err, &stepErr) + assert.Equal(t, 2, stepErr.Step) + assert.Equal(t, executor.StepConcurrentIndexBuild, stepErr.Kind) + var invalidErr *executor.InvalidIndexError + require.ErrorAs(t, err, &invalidErr, "the build's typed verdict must stay reachable through the step error") + exists, _ := constraintState(t, pool, schema, "t", "v_positive") + assert.True(t, exists, "the committed prefix must remain, per the partial-failure contract") +} + +// TestRunSequenceRefusesSingleConnectionPoolBeforeAnyStep covers the +// admission-time pool guard for sequences containing a concurrent build: +// a pool refusal decidable up front must precede the first step, never +// fire mid-run after a committed prefix. +func TestRunSequenceRefusesSingleConnectionPoolBeforeAnyStep(t *testing.T) { + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: testutil.StartPostgres(t), MaxConns: 1}) + require.NoError(t, err) + t.Cleanup(pool.Close) + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v int)", schema)) + require.NoError(t, err) + pt := mustPreflight(t, pool, schema, "t") + + // The brief step comes before the build: if the pool guard fired only + // inside the delegated executor, the constraint would already be + // committed. + steps := []string{ + fmt.Sprintf("ALTER TABLE %s.t ADD CONSTRAINT v_positive CHECK (v > 0) NOT VALID", schema), + fmt.Sprintf("CREATE UNIQUE INDEX CONCURRENTLY i_v ON %s.t (v)", schema), + } + _, err = executor.RunSequence(t.Context(), pool, pt, steps, runBudget) + + require.ErrorIs(t, err, executor.ErrPoolTooSmall) + var stepErr *executor.SequenceStepError + assert.False(t, errors.As(err, &stepErr), "the pool refusal must precede any step execution") + exists, _ := constraintState(t, pool, schema, "t", "v_positive") + assert.False(t, exists, "nothing may execute when the sequence cannot finish") } diff --git a/pkg/executor/sequence_internal_test.go b/pkg/executor/sequence_internal_test.go index 759380e..8f33186 100644 --- a/pkg/executor/sequence_internal_test.go +++ b/pkg/executor/sequence_internal_test.go @@ -84,6 +84,16 @@ func TestAdmitStepRefusals(t *testing.T) { sql: `CREATE INDEX i ON s.t (v)`, wantErr: ErrUnsupportedSequenceStep, }, + { + name: "an unnamed concurrent build is refused at admission, not mid-run", + sql: `CREATE UNIQUE INDEX CONCURRENTLY ON s.t (v)`, + wantErr: ErrUnnamedIndex, + }, + { + name: "IF NOT EXISTS on a concurrent build is refused at admission, not mid-run", + sql: `CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS i ON s.t (v)`, + wantErr: ErrIfNotExistsUnsupported, + }, { name: "DROP INDEX CONCURRENTLY is not driven", sql: `DROP INDEX CONCURRENTLY s.i`, @@ -115,9 +125,9 @@ func TestAdmitStepRefusals(t *testing.T) { wantErr: ErrInvariantViolation, }, { - name: "an unqualified index build against a qualified preflight breaks the binding", + name: "an unqualified index build is refused by the build admission before the target check", sql: `CREATE UNIQUE INDEX CONCURRENTLY i ON t (v)`, - wantErr: ErrInvariantViolation, + wantErr: ErrUnqualifiedTable, }, } for _, tt := range tests { @@ -128,6 +138,16 @@ func TestAdmitStepRefusals(t *testing.T) { } } +// TestAdmitStepRefusesUnqualifiedBuildUnderUnqualifiedPreflight covers the +// one build-qualification gap the ST-7 target check cannot see: with an +// unqualified preflight, an unqualified build's schemas match ("" == ""), +// so only the delegated executor's own admission refuses it — and that +// refusal must fire here, before anything executes. +func TestAdmitStepRefusesUnqualifiedBuildUnderUnqualifiedPreflight(t *testing.T) { + _, err := admitStep("", "t", `CREATE UNIQUE INDEX CONCURRENTLY i ON t (v)`) + require.ErrorIs(t, err, ErrUnqualifiedTable) +} + func TestAdmitSequenceNamesTheOffendingStep(t *testing.T) { steps := []string{ `ALTER TABLE s.t ADD CONSTRAINT c CHECK (v > 0) NOT VALID`, diff --git a/pkg/executor/sequence_test.go b/pkg/executor/sequence_test.go index 847594f..19b4c88 100644 --- a/pkg/executor/sequence_test.go +++ b/pkg/executor/sequence_test.go @@ -38,6 +38,24 @@ func TestRunSequenceRejectsUnboundedBudgets(t *testing.T) { b.Validate.Overall = time.Duration(math.MaxInt32)*time.Millisecond + time.Millisecond }, }, + { + name: "validate lock budget above the server ceiling", + mutate: func(b *executor.SequenceBudget) { + b.Validate.LockTimeout = time.Duration(math.MaxInt32)*time.Millisecond + time.Millisecond + }, + }, + { + name: "brief statement budget above the server ceiling", + mutate: func(b *executor.SequenceBudget) { + b.Brief.StatementTimeout = time.Duration(math.MaxInt32)*time.Millisecond + time.Millisecond + }, + }, + { + name: "brief lock budget above the server ceiling", + mutate: func(b *executor.SequenceBudget) { + b.Brief.LockTimeout = time.Duration(math.MaxInt32)*time.Millisecond + time.Millisecond + }, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) {