From fa357bdd532853c95db0f06b409c16223078fa9e Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 12 Aug 2026 14:27:25 +1000 Subject: [PATCH 1/3] feat(cli): substitute safer sequences by default, gated --force migrate now executes the planner's classified route: a blocking submitted form runs as the safer native sequence (reported in the verdict's executed_sql), --force requires typing the resolved schema-qualified table and is audit-logged, and every typed executor outcome carries a stable string code (executor.OutcomeCode) so orchestrators branch on one vocabulary, not prose. --- README.md | 8 +- docs/architecture.md | 2 +- docs/high-level-design.md | 41 +-- docs/low-level-design.md | 76 ++--- docs/postgres-online-ddl-reference.md | 4 +- internal/cli/cli.go | 31 +- internal/cli/migrate.go | 292 +++++++++++++++-- internal/cli/migrate_integration_test.go | 391 ++++++++++++++++++++++- pkg/executor/code.go | 158 +++++++++ pkg/executor/code_test.go | 145 +++++++++ pkg/executor/native.go | 13 +- pkg/executor/sequence.go | 12 +- pkg/preflight/preflight.go | 7 + pkg/statement/desired.go | 8 +- pkg/statement/desired_test.go | 6 +- pkg/verdict/verdict.go | 28 ++ 16 files changed, 1098 insertions(+), 124 deletions(-) create mode 100644 pkg/executor/code.go create mode 100644 pkg/executor/code_test.go diff --git a/README.md b/README.md index 0847bcb..246f195 100644 --- a/README.md +++ b/README.md @@ -27,9 +27,11 @@ verdict when it can't prove one (see **Status: Phases 1 and 2.1–2.5.** The parse boundary, declarative diff, classifier, router seam, versioned dry-run plan report, offline linter, and -advisory `suggest` command are implemented. `pg-sprite migrate --alter '…'` runs a bounded optimistic -native attempt; routed execution beyond that attempt lands in Phase 3. -Changes without an available backend get a structured refusal (exit code 2). +advisory `suggest` command are implemented. `pg-sprite migrate --alter '…'` classifies and +routes the statement, then executes the routed SQL — the planner's safer native sequence by +default when the submitted form blocks (reported in the verdict's `executed_sql`), a bounded +optimistic native attempt otherwise. A gated `--force` runs the submitted form as-is under the +same budgets. Changes without an available backend get a structured refusal (exit code 2). The design docs and the phased build plan live in [docs/](docs/) — start with [docs/README.md](docs/README.md); the vision — what pg-sprite is and is not — diff --git a/docs/architecture.md b/docs/architecture.md index 768fea1..6149a2f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -135,7 +135,7 @@ different levels of commitment: | `pkg/plan` | Versioned machine-readable dry-run plan report — the one JSON contract both front doors emit and an orchestrator consumes | exists (Phase 2.5) | | `pkg/diffplan` | The declarative front door as a library: desired schema in, routed `plan.Report` out — the CLI `diff` and embedding orchestrators share this one pipeline | exists | | `pkg/router` | Route classified statements to native / copy-and-swap / refuse dispositions; copy-and-swap reports unavailable until that backend lands | exists (Phase 2.4) | -| `pkg/executor` | Bounded optimistic native attempt; the `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) lands in Phase 3 | bounded optimistic attempt exists | +| `pkg/executor` | Bounded optimistic native attempt, the concurrent index build, and the autocommit safer-sequence runner, with stable outcome codes; the full `Executor` contract (`Plan`/`Execute`/`Status`/`Abort`) arrives with the copy-and-swap backend | native execution exists | | `pkg/table` | PK-range chunkers (single-column fast path, composite), dynamic time-based sizing | Phase 4 | | `pkg/copier` | Parallel chunked copy into the shadow table (never overwrites) | Phase 4 | | `pkg/checksum` | The mandatory correctness gate; continuous checker; repair primitive | Phase 5 | diff --git a/docs/high-level-design.md b/docs/high-level-design.md index f17d680..27f0232 100644 --- a/docs/high-level-design.md +++ b/docs/high-level-design.md @@ -102,8 +102,9 @@ the planner's classifier: almost no parsing logic. This path lives at the `migrate` front door. - **Classified planning path (Phases 2.1–2.4).** Parse the statement and introspect the live schema to **predict the path up front** — native-safe, needs-rewrite, or refuse — without trial - execution. The planner drives `diff` and `migrate --dry-run`; Phase 3 will make its classified - route drive execution and remove the wasted/aborted attempts that the optimistic path can incur. + execution. The planner drives `diff` and `migrate --dry-run`, and its classified route drives + `migrate`'s execution: a blocking submitted form is substituted with the planner's safer + native sequence instead of incurring a wasted/aborted blind attempt. > **PostgreSQL caveat.** Unlike MySQL's `ALGORITHM=INSTANT`, PostgreSQL has **no assertion** that > forces a change to be instant-or-error — a rewrite attempt acquires `ACCESS EXCLUSIVE` and does @@ -171,7 +172,7 @@ The engine accepts a change two ways, both feeding the **same** planner pipeline - **Declarative** — the user supplies the **desired end-state** (a checked-in `CREATE TABLE` `.sql` file); the engine **derives** the `ALTER` by diffing desired vs live, then runs it - through classify → route. Phase 3 adds execution of the classified route. + through classify → route — the same classified route `migrate` executes. - **Imperative** — the user supplies the `ALTER` directly. It is the **same** pipeline with the diff step skipped. @@ -200,16 +201,18 @@ literal statement: │ yes ▼ ┌──────────────────────────────────────────────────────────-┐ - │ RECOMMENDATION (does NOT execute): │ + │ RECOMMENDATION (dry-run does NOT execute): │ │ you asked: CREATE INDEX idx ON orders (customer_id) │ │ safer form: CREATE INDEX CONCURRENTLY idx ON orders … │ │ why: a plain CREATE INDEX takes SHARE and blocks writes │ │ for the whole build; CONCURRENTLY does not. │ └──────────────────────────────────────────────────────────-┘ - │ Phase 3: apply recommendation │ Phase 3: insist on literal - ▼ ▼ - execute the safe idiom --force ⇒ DANGER prompt + - (classified sequence) explicit approval, then run as-is + │ migrate (default): │ migrate --force : + │ apply recommendation │ insist on literal + ▼ ▼ + execute the safe idiom typed acknowledgement of the + (classified sequence) resolved table, then run as-is + under the same budgets ``` Examples of what it suggests (the same idioms the classifier already knows): @@ -232,17 +235,17 @@ Two principles govern this: that must be detected and rebuilt — see the [online DDL reference](postgres-online-ddl-reference.md)), which is why the engine owns executing it rather than handing it to the user to run manually. -- **The planned force route is loud and explicit.** Phase 3 adds a `--force` - (run-as-submitted) flag for the rare case where the operator genuinely wants the literal - statement. It will be gated behind - prominent **DANGER / CAUTION** output explaining exactly what will block and for how long, and - 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, `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`. +- **The force route is loud and explicit.** `--force` (run-as-submitted) exists for the rare + case where the operator genuinely wants the literal statement. It is gated behind an + **explicit typed acknowledgement** — the flag's value must name the resolved schema-qualified + target table, not a bare `-y` — the override is logged unconditionally and recorded in the + verdict's `forced` field, and the statement still runs under the executor's budgets and the + size guard. Force is an escape hatch, not a convenience (mechanics in the + [low-level design](low-level-design.md#the---force-gate)). + +The classifier constructs safer sequences, `diff` / `migrate --dry-run` render them, and +default `migrate` substitutes and executes them; the submitted form runs as-is only when it is +already the safe idiom or under an acknowledged `--force`. 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/low-level-design.md b/docs/low-level-design.md index 6a963ff..98742bf 100644 --- a/docs/low-level-design.md +++ b/docs/low-level-design.md @@ -337,35 +337,37 @@ Classification belongs to `pkg/planner`; `pkg/statement` supplies typed operatio | --- | --- | | `lint` | Offline (no database): classify every statement with zero live facts and report typed findings — unsupported operations are errors (non-zero exit), blocking idioms (with the safer SQL), conservative rewrites, and destructive drops are warnings. | | `diff` / `migrate --dry-run` | Print the classified, routed plan and safer SQL where applicable. **Never writes the live table.** `diff` has no `--dry-run` flag because it never executes the plan (its desired-state diff does run the desired DDL in an always-rolled-back scratch transaction — see the scratch-schema note above). | -| `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, and the -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. +| `migrate` (default) | Run the statement gate, classify and route exactly as dry-run would, then execute the routed SQL: the planner's **safer native sequence by default** when the submitted form blocks, the submitted form as a bounded optimistic attempt otherwise. The verdict's `executed_sql` reports the substitution. | +| `migrate --force` | Run the statement **exactly as submitted**, overriding a safer-sequence substitution or a rewrite-required / backend-unavailable refusal. Gated — see below. | + +The classifier constructs `CREATE INDEX CONCURRENTLY` and other safer sequences, and `migrate` +executes them through `pkg/executor`'s sequence executor 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 driven: `migrate`'s statement gate +refuses `DROP INDEX` and `REINDEX` with the safer-idiom pointer, and the sequence executor's +whole-sequence admission refuses a substituted `DETACH PARTITION CONCURRENTLY` typed before +anything executes — a cancelled wait leaves recovery states it does not yet own. When the +planner says a safer sequence is required but cannot construct one, `migrate` refuses +(`not-native-safe-rewrite-required`) rather than run the blocking form the plan itself +flagged. ### The `--force` gate -This entire gate is planned Phase 3 behavior; no `--force` flag exists today. It is deliberately -high-friction: +`--force` is deliberately high-friction: -1. Print a prominent **DANGER / CAUTION** block: the exact statement, the lock it will take, what - it blocks (reads? writes?), and the expected/worst-case duration and lock-queue impact - (cross-link to 12-mysql-vs-postgresql.md § the lock queue). -2. Require an **explicit typed acknowledgement** (e.g. type the table name, or - `--i-understand-the-risk`), not a bare `-y`/`--yes`. -3. Still wrap the statement in `lock_timeout` + bounded retry unless the user *also* opts out of - that explicitly (a second, separate flag) — force means "run my statement", not "remove every - guardrail". -4. **Log the override** (who, when, what statement, what the recommendation was) for audit. +1. Require an **explicit typed acknowledgement**: the flag's value must name the resolved + schema-qualified target table exactly — the operator names the relation whose lock they are + accepting. A mismatch is a usage error; nothing executes. +2. Still run under `lock_timeout` / `statement_timeout` budgets and the table-size guard — + force overrides the *routing*, never the executor's protections. There is no opt-out: + force means "run my statement", not "remove every guardrail". +3. **Log the override** unconditionally (warn-level, not gated by `--debug`) and record it + machine-readably in the verdict's `forced` field for audit. +4. Planner refusals (no known safe path) and unsupported statement kinds are **not** forceable — + there is nothing bounded to acknowledge. -Force is planned for the rare legitimate case (e.g. a maintenance window where the table is known +Force exists for the rare legitimate case (e.g. a maintenance window where the table is known idle and a plain rewrite is acceptable); it is an escape hatch, not a shortcut, consistent with *decisions, not options*. @@ -796,21 +798,19 @@ path exists in `pkg/executor` — session-scoped, outside any transaction, under wait policy (no per-lock timeout, one overall deadline), with invalid-index detection that fails closed into a typed, state-specific outcome (the executor never drops an index: a name-based drop cannot prove ownership until the LK-1 lease exists; the operator runbook is -[invalid-index-recovery.md](invalid-index-recovery.md)). The executor is deliberately not yet -reachable from the CLI — the engine lands first, the front door next. Remaining Phase 3 work, -roughly in order: - -- the CLI front door for the native path: `migrate` routing an admitted statement to the - executor, resolving an unqualified table name once against the session's `search_path` and - re-emitting the qualified statement (the library-level `ErrUnqualifiedTable` refusal stays; - the CLI moves the qualification burden off the user), and rendering the typed outcomes — - 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, +[invalid-index-recovery.md](invalid-index-recovery.md)). The CLI front door for the native +path is wired: `migrate` routes an admitted statement through classify → route, resolves an +unqualified table name once against the session's `search_path` and re-emits the qualified +statement (the library-level `ErrUnqualifiedTable` refusal stays; the CLI moves the +qualification burden off the user), substitutes and executes classifier-produced safer +sequences by default with the guarded `--force` escape hatch, and renders the typed outcomes — +each executor outcome carries a stable string code in the report contracts +(`executor.OutcomeCode`), the same treatment `pkg/lint` gave its findings, so orchestrators +branch on one vocabulary. Remaining Phase 3 work, roughly in order: + - 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 - captures for its ownership proof). +- progress reporting (`pg_stat_progress_create_index` by the build's backend PID, which the + executor already captures for its ownership proof). The copy-and-swap backend, including change capture, copying, applying, checksumming, and cutover, follows Phase 3. diff --git a/docs/postgres-online-ddl-reference.md b/docs/postgres-online-ddl-reference.md index 3a66855..39727c2 100644 --- a/docs/postgres-online-ddl-reference.md +++ b/docs/postgres-online-ddl-reference.md @@ -216,8 +216,8 @@ cutover) are the **table-rewrite** ones, because PostgreSQL cannot do them onlin **Everything else can be done natively-safe** with PostgreSQL's own `CONCURRENTLY` / `NOT VALID`+`VALIDATE` / fast-default / `USING INDEX` patterns. By the [*classify-before-copy* principle](design-principles.md#classify-first-leverage-native-postgresql), -the plan **detects those and routes them to native DDL** instead of a copy; executing the routed -backend lands in Phase 3 — the same bypass Spirit applies when it attempts `INSTANT`/`INPLACE` +the plan **detects those and routes them to native DDL** instead of a copy, and `migrate` +executes the routed sequence — the same bypass Spirit applies when it attempts `INSTANT`/`INPLACE` before falling back to a table copy. A shadow-table copy is the last resort, used only when no native online path exists. diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 710a346..f564827 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -74,16 +74,33 @@ func (f DBFlags) diag() *slog.Logger { return slog.New(slog.NewTextHandler(out, &slog.HandlerOptions{Level: slog.LevelDebug})) } -// MigrateCmd runs a schema change (imperative front-end): the Phase 1 -// optimistic front door. Easy changes execute directly under tight budgets; -// everything else is refused with an explicit verdict. +// audit returns the operator audit logger: warn-level text on stderr (or the +// test override), always on — an audit record of a deliberate safety +// override must not depend on --debug. The machine-readable counterpart is +// the verdict itself. +func (f DBFlags) audit() *slog.Logger { + out := f.diagOut + if out == nil { + out = os.Stderr + } + return slog.New(slog.NewTextHandler(out, &slog.HandlerOptions{Level: slog.LevelWarn})) +} + +// MigrateCmd runs a schema change (imperative front-end): classify the +// statement, substitute the planner's safer native sequence by default when +// the submitted form blocks, and execute every step under bounded budgets. +// Everything the engine cannot run safely is refused with an explicit +// verdict. type MigrateCmd struct { DBFlags `embed:""` - Alter string `help:"Imperative ALTER statement to run." name:"alter" required:""` - MaxTableSize byteSize `help:"Size threshold above which the optimistic attempt is skipped, measured as the table's full on-disk footprint: heap, indexes, and TOAST, all partitions (binary units: B, KiB, MiB, GiB, TiB)." default:"1GiB"` - DryRun bool `help:"Classify and route the statement, print the plan, and execute nothing."` - JSON bool `help:"Emit the verdict (or dry-run plan) as JSON."` + Alter string `help:"Imperative ALTER statement to run." name:"alter" required:""` + MaxTableSize byteSize `help:"Size threshold above which the optimistic attempt is skipped, measured as the table's full on-disk footprint: heap, indexes, and TOAST, all partitions (binary units: B, KiB, MiB, GiB, TiB). Planner-proven online steps (concurrent index builds, constraint validation) are not size-guarded." default:"1GiB"` + IndexBuildTimeout time.Duration `help:"Overall bound (statement_timeout) for one concurrent index build step; expect large tables to need a generous value." default:"30m"` + ValidateTimeout time.Duration `help:"Overall bound (statement_timeout) for one VALIDATE CONSTRAINT step; expect large tables to need a generous value." default:"30m"` + Force string `help:"Run the submitted form as-is, overriding a safer-sequence substitution or a rewrite-required/backend-unavailable refusal. The value is the typed acknowledgement: it must name the resolved schema-qualified target table exactly. The forced run is still parsed, preflighted, size-guarded, and budget-bounded; planner refusals (no known safe path) and unsupported statement kinds cannot be forced." placeholder:"SCHEMA.TABLE"` + DryRun bool `help:"Classify and route the statement, print the plan, and execute nothing."` + JSON bool `help:"Emit the verdict (or dry-run plan) as JSON."` } // Run implements the migrate subcommand. diff --git a/internal/cli/migrate.go b/internal/cli/migrate.go index 3e431c2..6d3bbbc 100644 --- a/internal/cli/migrate.go +++ b/internal/cli/migrate.go @@ -5,20 +5,27 @@ import ( "errors" "fmt" "io" + "log/slog" "time" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/executor" + "github.com/block/pg-sprite/pkg/planner" "github.com/block/pg-sprite/pkg/preflight" + "github.com/block/pg-sprite/pkg/router" "github.com/block/pg-sprite/pkg/statement" "github.com/block/pg-sprite/pkg/verdict" ) -// run is the migrate flow: gate the statement type, size-guard the table, -// attempt the change under budget, and end in exactly one verdict. Refusal -// verdicts are printed to out and returned as verdict.ErrRefused so the entry -// point maps them to the refusal exit code. --dry-run diverts to the -// classify-and-route plan instead. +// run is the migrate flow: gate the statement type, classify and route it +// exactly as dry-run would, execute the routed SQL — the planner's safer +// native sequence by default when the submitted form blocks — and end in +// exactly one verdict. Refusal verdicts are printed to out and returned as +// verdict.ErrRefused so the entry point maps them to the refusal exit code. +// --dry-run diverts to the classify-and-route plan instead. func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error { if c.DryRun { return c.runDryRun(ctx, out) @@ -39,7 +46,145 @@ func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error { } defer pool.Close() - pt, err := preflight.CheckTable(ctx, pool, st.Schema(), st.Table(), int64(c.MaxTableSize)) + st, err = resolveTarget(ctx, pool, st, logger) + if err != nil { + return err + } + if c.Force != "" { + if err := c.checkForceAck(st); err != nil { + return err + } + } + + facts, err := dryRunFacts(ctx, pool, st) + if err != nil { + return err + } + canonical, err := statement.Canonical(st.SQL()) + if err != nil { + return err + } + classified, err := planner.Classify(canonical, facts) + if err != nil { + return err + } + routed := router.Route([]planner.Plan{classified}) + rs := routed.Statements[0] + logger.Debug("statement routed", + "route", string(classified.Route), "disposition", string(rs.Disposition)) + + switch rs.Disposition { + case router.DispositionExecute: + execSQL := rs.ExecSQL + substituted := len(execSQL) != 1 || execSQL[0] != rs.Statement + forced := substituted && c.Force != "" + if forced { + // The acknowledged override: run the submitted form as a + // blind bounded attempt instead of the safer sequence. + execSQL, substituted = []string{canonical}, false + c.auditForce(st, rs) + } + return c.execute(ctx, out, pool, st, execSQL, rs.Plan, substituted, forced, logger) + case router.DispositionRewriteRequired: + if c.Force == "" { + return c.emit(out, rewriteRequiredVerdict(st)) + } + c.auditForce(st, rs) + return c.execute(ctx, out, pool, st, []string{canonical}, rs.Plan, false, true, logger) + case router.DispositionUnavailable: + if c.Force == "" { + return c.emit(out, backendUnavailableVerdict(st, rs)) + } + c.auditForce(st, rs) + return c.execute(ctx, out, pool, st, []string{canonical}, rs.Plan, false, true, logger) + case router.DispositionRefuse: + // A planner refusal means no known safe path — there is nothing + // bounded to acknowledge, so --force does not apply. + return c.emit(out, routeRefusalVerdict(st, rs)) + default: + // A disposition this build does not know is a router/CLI version + // skew; refuse to act rather than guess. + return fmt.Errorf("unknown disposition %q", rs.Disposition) + } +} + +// resolveTarget qualifies an unqualified statement against the session's +// search_path exactly once and re-emits it in schema-qualified form, so +// every later stage — facts, classification, the planner's safer sequences, +// preflight, and the executor — names the same relation regardless of any +// session's search_path. The executor's own unqualified-table refusals stay: +// this is the CLI resolving its user's intent, not the executor trusting a +// name. Statements already qualified (or without a table target) pass +// through unchanged. +func resolveTarget(ctx context.Context, pool *pgxpool.Pool, st statement.Statement, + logger *slog.Logger) (statement.Statement, error) { + if st.Schema() != "" || st.Table() == "" { + return st, nil + } + // Re-emitting goes through the deparser, which drops comments; refuse + // commented input instead of silently discarding content. + if err := statement.CheckNoComments(st.SQL()); err != nil { + return statement.Statement{}, err + } + const q = ` + SELECT n.nspname + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.oid = to_regclass(quote_ident($1))` + var schema string + err := pool.QueryRow(ctx, q, st.Table()).Scan(&schema) + if errors.Is(err, pgx.ErrNoRows) { + return statement.Statement{}, fmt.Errorf("%w: %s is not visible on the session search_path", + preflight.ErrTableNotFound, st.Table()) + } + if err != nil { + return statement.Statement{}, fmt.Errorf("resolve %s against search_path: %w", st.Table(), err) + } + sql, err := statement.Qualify(st.SQL(), schema) + if err != nil { + return statement.Statement{}, fmt.Errorf("qualify %s as %s.%s: %w", st.Table(), schema, st.Table(), err) + } + logger.Debug("unqualified table resolved", "table", st.Table(), "schema", schema) + return statement.ParseOne(sql) +} + +// checkForceAck validates the --force acknowledgement: it must name the +// resolved schema-qualified target table exactly, proving the operator +// names the relation whose lock they are accepting. A mismatch is a usage +// error — nothing has executed. +func (c *MigrateCmd) checkForceAck(st statement.Statement) error { + if c.Force == qualified(st) { + return nil + } + return fmt.Errorf("--force must acknowledge the resolved target table %q, got %q; nothing was executed", + qualified(st), c.Force) +} + +// auditForce records the override decision before anything executes: the +// operator chose the submitted form over the engine's routing. The record +// is warn-level and unconditional — an audit trail must not depend on +// --debug — and the verdict's Forced field is its machine-readable twin. +func (c *MigrateCmd) auditForce(st statement.Statement, rs router.Statement) { + c.audit().Warn("forced execution of submitted form", + "table", qualified(st), + "kind", st.Kind().String(), + "disposition", string(rs.Disposition)) +} + +// execute runs execSQL through the sequence executor: the planner's safer +// sequence when one was substituted, otherwise the submitted form. Blind +// attempts of the submitted form — including forced ones — are size-guarded; +// substituted sequences and planner-proven online idioms are not — long work +// on large tables is their purpose, and every brief step is still +// budget-bounded. +func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.Pool, + st statement.Statement, execSQL []string, plan planner.Plan, + substituted, forced bool, logger *slog.Logger) error { + limit := int64(c.MaxTableSize) + if !sizeGuardApplies(plan, substituted) { + limit = preflight.NoSizeLimit + } + pt, err := preflight.CheckTable(ctx, pool, st.Schema(), st.Table(), limit) var sizeErr *preflight.SizeError if errors.As(err, &sizeErr) { return c.emit(out, sizeGuardVerdict(st, sizeErr)) @@ -48,29 +193,75 @@ func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error { return err } logger.Debug("preflight passed", - "table", qualified(st), "total_bytes", pt.TotalBytes(), "limit_bytes", int64(c.MaxTableSize)) + "table", qualified(st), "total_bytes", pt.TotalBytes(), "limit_bytes", limit) + if substituted { + logger.Debug("substituting safer native sequence", + "table", qualified(st), "steps", len(execSQL)) + } - budget := executor.Budget{LockTimeout: c.LockTimeout, StatementTimeout: c.StatementTimeout} + budget := executor.SequenceBudget{ + Brief: executor.Budget{LockTimeout: c.LockTimeout, StatementTimeout: c.StatementTimeout}, + Concurrent: executor.ConcurrentBudget{Overall: c.IndexBuildTimeout}, + Validate: executor.ValidateBudget{LockTimeout: c.LockTimeout, Overall: c.ValidateTimeout}, + } start := time.Now() - err = executor.AttemptNative(ctx, pool, pt, st, budget) + _, err = executor.RunSequence(ctx, pool, pt, execSQL, budget) elapsed := time.Since(start) var budgetErr *executor.BudgetError - if errors.As(err, &budgetErr) { + if err != nil && !substituted && errors.As(err, &budgetErr) { + // The blind attempt of the submitted form exceeded a budget and was + // cancelled without committing — the Phase 1 refusal contract. A + // forced attempt is bounded by the same budgets: --force overrides + // routing, never the executor's protections. logger.Debug("optimistic attempt cancelled", "cause", budgetErr.Cause, "budget", budgetErr.Budget, "elapsed", elapsed) return c.emit(out, budgetVerdict(st, budgetErr)) } if err != nil { - return err + // A failed substituted step is an operational failure, not a + // refusal: the typed *SequenceStepError names the failed step and + // the committed prefix that remains. + return fmt.Errorf("run schema change on %s: %w", qualified(st), err) } - logger.Debug("optimistic attempt committed", "table", qualified(st), "elapsed", elapsed) - return c.emit(out, verdict.Verdict{ + logger.Debug("schema change committed", + "table", qualified(st), "steps", len(execSQL), "elapsed", elapsed) + + v := verdict.Verdict{ Outcome: verdict.OutcomeExecuted, Statement: st.SQL(), Table: qualified(st), + Forced: forced, Detail: fmt.Sprintf("committed within budgets (lock %s, statement %s): the change was effectively instant", - budget.LockTimeout, budget.StatementTimeout), - }) + c.LockTimeout, c.StatementTimeout), + } + if substituted { + v.ExecutedSQL = execSQL + v.Detail = fmt.Sprintf("the submitted form blocks; pg-sprite ran the safer native sequence instead — all %d steps committed", + len(execSQL)) + } + if forced { + v.Detail = fmt.Sprintf("forced: the submitted form ran as-is under budgets (lock %s, statement %s), overriding the engine's routing", + c.LockTimeout, c.StatementTimeout) + } + return c.emit(out, v) +} + +// sizeGuardApplies reports whether the size guard protects this run. It +// guards exactly the blind attempt of the submitted form: when the engine +// substituted the planner's safer sequence, or when the plan proved every +// operation an online idiom (CONCURRENTLY, NOT VALID, VALIDATE), long work +// on a large table is the pattern's purpose and the guard would refuse the +// very tables the pattern serves. +func sizeGuardApplies(p planner.Plan, substituted bool) bool { + if substituted { + return false + } + for _, d := range p.Decisions { + if d.Reason != planner.ReasonOnlineIdiom { + return true + } + } + return false } // emit prints the verdict in the selected format and returns ErrRefused for @@ -92,15 +283,18 @@ func (c *MigrateCmd) emit(out io.Writer, v verdict.Verdict) error { return nil } -// gateVerdict is the Phase 1 statement-type gate: only ALTER TABLE proceeds; -// index maintenance is pointed at its concurrent idiom, everything else is -// unsupported. Refused statements are never executed. +// gateVerdict is the statement-type gate: ALTER TABLE and CREATE INDEX +// proceed to classification (a blocking CREATE INDEX is substituted with +// its concurrent build, a submitted concurrent build is driven directly); +// the index-maintenance forms the executor cannot drive yet are pointed at +// their concurrent idiom, everything else is unsupported. Refused +// statements are never executed. func gateVerdict(st statement.Statement) (verdict.Verdict, bool) { v := verdict.Verdict{Outcome: verdict.OutcomeRefused, Statement: st.SQL()} switch st.Kind() { - case statement.KindAlterTable: + case statement.KindAlterTable, statement.KindCreateIndex: return verdict.Verdict{}, false - case statement.KindCreateIndex, statement.KindDropIndex, statement.KindReindex: + case statement.KindDropIndex, statement.KindReindex: v.Reason = verdict.ReasonIndexStatement v.Detail, v.SaferIdiom = indexAdvice(st) case statement.KindCreateTable: @@ -109,21 +303,20 @@ func gateVerdict(st statement.Statement) (verdict.Verdict, bool) { v.SaferIdiom = "pg-sprite diff --desired schema.sql" case statement.KindOther: v.Reason = verdict.ReasonUnsupportedStatement - v.Detail = "only ALTER TABLE statements are supported by the optimistic front door" + v.Detail = "only ALTER TABLE and CREATE INDEX statements are supported by the imperative front door" } return v, true } -// indexAdvice explains an index-statement refusal. The already-concurrent +// indexAdvice explains an index-statement refusal for the maintenance forms +// the executor does not drive (DROP INDEX, REINDEX). The already-concurrent // forms carry no safer idiom: suggesting the statement the user submitted // would confuse a human once and send a resubmitting automation into a loop. func indexAdvice(st statement.Statement) (detail, saferIdiom string) { if st.Concurrent() { - return "this is already the safe concurrent idiom; pg-sprite does not drive index maintenance yet — run it directly against the database", "" + return "this is already the safe concurrent idiom; pg-sprite does not drive this maintenance form yet — run it directly against the database", "" } switch st.Kind() { - case statement.KindCreateIndex: - return "a plain CREATE INDEX blocks writes for the whole build; the concurrent build does not", "CREATE INDEX CONCURRENTLY" case statement.KindDropIndex: return "a plain DROP INDEX takes ACCESS EXCLUSIVE on the table; the concurrent drop does not", "DROP INDEX CONCURRENTLY" case statement.KindReindex: @@ -133,6 +326,55 @@ func indexAdvice(st statement.Statement) (detail, saferIdiom string) { } } +// rewriteRequiredVerdict is the refusal for a statement whose submitted +// form blocks but for which the planner could not construct the safer +// native sequence — a multi-operation statement, or a pattern it cannot +// build. Running the submitted form would falsify the plan's own reason. +func rewriteRequiredVerdict(st statement.Statement) verdict.Verdict { + return verdict.Verdict{ + Outcome: verdict.OutcomeRefused, + Reason: verdict.ReasonRewriteRequired, + Statement: st.SQL(), + Table: qualified(st), + Detail: "the submitted form blocks and must run as a safer native sequence, but pg-sprite could not " + + "construct one for this statement; submit each operation as its own single-operation statement " + + "so the engine can build its safer form (run with --dry-run to see each operation's classification)", + } +} + +// backendUnavailableVerdict is the refusal for a change that routes to an +// execution strategy this build does not implement. +func backendUnavailableVerdict(st statement.Statement, rs router.Statement) verdict.Verdict { + return verdict.Verdict{ + Outcome: verdict.OutcomeRefused, + Reason: verdict.ReasonBackendUnavailable, + Statement: st.SQL(), + Table: qualified(st), + Detail: fmt.Sprintf("the change requires the %s strategy, which this build does not implement yet: "+ + "PostgreSQL would rewrite the table under ACCESS EXCLUSIVE for the whole operation", rs.Backend), + } +} + +// routeRefusalVerdict is the refusal for a statement the planner refused: +// it carries the refused operations by name so the operator knows which +// part of the statement the engine does not know a safe path for. +func routeRefusalVerdict(st statement.Statement, rs router.Statement) verdict.Verdict { + v := verdict.Verdict{ + Outcome: verdict.OutcomeRefused, + Reason: verdict.ReasonUnsupportedStatement, + Statement: st.SQL(), + Table: qualified(st), + Detail: "the planner knows no safe path for this statement", + } + for _, d := range rs.Decisions { + if d.Route == planner.RouteRefuse { + v.Detail = fmt.Sprintf("the planner knows no safe path for %s", d.Operation) + break + } + } + return v +} + // sizeGuardVerdict is the refusal for tables above the size threshold, where // even a budget-bounded attempt would visibly stall the table. func sizeGuardVerdict(st statement.Statement, sizeErr *preflight.SizeError) verdict.Verdict { diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index a8889d2..1e094de 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -19,6 +19,7 @@ import ( "github.com/block/pg-sprite/internal/testutil" "github.com/block/pg-sprite/pkg/dbconn" + "github.com/block/pg-sprite/pkg/preflight" "github.com/block/pg-sprite/pkg/verdict" ) @@ -30,8 +31,10 @@ func newMigrateCmd(url, alter string) *MigrateCmd { LockTimeout: 3 * time.Second, StatementTimeout: 30 * time.Second, }, - Alter: alter, - MaxTableSize: 1 << 30, + Alter: alter, + MaxTableSize: 1 << 30, + IndexBuildTimeout: time.Minute, + ValidateTimeout: time.Minute, } } @@ -90,9 +93,10 @@ func TestMigrateExecutesRenameColumn(t *testing.T) { assert.Equal(t, 1, n, "the rename must have committed") } -// Acceptance (ii): a rewrite-requiring change is cancelled, leaves schema and -// data unchanged, and returns the not-native-safe verdict with its reason. -func TestMigrateRefusesRewriteWithBudgetVerdict(t *testing.T) { +// Acceptance (ii): a rewrite-requiring change is refused up front — the +// planner classifies the type change as copy-and-swap before anything runs, +// so no attempt ever holds ACCESS EXCLUSIVE doing rewrite work. +func TestMigrateRefusesTypeRewriteAsUnavailable(t *testing.T) { url := testutil.StartPostgres(t) pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) require.NoError(t, err) @@ -101,11 +105,10 @@ func TestMigrateRefusesRewriteWithBudgetVerdict(t *testing.T) { _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) require.NoError(t, err) _, err = pool.Exec(t.Context(), fmt.Sprintf( - "INSERT INTO %s.t SELECT g, repeat('x', 100) FROM generate_series(1, 300000) g", schema)) + "INSERT INTO %s.t SELECT g, repeat('x', 100) FROM generate_series(1, 1000) g", schema)) require.NoError(t, err) cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) - cmd.StatementTimeout = 50 * time.Millisecond cmd.JSON = true var out strings.Builder err = cmd.run(t.Context(), &out) @@ -114,19 +117,380 @@ func TestMigrateRefusesRewriteWithBudgetVerdict(t *testing.T) { var v verdict.Verdict require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) assert.Equal(t, verdict.OutcomeRefused, v.Outcome) - assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) - assert.Equal(t, verdict.CauseStatementBudget, v.Cause) + assert.Equal(t, verdict.ReasonBackendUnavailable, v.Reason) assert.Equal(t, schema+".t", v.Table) var typ string require.NoError(t, pool.QueryRow(t.Context(), `SELECT data_type FROM information_schema.columns WHERE table_schema = $1 AND table_name = 't' AND column_name = 'id'`, schema).Scan(&typ)) - assert.Equal(t, "integer", typ, "the cancelled attempt must not change the schema") + assert.Equal(t, "integer", typ, "the refused change must not touch the schema") var count int require.NoError(t, pool.QueryRow(t.Context(), fmt.Sprintf("SELECT count(*) FROM %s.t", schema)).Scan(&count)) - assert.Equal(t, 300000, count, "the cancelled attempt must not change the data") + assert.Equal(t, 1000, count, "the refused change must not touch the data") +} + +// A blind attempt that cannot get its lock is cancelled by the lock budget +// and refused with the typed cause — the Phase 1 refusal contract, now +// reached through the routed execute path. +func TestMigrateRefusesOnLockBudgetContention(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) + require.NoError(t, err) + + // An open transaction holding ACCESS SHARE on the table makes the + // ALTER's ACCESS EXCLUSIVE unobtainable until rollback. + tx, err := pool.Begin(t.Context()) + require.NoError(t, err) + defer func() { + if err := tx.Rollback(context.WithoutCancel(t.Context())); err != nil { + t.Logf("rollback lock-holding transaction: %v", err) + } + }() + _, err = tx.Exec(t.Context(), fmt.Sprintf("SELECT count(*) FROM %s.t", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN age int", schema)) + cmd.LockTimeout = 100 * time.Millisecond + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) + assert.Equal(t, verdict.CauseLockBudget, v.Cause) +} + +// The engine substitutes the planner's safer sequence by default: a direct +// SET NOT NULL runs as the NOT VALID + VALIDATE + SET NOT NULL + DROP +// scaffold sequence, the verdict reports what actually ran, and no scaffold +// constraint is left behind. +func TestMigrateSubstitutesSetNotNullSequence(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + _, err = pool.Exec(t.Context(), fmt.Sprintf( + "INSERT INTO %s.t SELECT g, 'v' FROM generate_series(1, 1000) g", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN v SET NOT NULL", schema)) + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.Len(t, v.ExecutedSQL, 4, "the four-step SET NOT NULL sequence must be reported") + + var notNull bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT attnotnull FROM pg_attribute + WHERE attrelid = ($1 || '.t')::regclass AND attname = 'v'`, schema).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 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 = 't' AND con.contype = 'c'`, schema).Scan(&scaffolds)) + assert.Zero(t, scaffolds, "the scaffold CHECK constraint must be dropped") +} + +// A blocking CREATE INDEX is substituted with its concurrent build and +// driven to a verified valid index. +func TestMigrateSubstitutesBlockingCreateIndex(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("CREATE INDEX t_c_idx ON %s.t (c)", schema)) + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + require.Len(t, v.ExecutedSQL, 1, "the substituted concurrent build must be reported") + + var valid bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT i.indisvalid FROM pg_index i + WHERE i.indexrelid = ($1 || '.t_c_idx')::regclass`, schema).Scan(&valid)) + assert.True(t, valid, "the index must be built and valid") +} + +// A submitted CREATE INDEX CONCURRENTLY is already the online idiom: the +// engine drives it directly, with no substitution to report. +func TestMigrateRunsSubmittedConcurrentIndexBuild(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("CREATE INDEX CONCURRENTLY t_cic_idx ON %s.t (c)", schema)) + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.Empty(t, v.ExecutedSQL, "no substitution happened, so none is reported") + + var valid bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT i.indisvalid FROM pg_index i + WHERE i.indexrelid = ($1 || '.t_cic_idx')::regclass`, schema).Scan(&valid)) + assert.True(t, valid, "the index must be built and valid") +} + +// A safer-idiom decision without a constructible rewrite (an inline +// constraint on ADD COLUMN) is refused as rewrite-required — running the +// submitted form would falsify the plan's own reason. +func TestMigrateRefusesInlineConstraintAsRewriteRequired(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ADD COLUMN e int UNIQUE", schema)) + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + assert.Equal(t, verdict.ReasonRewriteRequired, v.Reason) + + var n int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 't' AND column_name = 'e'`, schema).Scan(&n)) + assert.Zero(t, n, "the refused change must not execute") +} + +// R3: an unqualified statement is resolved once against the session's +// search_path and re-emitted qualified, so the substituted concurrent index +// build — which the executor refuses to run against an unqualified name — +// succeeds, and the verdict names the resolved relation. +func TestMigrateResolvesUnqualifiedTableViaSearchPath(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + + // The command's sessions resolve the bare name through search_path. + u, err := neturl.Parse(url) + require.NoError(t, err) + q := u.Query() + q.Set("options", "-csearch_path="+schema) + u.RawQuery = q.Encode() + cmd := newMigrateCmd(u.String(), "CREATE INDEX t_c_idx ON t (c)") + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.Equal(t, schema+".t", v.Table, "the verdict must carry the resolved qualified name") + require.Len(t, v.ExecutedSQL, 1) + assert.Contains(t, v.ExecutedSQL[0], schema+".t", "the executed SQL must be schema-qualified") + + var valid bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT i.indisvalid FROM pg_index i + WHERE i.indexrelid = ($1 || '.t_c_idx')::regclass`, schema).Scan(&valid)) + assert.True(t, valid, "the index must be built and valid") +} + +// An unqualified name that resolves nowhere on the session search_path is an +// operational error before anything is classified or executed. +func TestMigrateUnresolvableTableIsAnError(t *testing.T) { + url := testutil.StartPostgres(t) + cmd := newMigrateCmd(url, "ALTER TABLE nowhere_to_be_found ADD COLUMN c int") + var out strings.Builder + err := cmd.run(t.Context(), &out) + require.ErrorIs(t, err, preflight.ErrTableNotFound) + assert.Empty(t, out.String(), "no verdict is printed for an operational error") +} + +// --force with the exact qualified-table acknowledgement runs the submitted +// form as-is: no substitution happens, the verdict records the override, and +// the change commits. +func TestMigrateForceRunsSubmittedFormOverSubstitution(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + _, err = pool.Exec(t.Context(), fmt.Sprintf( + "INSERT INTO %s.t SELECT g, 'v' FROM generate_series(1, 1000) g", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN v SET NOT NULL", schema)) + cmd.Force = schema + ".t" + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.True(t, v.Forced, "the verdict must record the override") + assert.Empty(t, v.ExecutedSQL, "the submitted form ran as-is; no substitution to report") + + var notNull bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT attnotnull FROM pg_attribute + WHERE attrelid = ($1 || '.t')::regclass AND attname = 'v'`, schema).Scan(¬Null)) + assert.True(t, notNull, "the forced change must have committed") +} + +// --force also overrides a backend-unavailable refusal: the rewrite-carrying +// type change runs as a blind bounded attempt and commits on a small table. +func TestMigrateForceRunsUnavailableRewrite(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) + cmd.Force = schema + ".t" + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.True(t, v.Forced) + + var typ string + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT data_type FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 't' AND column_name = 'id'`, schema).Scan(&typ)) + assert.Equal(t, "bigint", typ, "the forced rewrite must have committed") +} + +// A --force acknowledgement that does not name the resolved target table is +// a usage error: nothing executes. +func TestMigrateForceAckMismatchExecutesNothing(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN v SET NOT NULL", schema)) + cmd.Force = "wrong.table" + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.Error(t, err) + require.NotErrorIs(t, err, verdict.ErrRefused, "an acknowledgement mismatch is a usage error, not a refusal") + assert.Empty(t, out.String(), "no verdict is printed") + + var notNull bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT attnotnull FROM pg_attribute + WHERE attrelid = ($1 || '.t')::regclass AND attname = 'v'`, schema).Scan(¬Null)) + assert.False(t, notNull, "nothing must have executed") +} + +// --force overrides routing only, never the executor's protections: the +// forced blind attempt is still size-guarded. +func TestMigrateForceKeepsSizeGuard(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + _, err = pool.Exec(t.Context(), fmt.Sprintf( + "INSERT INTO %s.t SELECT g, 'v' FROM generate_series(1, 1000) g", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) + cmd.Force = schema + ".t" + cmd.MaxTableSize = 1 + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.ReasonTableTooLarge, v.Reason) + + var typ string + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT data_type FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 't' AND column_name = 'id'`, schema).Scan(&typ)) + assert.Equal(t, "integer", typ, "the guarded change must not have executed") +} + +// The size guard protects blind attempts only: a substituted safer sequence +// runs on a table above the threshold, because its long steps are online by +// design and its brief steps are budget-bounded. +func TestMigrateSizeGuardSkippedForSubstitutedSequence(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + _, err = pool.Exec(t.Context(), fmt.Sprintf( + "INSERT INTO %s.t SELECT g, 'v' FROM generate_series(1, 1000) g", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN v SET NOT NULL", schema)) + cmd.MaxTableSize = 1 // would refuse a blind attempt; must not gate the sequence + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.NotEmpty(t, v.ExecutedSQL) } // Acceptance (iii): a table above the size threshold skips the attempt and @@ -172,12 +536,13 @@ func TestMigrateGateRefusesWithoutDatabase(t *testing.T) { reason verdict.Reason saferIdiom string }{ - {"create index", "CREATE INDEX i ON t (c)", verdict.ReasonIndexStatement, "CREATE INDEX CONCURRENTLY"}, {"drop index", "DROP INDEX i", verdict.ReasonIndexStatement, "DROP INDEX CONCURRENTLY"}, {"reindex", "REINDEX TABLE t", verdict.ReasonIndexStatement, "REINDEX ... CONCURRENTLY"}, // The already-concurrent forms carry no safer idiom: suggesting the // statement the user submitted would loop a resubmitting automation. - {"create index concurrently", "CREATE INDEX CONCURRENTLY i ON t (c)", verdict.ReasonIndexStatement, ""}, + // CREATE INDEX (both forms) passes the gate: the blocking form is + // substituted with its concurrent build, the concurrent form is + // driven directly. {"drop index concurrently", "DROP INDEX CONCURRENTLY i", verdict.ReasonIndexStatement, ""}, {"reindex concurrently", "REINDEX TABLE CONCURRENTLY t", verdict.ReasonIndexStatement, ""}, {"alter index", "ALTER INDEX i SET (fillfactor = 90)", verdict.ReasonUnsupportedStatement, ""}, diff --git a/pkg/executor/code.go b/pkg/executor/code.go new file mode 100644 index 0000000..8ab03cf --- /dev/null +++ b/pkg/executor/code.go @@ -0,0 +1,158 @@ +// This file is the executor's stable outcome vocabulary: every typed +// failure this package can return maps to exactly one flat kebab-case +// Code, the same treatment pkg/lint gave its findings. Orchestrators and +// report consumers branch on the code — never on error prose, which is +// free to change — and the mapping is derived from the typed errors +// themselves, so a code cannot drift from the error it names. + +package executor + +import "errors" + +// Code is the stable string identity of one executor outcome; automation +// branches on it, never on error text. Codes are part of the report +// contract: existing values never change meaning, new outcomes add new +// codes. +type Code string + +// The codes an executor outcome can carry. +const ( + // CodeBudgetLockExceeded: the lock was not granted within + // lock_timeout; nothing was executed. + CodeBudgetLockExceeded Code = "budget-lock-exceeded" + // CodeBudgetStatementExceeded: the statement ran past + // statement_timeout and was cancelled; the change does real work. + CodeBudgetStatementExceeded Code = "budget-statement-exceeded" + // CodeCancelledExternally: the build's statement was cancelled from + // outside the executor before its budget elapsed. + CodeCancelledExternally Code = "cancelled-externally" + // CodeInvalidIndexOwnLeftover: the failed build's own invalid index + // remains and is proven this run's leftover; the recovery runbook + // applies. + CodeInvalidIndexOwnLeftover Code = "invalid-index-own-leftover" + // CodeInvalidIndexPreexisting: an invalid index under the requested + // name predates this run; it may be another actor's build in progress. + CodeInvalidIndexPreexisting Code = "invalid-index-preexisting" + // CodeInvalidIndexUnproven: an invalid index may remain but the + // catalog state could not be proven; an operator must inspect. + CodeInvalidIndexUnproven Code = "invalid-index-unproven" + // CodeEmptySequence: the sequence had no steps to run. + CodeEmptySequence Code = "empty-sequence" + // CodeUnsupportedSequenceStep: a step is not a shape the sequence + // executor can run safely. + CodeUnsupportedSequenceStep Code = "unsupported-sequence-step" + // CodeNotConcurrentIndexBuild: the statement handed to the concurrent + // build executor is not a CREATE INDEX CONCURRENTLY. + CodeNotConcurrentIndexBuild Code = "not-concurrent-index-build" + // CodeUnnamedIndex: the concurrent build does not name its index, so + // its outcome could not be verified. + CodeUnnamedIndex Code = "unnamed-index" + // CodeUnqualifiedTable: the target table is not schema-qualified at + // the library boundary. + CodeUnqualifiedTable Code = "unqualified-table" + // CodeIfNotExistsUnsupported: CREATE INDEX CONCURRENTLY IF NOT EXISTS + // cannot prove what its no-op would mean. + CodeIfNotExistsUnsupported Code = "if-not-exists-unsupported" + // CodePoolTooSmall: the pool cannot hold the build session and the + // verdict connection at once. + CodePoolTooSmall Code = "pool-too-small" + // CodeTableNotFound: the statement's qualified table does not exist. + CodeTableNotFound Code = "table-not-found" + // CodeInvariantViolation: a breach of the invariant registry; never a + // retry candidate. + CodeInvariantViolation Code = "invariant-violation" + // CodeExecutionFailed: the fallback for a failure outside the typed + // set — a server error surfaced as-is, a connection failure, a + // context cancellation. Consumers treat it as an operational error to + // investigate, not a refusal to branch on. + CodeExecutionFailed Code = "execution-failed" +) + +// OutcomeCode maps an error returned by this package to its stable code. +// A nil error has no outcome code and maps to the empty Code. A +// *SequenceStepError carries its failed step's own cause, so it maps to +// that underlying code — the step position and committed prefix ride on +// the struct itself, not the vocabulary. An error outside the typed set +// maps to CodeExecutionFailed. +func OutcomeCode(err error) Code { + if err == nil { + return "" + } + var stepErr *SequenceStepError + if errors.As(err, &stepErr) { + return OutcomeCode(stepErr.Err) + } + var invalidErr *InvalidIndexError + if errors.As(err, &invalidErr) { + return invalidErr.Code() + } + var budgetErr *BudgetError + if errors.As(err, &budgetErr) { + return budgetErr.Code() + } + return sentinelCode(err) +} + +// sentinelCode maps the package's sentinel errors to their codes. The +// invariant sentinel is checked first: an invariant breach is the +// fail-closed outcome regardless of which path wrapped it. +func sentinelCode(err error) Code { + switch { + case errors.Is(err, ErrInvariantViolation): + return CodeInvariantViolation + case errors.Is(err, ErrCancelledExternally): + return CodeCancelledExternally + case errors.Is(err, ErrEmptySequence): + return CodeEmptySequence + case errors.Is(err, ErrUnsupportedSequenceStep): + return CodeUnsupportedSequenceStep + case errors.Is(err, ErrNotConcurrentIndexBuild): + return CodeNotConcurrentIndexBuild + case errors.Is(err, ErrUnnamedIndex): + return CodeUnnamedIndex + case errors.Is(err, ErrUnqualifiedTable): + return CodeUnqualifiedTable + case errors.Is(err, ErrIfNotExistsUnsupported): + return CodeIfNotExistsUnsupported + case errors.Is(err, ErrPoolTooSmall): + return CodePoolTooSmall + case errors.Is(err, ErrTableNotFound): + return CodeTableNotFound + default: + return CodeExecutionFailed + } +} + +// Code returns the budget outcome's stable code. +func (e *BudgetError) Code() Code { + switch e.Cause { + case CauseLock: + return CodeBudgetLockExceeded + case CauseStatement: + return CodeBudgetStatementExceeded + default: + // A cause outside the closed set is a programming error; the + // fallback keeps the mapping total without inventing a meaning. + return CodeExecutionFailed + } +} + +// Code returns the invalid-index outcome's stable code, derived from the +// same cleanup state the error's rendering distinguishes: proven own +// leftover, proven preexisting, or unproven. +func (e *InvalidIndexError) Code() Code { + switch { + case errors.Is(e.Cleanup, ErrBuildLeftInvalidIndex): + return CodeInvalidIndexOwnLeftover + case errors.Is(e.Cleanup, ErrPreexistingInvalidIndex): + return CodeInvalidIndexPreexisting + default: + return CodeInvalidIndexUnproven + } +} + +// Code returns the failed step's own cause code; the step position and +// the committed prefix ride on the struct's fields. +func (e *SequenceStepError) Code() Code { + return OutcomeCode(e.Err) +} diff --git a/pkg/executor/code_test.go b/pkg/executor/code_test.go new file mode 100644 index 0000000..6c497f1 --- /dev/null +++ b/pkg/executor/code_test.go @@ -0,0 +1,145 @@ +package executor_test + +import ( + "encoding/json" + "errors" + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/executor" +) + +func TestOutcomeCodeMapsTypedOutcomes(t *testing.T) { + tests := []struct { + name string + err error + want executor.Code + }{ + {name: "nil error has no code", err: nil, want: executor.Code("")}, + { + name: "lock budget", + err: &executor.BudgetError{Cause: executor.CauseLock, Budget: time.Second}, + want: executor.CodeBudgetLockExceeded, + }, + { + name: "statement budget", + err: &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Second}, + want: executor.CodeBudgetStatementExceeded, + }, + { + name: "own invalid leftover", + err: &executor.InvalidIndexError{Schema: "s", Index: "i", Cleanup: executor.ErrBuildLeftInvalidIndex}, + want: executor.CodeInvalidIndexOwnLeftover, + }, + { + name: "preexisting invalid index", + err: &executor.InvalidIndexError{Schema: "s", Index: "i", Cleanup: executor.ErrPreexistingInvalidIndex}, + want: executor.CodeInvalidIndexPreexisting, + }, + { + name: "unproven invalid index", + err: &executor.InvalidIndexError{Schema: "s", Index: "i", Cleanup: executor.ErrTargetIdentityChanged}, + want: executor.CodeInvalidIndexUnproven, + }, + {name: "cancelled externally", err: executor.ErrCancelledExternally, want: executor.CodeCancelledExternally}, + {name: "empty sequence", err: executor.ErrEmptySequence, want: executor.CodeEmptySequence}, + {name: "unsupported sequence step", err: executor.ErrUnsupportedSequenceStep, want: executor.CodeUnsupportedSequenceStep}, + {name: "not a concurrent build", err: executor.ErrNotConcurrentIndexBuild, want: executor.CodeNotConcurrentIndexBuild}, + {name: "unnamed index", err: executor.ErrUnnamedIndex, want: executor.CodeUnnamedIndex}, + {name: "unqualified table", err: executor.ErrUnqualifiedTable, want: executor.CodeUnqualifiedTable}, + {name: "if not exists", err: executor.ErrIfNotExistsUnsupported, want: executor.CodeIfNotExistsUnsupported}, + {name: "pool too small", err: executor.ErrPoolTooSmall, want: executor.CodePoolTooSmall}, + {name: "table not found", err: executor.ErrTableNotFound, want: executor.CodeTableNotFound}, + {name: "invariant violation", err: executor.ErrInvariantViolation, want: executor.CodeInvariantViolation}, + {name: "untyped error is the fallback", err: errors.New("connection reset"), want: executor.CodeExecutionFailed}, + { + name: "wrapped sentinel still maps", + err: fmt.Errorf("sequence step 2 of 3: %w", executor.ErrUnsupportedSequenceStep), + want: executor.CodeUnsupportedSequenceStep, + }, + { + name: "step error carries its cause's code", + err: &executor.SequenceStepError{ + Step: 2, Total: 3, Kind: executor.StepBrief, SQL: "ALTER TABLE s.t ADD c int", + Err: &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Second}, + }, + want: executor.CodeBudgetStatementExceeded, + }, + { + name: "step error with an untyped cause is the fallback", + err: &executor.SequenceStepError{ + Step: 1, Total: 1, Kind: executor.StepBrief, SQL: "ALTER TABLE s.t ADD c int", + Err: errors.New("server closed the connection"), + }, + want: executor.CodeExecutionFailed, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, executor.OutcomeCode(tt.err)) + }) + } +} + +func TestSequenceStepErrorCodeMatchesOutcomeCode(t *testing.T) { + stepErr := &executor.SequenceStepError{ + Step: 1, Total: 2, Kind: executor.StepConcurrentIndexBuild, SQL: "CREATE INDEX CONCURRENTLY i ON s.t (c)", + Err: &executor.InvalidIndexError{Schema: "s", Index: "i", Cleanup: executor.ErrBuildLeftInvalidIndex}, + } + assert.Equal(t, executor.CodeInvalidIndexOwnLeftover, stepErr.Code()) + assert.Equal(t, stepErr.Code(), executor.OutcomeCode(stepErr)) +} + +// TestSequenceReportJSONContractIsStable pins the report's wire shape: +// consumers parse these exact keys, so a key change is a contract break +// this test makes deliberate. +func TestSequenceReportJSONContractIsStable(t *testing.T) { + rep := executor.SequenceReport{ + Steps: []executor.StepReport{ + { + SQL: "ALTER TABLE s.t ADD CONSTRAINT c CHECK (v > 0) NOT VALID", + Kind: executor.StepBrief, + Duration: 20 * time.Millisecond, + }, + { + SQL: "CREATE INDEX CONCURRENTLY i ON s.t (v)", + Kind: executor.StepConcurrentIndexBuild, + Duration: 1500 * time.Millisecond, + Index: &executor.IndexBuildReport{ + Schema: "s", + Index: "i", + IndexOID: 41235, + Duration: 1400 * time.Millisecond, + ServerVersion: "16.4", + }, + }, + }, + } + got, err := json.Marshal(rep) + require.NoError(t, err) + require.JSONEq(t, `{ + "steps": [ + { + "sql": "ALTER TABLE s.t ADD CONSTRAINT c CHECK (v > 0) NOT VALID", + "kind": "brief", + "duration_ns": 20000000 + }, + { + "sql": "CREATE INDEX CONCURRENTLY i ON s.t (v)", + "kind": "concurrent-index-build", + "duration_ns": 1500000000, + "index": { + "schema": "s", + "index": "i", + "index_oid": 41235, + "duration_ns": 1400000000, + "server_version": "16.4" + } + } + ] + }`, string(got)) +} diff --git a/pkg/executor/native.go b/pkg/executor/native.go index eed5c2e..bf69dc2 100644 --- a/pkg/executor/native.go +++ b/pkg/executor/native.go @@ -157,19 +157,20 @@ func (b ConcurrentBudget) validate() error { type IndexBuildReport struct { // Schema is the schema the index lives in, resolved from the target // table (an index is always created in its table's schema). - Schema string + Schema string `json:"schema"` // Index is the index name from the statement. - Index string + Index string `json:"index"` // IndexOID is the verified index's catalog identity: the durable // handle a later reconciliation can use where the name alone could // have been reassigned. - IndexOID uint32 + IndexOID uint32 `json:"index_oid"` // Duration is the wall-clock time of the build statement itself, - // excluding session setup and the validity verification. - Duration time.Duration + // excluding session setup and the validity verification. It encodes + // as integer nanoseconds. + Duration time.Duration `json:"duration_ns"` // ServerVersion is the server_version of the PostgreSQL server that // ran the build. - ServerVersion string + ServerVersion string `json:"server_version"` } // InvalidIndexError reports that an invalid index exists (or may remain) diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go index 4f3350b..eb30d01 100644 --- a/pkg/executor/sequence.go +++ b/pkg/executor/sequence.go @@ -137,22 +137,22 @@ func (b SequenceBudget) validate() error { // StepReport says what one committed step did, machine-readably. type StepReport struct { // SQL is the step's statement as submitted. - SQL string + SQL string `json:"sql"` // Kind is the execution class the step ran under. - Kind StepKind + Kind StepKind `json:"kind"` // Duration is the wall-clock time of the step, session setup and - // verification included. - Duration time.Duration + // verification included. It encodes as integer nanoseconds. + Duration time.Duration `json:"duration_ns"` // Index carries the concurrent build's verified report; nil for every // other step kind. - Index *IndexBuildReport + Index *IndexBuildReport `json:"index,omitempty"` } // 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 + Steps []StepReport `json:"steps"` } // SequenceStepError reports that a step failed and the run stopped there. diff --git a/pkg/preflight/preflight.go b/pkg/preflight/preflight.go index 59bed7c..53a8d4c 100644 --- a/pkg/preflight/preflight.go +++ b/pkg/preflight/preflight.go @@ -13,11 +13,18 @@ import ( "context" "errors" "fmt" + "math" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) +// NoSizeLimit is a size limit no PostgreSQL relation can exceed. Callers +// pass it when the check should prove only existence and kind — the online +// sequence path, whose long steps are safe on any size by design (the size +// guard protects blind attempts, not planner-proven online idioms). +const NoSizeLimit int64 = math.MaxInt64 + // ErrTableNotFound is returned when the target table does not exist (or is // not visible with the session's search_path). var ErrTableNotFound = errors.New("table not found") diff --git a/pkg/statement/desired.go b/pkg/statement/desired.go index bf655e7..cce1f05 100644 --- a/pkg/statement/desired.go +++ b/pkg/statement/desired.go @@ -158,9 +158,9 @@ func refuseForeignKeys(create *pganalyze.CreateStmt) error { // Qualify returns sql with its target relation qualified by schema; an // empty schema strips an existing qualification instead. It supports exactly -// one CREATE TABLE or CREATE INDEX statement. This touches qualification -// only — no semantics are ever derived or transformed at the AST level -// (that is the scratch database's job). +// one CREATE TABLE, CREATE INDEX, or ALTER TABLE statement. This touches +// qualification only — no semantics are ever derived or transformed at the +// AST level (that is the scratch database's job). func Qualify(sql, schema string) (string, error) { tree, err := pgquery.Parse(sql) if err != nil { @@ -176,6 +176,8 @@ func Qualify(sql, schema string) (string, error) { rel = node.GetCreateStmt().GetRelation() case node.GetIndexStmt() != nil: rel = node.GetIndexStmt().GetRelation() + case node.GetAlterTableStmt() != nil: + rel = node.GetAlterTableStmt().GetRelation() default: return "", ErrDisallowedStatement } diff --git a/pkg/statement/desired_test.go b/pkg/statement/desired_test.go index 9b07213..fb949f3 100644 --- a/pkg/statement/desired_test.go +++ b/pkg/statement/desired_test.go @@ -84,6 +84,10 @@ func TestQualify(t *testing.T) { got, err = Qualify("CREATE TABLE t (id int)", "s1") require.NoError(t, err) assert.Equal(t, "CREATE TABLE s1.t (id int)", got) + + got, err = Qualify("ALTER TABLE t ADD COLUMN c int", "s1") + require.NoError(t, err) + assert.Equal(t, "ALTER TABLE s1.t ADD COLUMN c int", got) } func TestQualifyEmptySchemaStripsQualification(t *testing.T) { @@ -93,7 +97,7 @@ func TestQualifyEmptySchemaStripsQualification(t *testing.T) { } func TestQualifyRefusesOtherStatements(t *testing.T) { - _, err := Qualify("ALTER TABLE t ADD COLUMN c int", "s1") + _, err := Qualify("DROP INDEX i", "s1") require.ErrorIs(t, err, ErrDisallowedStatement) _, err = Qualify("CREATE TABLE a (id int); CREATE TABLE b (id int)", "s1") diff --git a/pkg/verdict/verdict.go b/pkg/verdict/verdict.go index 676093e..1f2e0f2 100644 --- a/pkg/verdict/verdict.go +++ b/pkg/verdict/verdict.go @@ -51,6 +51,15 @@ const ( // ReasonBudgetExceeded: the optimistic attempt exceeded its lock or // statement budget and was cancelled. ReasonBudgetExceeded Reason = "not-native-safe-budget-exceeded" + // ReasonRewriteRequired: the submitted form blocks and must run as a + // safer native sequence, but the planner could not construct one (a + // multi-operation statement, or a pattern it cannot build). Running + // the submitted form would falsify the plan's own reason, so the + // engine refuses instead. + ReasonRewriteRequired Reason = "not-native-safe-rewrite-required" + // ReasonBackendUnavailable: the change routes to an execution strategy + // this build does not implement (copy-and-swap). + ReasonBackendUnavailable Reason = "backend-unavailable" ) // Cause narrows ReasonBudgetExceeded to the budget that was exceeded, so @@ -88,6 +97,16 @@ type Verdict struct { // SaferIdiom is a native alternative to the refused statement, when one // exists (e.g. CREATE INDEX CONCURRENTLY, ADD CONSTRAINT ... NOT VALID). SaferIdiom string `json:"safer_idiom,omitempty"` + // ExecutedSQL is the ordered SQL the engine actually ran when it + // substituted the planner's safer native sequence for the submitted + // form. Empty when the submitted form ran as-is — a non-empty value is + // what tells automation a substitution happened. + ExecutedSQL []string `json:"executed_sql,omitempty"` + // Forced reports that --force overrode the engine's routing: the + // submitted form ran as-is instead of a safer substitution or a + // strategy refusal. It is the machine-readable audit record of the + // override. + Forced bool `json:"forced,omitempty"` } // JSON renders the verdict as a single JSON object. @@ -120,5 +139,14 @@ func (v Verdict) String() string { if v.SaferIdiom != "" { fmt.Fprintf(&b, "\n safer: %s", v.SaferIdiom) } + if v.Forced { + b.WriteString("\n forced: the submitted form ran as-is (--force)") + } + if len(v.ExecutedSQL) > 0 { + b.WriteString("\n executed as:") + for i, sql := range v.ExecutedSQL { + fmt.Fprintf(&b, "\n %d. %s", i+1, sql) + } + } return b.String() } From 769b3e8ade506946414c72899ef932ec25ae0188 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 12 Aug 2026 18:45:45 +1000 Subject: [PATCH 2/3] Map execution failures to refusal verdicts; run forced form via AttemptNative MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A forced submitted form was routed through RunSequence, whose shape admission categorically refused it before any DB access — the documented --force CREATE INDEX case could never execute. Forced runs now bypass sequence admission and execute as one blind bounded attempt. Executor admission refusals (unnamed index, IF NOT EXISTS, undriven DETACH PARTITION CONCURRENTLY) now surface as typed refusal verdicts instead of raw operational errors, invalid-index debris is never masked as a plain budget refusal, refused forced attempts keep the machine-readable forced flag, and the statement-budget advice is tailored for submitted online idioms. --- docs/low-level-design.md | 12 +- internal/cli/migrate.go | 157 ++++++++++++++----- internal/cli/migrate_integration_test.go | 188 +++++++++++++++++++++++ internal/cli/migrate_test.go | 87 ++++++++++- 4 files changed, 398 insertions(+), 46 deletions(-) diff --git a/docs/low-level-design.md b/docs/low-level-design.md index 98742bf..05b1f8d 100644 --- a/docs/low-level-design.md +++ b/docs/low-level-design.md @@ -347,7 +347,8 @@ builds under their own budgets). The one-step `CONCURRENTLY` rewrites the classi emits (`DROP INDEX`, `REINDEX`, `DETACH PARTITION`) are not driven: `migrate`'s statement gate refuses `DROP INDEX` and `REINDEX` with the safer-idiom pointer, and the sequence executor's whole-sequence admission refuses a substituted `DETACH PARTITION CONCURRENTLY` typed before -anything executes — a cancelled wait leaves recovery states it does not yet own. When the +anything executes — a cancelled wait leaves recovery states it does not yet own — which +`migrate` reports as a refusal verdict, not an operational error. When the planner says a safer sequence is required but cannot construct one, `migrate` refuses (`not-native-safe-rewrite-required`) rather than run the blocking form the plan itself flagged. @@ -803,10 +804,11 @@ path is wired: `migrate` routes an admitted statement through classify → route unqualified table name once against the session's `search_path` and re-emits the qualified statement (the library-level `ErrUnqualifiedTable` refusal stays; the CLI moves the qualification burden off the user), substitutes and executes classifier-produced safer -sequences by default with the guarded `--force` escape hatch, and renders the typed outcomes — -each executor outcome carries a stable string code in the report contracts -(`executor.OutcomeCode`), the same treatment `pkg/lint` gave its findings, so orchestrators -branch on one vocabulary. Remaining Phase 3 work, roughly in order: +sequences by default with the guarded `--force` escape hatch, and renders the typed outcomes. +At the library seam, each executor outcome maps to a stable string code +(`executor.OutcomeCode`), the same treatment `pkg/lint` gave its findings, so an orchestrator +embedding `pkg/executor` branches on one vocabulary; the CLI's verdict JSON does not carry +the executor codes yet. Remaining Phase 3 work, roughly in order: - bound lock acquisition with timeout and retry for the blocking idioms, - progress reporting (`pg_stat_progress_create_index` by the build's backend PID, which the diff --git a/internal/cli/migrate.go b/internal/cli/migrate.go index 6d3bbbc..62a036f 100644 --- a/internal/cli/migrate.go +++ b/internal/cli/migrate.go @@ -172,11 +172,14 @@ func (c *MigrateCmd) auditForce(st statement.Statement, rs router.Statement) { } // execute runs execSQL through the sequence executor: the planner's safer -// sequence when one was substituted, otherwise the submitted form. Blind -// attempts of the submitted form — including forced ones — are size-guarded; -// substituted sequences and planner-proven online idioms are not — long work -// on large tables is their purpose, and every brief step is still -// budget-bounded. +// sequence when one was substituted, otherwise the submitted form. A forced +// run bypasses the sequence executor's shape admission — the acknowledged +// override runs the submitted form as one blind bounded attempt, whatever +// its kind — so it goes through the optimistic executor directly, under the +// same brief budgets. Blind attempts of the submitted form — including +// forced ones — are size-guarded; substituted sequences and planner-proven +// online idioms are not — long work on large tables is their purpose, and +// every brief step is still budget-bounded. func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.Pool, st statement.Statement, execSQL []string, plan planner.Plan, substituted, forced bool, logger *slog.Logger) error { @@ -187,7 +190,7 @@ func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.P pt, err := preflight.CheckTable(ctx, pool, st.Schema(), st.Table(), limit) var sizeErr *preflight.SizeError if errors.As(err, &sizeErr) { - return c.emit(out, sizeGuardVerdict(st, sizeErr)) + return c.emit(out, sizeGuardVerdict(st, sizeErr, forced)) } if err != nil { return err @@ -205,22 +208,22 @@ func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.P Validate: executor.ValidateBudget{LockTimeout: c.LockTimeout, Overall: c.ValidateTimeout}, } start := time.Now() - _, err = executor.RunSequence(ctx, pool, pt, execSQL, budget) + if forced { + err = executor.AttemptNative(ctx, pool, pt, st, budget.Brief) + } else { + _, err = executor.RunSequence(ctx, pool, pt, execSQL, budget) + } elapsed := time.Since(start) - var budgetErr *executor.BudgetError - if err != nil && !substituted && errors.As(err, &budgetErr) { - // The blind attempt of the submitted form exceeded a budget and was - // cancelled without committing — the Phase 1 refusal contract. A - // forced attempt is bounded by the same budgets: --force overrides - // routing, never the executor's protections. - logger.Debug("optimistic attempt cancelled", - "cause", budgetErr.Cause, "budget", budgetErr.Budget, "elapsed", elapsed) - return c.emit(out, budgetVerdict(st, budgetErr)) + if v, refused := execRefusal(st, err, substituted, forced, onlineIdiomPlan(plan)); refused { + logger.Debug("execution refused", + "reason", string(v.Reason), "cause", string(v.Cause), "elapsed", elapsed) + return c.emit(out, v) } if err != nil { - // A failed substituted step is an operational failure, not a - // refusal: the typed *SequenceStepError names the failed step and - // the committed prefix that remains. + // Everything else is an operational failure, not a refusal: the + // typed *SequenceStepError names the failed step and the committed + // prefix that remains, and an *InvalidIndexError carries the + // operator recovery guidance. return fmt.Errorf("run schema change on %s: %w", qualified(st), err) } logger.Debug("schema change committed", @@ -246,22 +249,70 @@ func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.P return c.emit(out, v) } -// sizeGuardApplies reports whether the size guard protects this run. It -// guards exactly the blind attempt of the submitted form: when the engine -// substituted the planner's safer sequence, or when the plan proved every -// operation an online idiom (CONCURRENTLY, NOT VALID, VALIDATE), long work -// on a large table is the pattern's purpose and the guard would refuse the -// very tables the pattern serves. -func sizeGuardApplies(p planner.Plan, substituted bool) bool { - if substituted { +// execRefusal maps an execution failure to its refusal verdict, when the +// failure belongs to the refusal contract rather than the operational-error +// exit: a static admission refusal (decided before anything executed), or a +// budget cancellation of a non-substituted attempt. An *InvalidIndexError +// is never a refusal, even when a budget cancellation is buried inside it — +// invalid-index debris is the one outcome that needs an operator, and its +// typed error carries the recovery guidance a budget verdict would conceal. +func execRefusal(st statement.Statement, err error, + substituted, forced, online bool) (verdict.Verdict, bool) { + if err == nil { + return verdict.Verdict{}, false + } + if isAdmissionRefusal(err) { + return admissionRefusalVerdict(st, err, forced), true + } + var invalidErr *executor.InvalidIndexError + if errors.As(err, &invalidErr) { + return verdict.Verdict{}, false + } + var budgetErr *executor.BudgetError + if !substituted && errors.As(err, &budgetErr) { + // The blind attempt of the submitted form exceeded a budget and was + // cancelled without committing — the Phase 1 refusal contract. A + // forced attempt is bounded by the same budgets: --force overrides + // routing, never the executor's protections. + return budgetVerdict(st, budgetErr, forced, online), true + } + return verdict.Verdict{}, false +} + +// isAdmissionRefusal reports whether err is one of the executor's static +// admission refusals: decided from the statement's shape before anything +// executes, so it maps to a refusal verdict, not an operational error. A +// *SequenceStepError wrapper means execution started, which is never an +// admission refusal. +func isAdmissionRefusal(err error) bool { + var stepErr *executor.SequenceStepError + if errors.As(err, &stepErr) { return false } + return errors.Is(err, executor.ErrUnsupportedSequenceStep) || + errors.Is(err, executor.ErrUnnamedIndex) || + errors.Is(err, executor.ErrIfNotExistsUnsupported) +} + +// onlineIdiomPlan reports whether the plan proved every operation an online +// idiom (CONCURRENTLY, NOT VALID, VALIDATE): the submitted form already is +// the safe pattern, and running long on a large table is its purpose. +func onlineIdiomPlan(p planner.Plan) bool { for _, d := range p.Decisions { if d.Reason != planner.ReasonOnlineIdiom { - return true + return false } } - return false + return true +} + +// sizeGuardApplies reports whether the size guard protects this run. It +// guards exactly the blind attempt of the submitted form: when the engine +// substituted the planner's safer sequence, or when the plan proved every +// operation an online idiom, long work on a large table is the pattern's +// purpose and the guard would refuse the very tables the pattern serves. +func sizeGuardApplies(p planner.Plan, substituted bool) bool { + return !substituted && !onlineIdiomPlan(p) } // emit prints the verdict in the selected format and returns ErrRefused for @@ -376,13 +427,16 @@ func routeRefusalVerdict(st statement.Statement, rs router.Statement) verdict.Ve } // sizeGuardVerdict is the refusal for tables above the size threshold, where -// even a budget-bounded attempt would visibly stall the table. -func sizeGuardVerdict(st statement.Statement, sizeErr *preflight.SizeError) verdict.Verdict { +// even a budget-bounded attempt would visibly stall the table. A refused +// forced attempt still records the override: the operator asked for the +// submitted form and the guard said no. +func sizeGuardVerdict(st statement.Statement, sizeErr *preflight.SizeError, forced bool) verdict.Verdict { return verdict.Verdict{ Outcome: verdict.OutcomeRefused, Reason: verdict.ReasonTableTooLarge, Statement: st.SQL(), Table: qualified(st), + Forced: forced, Detail: fmt.Sprintf("table is %d bytes on disk (heap, indexes, and TOAST), above the %d-byte "+ "--max-table-size threshold. pg-sprite cannot yet prove this change is instant on a table this "+ "size; if it requires a rewrite, a cancelled attempt is not a free probe — it would hold "+ @@ -391,26 +445,53 @@ func sizeGuardVerdict(st statement.Statement, sizeErr *preflight.SizeError) verd } } +// admissionRefusalVerdict is the refusal for a statement the gate admits but +// the executor's static admission refuses before anything executes: an +// unnamed index build, IF NOT EXISTS on a concurrent build, or a substituted +// step shape the sequence executor does not drive yet (DETACH PARTITION +// CONCURRENTLY). The typed error carries the explanation. +func admissionRefusalVerdict(st statement.Statement, err error, forced bool) verdict.Verdict { + return verdict.Verdict{ + Outcome: verdict.OutcomeRefused, + Reason: verdict.ReasonUnsupportedStatement, + Statement: st.SQL(), + Table: qualified(st), + Forced: forced, + Detail: fmt.Sprintf("the engine cannot run this statement safely: %v; nothing was executed", err), + } +} + // budgetVerdict is the refusal for an attempt that exceeded its lock or -// statement budget and was cancelled without executing. -func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError) verdict.Verdict { +// statement budget and was cancelled without executing. online tailors the +// statement-budget advice: a submitted form the plan proved an online idiom +// (a concurrent build, a lone VALIDATE) needs a larger budget, not a +// different strategy, while a blind attempt that ran past its budget is +// doing rewrite work. A refused forced attempt still records the override. +func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError, forced, online bool) verdict.Verdict { v := verdict.Verdict{ Outcome: verdict.OutcomeRefused, Reason: verdict.ReasonBudgetExceeded, Statement: st.SQL(), Table: qualified(st), + Forced: forced, } switch budgetErr.Cause { case executor.CauseLock: v.Cause = verdict.CauseLockBudget v.Detail = fmt.Sprintf("the lock was not granted within the %s lock budget: the table is too "+ - "contended for a blind attempt; nothing was executed", budgetErr.Budget) + "contended right now; nothing was executed", budgetErr.Budget) case executor.CauseStatement: v.Cause = verdict.CauseStatementBudget - v.Detail = fmt.Sprintf("cancelled after the %s statement budget: the change does real rewrite work, "+ - "not an in-place catalog change, and needs a copy-and-swap rewrite that pg-sprite does not perform yet. "+ - "If it adds a constraint, ADD CONSTRAINT ... NOT VALID followed by VALIDATE CONSTRAINT avoids the long lock", - budgetErr.Budget) + if online { + v.Detail = fmt.Sprintf("cancelled after the %s budget: the statement already is the safe online "+ + "idiom — the work needs more time, not a different strategy; retry with a larger budget for "+ + "this step class", budgetErr.Budget) + } else { + v.Detail = fmt.Sprintf("cancelled after the %s statement budget: the change does real rewrite work, "+ + "not an in-place catalog change, and needs a copy-and-swap rewrite that pg-sprite does not perform yet. "+ + "If it adds a constraint, ADD CONSTRAINT ... NOT VALID followed by VALIDATE CONSTRAINT avoids the long lock", + budgetErr.Budget) + } default: v.Detail = budgetErr.Error() } diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index 1e094de..6244b84 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -458,6 +458,7 @@ func TestMigrateForceKeepsSizeGuard(t *testing.T) { var v verdict.Verdict require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) assert.Equal(t, verdict.ReasonTableTooLarge, v.Reason) + assert.True(t, v.Forced, "a refused forced attempt must still record the override machine-readably") var typ string require.NoError(t, pool.QueryRow(t.Context(), @@ -466,6 +467,193 @@ func TestMigrateForceKeepsSizeGuard(t *testing.T) { assert.Equal(t, "integer", typ, "the guarded change must not have executed") } +// The flagship force case from the design docs: a blocking CREATE INDEX, +// forced past its concurrent-build substitution, runs as-is as a blind +// bounded attempt and commits. +func TestMigrateForceRunsBlockingCreateIndex(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("CREATE INDEX t_forced_idx ON %s.t (c)", schema)) + cmd.Force = schema + ".t" + cmd.JSON = true + var out strings.Builder + require.NoError(t, cmd.run(t.Context(), &out)) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeExecuted, v.Outcome) + assert.True(t, v.Forced, "the verdict must record the override") + assert.Empty(t, v.ExecutedSQL, "the submitted form ran as-is; no substitution to report") + + var valid bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT i.indisvalid FROM pg_index i + WHERE i.indexrelid = ($1 || '.t_forced_idx')::regclass`, schema).Scan(&valid)) + assert.True(t, valid, "the forced blocking build must have committed a valid index") +} + +// A forced blind attempt that runs past its statement budget is cancelled +// and refused with the typed cause — the statement-budget refusal contract +// end to end — and the refusal still records the override. +func TestMigrateForcedRewriteRefusedByStatementBudget(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + _, err = pool.Exec(t.Context(), fmt.Sprintf( + "INSERT INTO %s.t SELECT g, repeat('x', 100) FROM generate_series(1, 300000) g", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) + cmd.Force = schema + ".t" + cmd.StatementTimeout = 50 * time.Millisecond + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) + assert.Equal(t, verdict.CauseStatementBudget, v.Cause) + assert.True(t, v.Forced, "a refused forced attempt must still record the override machine-readably") + + var typ string + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT data_type FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 't' AND column_name = 'id'`, schema).Scan(&typ)) + assert.Equal(t, "integer", typ, "the cancelled attempt must not change the schema") +} + +// A planner refusal (no known safe path) is not forceable: --force with a +// valid acknowledgement still ends in the refusal verdict and nothing +// executes. +func TestMigrateForceIgnoredForRefusedRoute(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, room int)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf( + "ALTER TABLE %s.t ADD CONSTRAINT ex EXCLUDE USING gist (room WITH =)", schema)) + cmd.Force = schema + ".t" + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + assert.Equal(t, verdict.ReasonUnsupportedStatement, v.Reason) + assert.False(t, v.Forced, "nothing ran, and no override was honored") + + var n int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) FROM pg_constraint con + JOIN pg_class c ON c.oid = con.conrelid + JOIN pg_namespace ns ON ns.oid = c.relnamespace + WHERE ns.nspname = $1 AND c.relname = 't' AND con.conname = 'ex'`, schema).Scan(&n)) + assert.Zero(t, n, "the refused change must not execute") +} + +// Statements the gate admits but the executor's static admission refuses end +// in a typed refusal verdict, not a raw operational error — the "exactly one +// verdict" contract holds for every gate-admitted statement, and nothing +// executes. +func TestMigrateRefusesExecutorAdmissionStatically(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + + t.Run("unnamed index build", func(t *testing.T) { + schema := testutil.NewSchema(t, pool) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("CREATE INDEX ON %s.t (c)", schema)) + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + assert.Equal(t, verdict.ReasonUnsupportedStatement, v.Reason) + + var n int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) FROM pg_indexes WHERE schemaname = $1 AND tablename = 't' AND indexname <> 't_pkey'`, + schema).Scan(&n)) + assert.Zero(t, n, "the refused build must not create an index") + }) + + t.Run("if not exists index build", func(t *testing.T) { + schema := testutil.NewSchema(t, pool) + _, err := pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, c int)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("CREATE INDEX IF NOT EXISTS t_c_idx ON %s.t (c)", schema)) + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + assert.Equal(t, verdict.ReasonUnsupportedStatement, v.Reason) + + var n int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) FROM pg_indexes WHERE schemaname = $1 AND indexname = 't_c_idx'`, + schema).Scan(&n)) + assert.Zero(t, n, "the refused build must not create an index") + }) + + t.Run("detach partition", func(t *testing.T) { + schema := testutil.NewSchema(t, pool) + _, err := pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.p (id int PRIMARY KEY) PARTITION BY RANGE (id)", schema)) + require.NoError(t, err) + _, err = pool.Exec(t.Context(), fmt.Sprintf( + "CREATE TABLE %s.p1 PARTITION OF %s.p FOR VALUES FROM (0) TO (100)", schema, schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.p DETACH PARTITION p1", schema)) + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.ErrorIs(t, err, verdict.ErrRefused) + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeRefused, v.Outcome) + + var attached bool + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT EXISTS (SELECT 1 FROM pg_inherits + WHERE inhrelid = ($1 || '.p1')::regclass AND inhparent = ($1 || '.p')::regclass)`, + schema).Scan(&attached)) + assert.True(t, attached, "the refused detach must leave the partition attached") + }) +} + // The size guard protects blind attempts only: a substituted safer sequence // runs on a table above the threshold, because its long steps are online by // design and its brief steps are budget-bounded. diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index bd0df77..d42b727 100644 --- a/internal/cli/migrate_test.go +++ b/internal/cli/migrate_test.go @@ -1,6 +1,7 @@ package cli import ( + "fmt" "testing" "time" @@ -17,26 +18,106 @@ func TestBudgetVerdict(t *testing.T) { require.NoError(t, err) t.Run("lock budget", func(t *testing.T) { - v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseLock, Budget: 3 * time.Second}) + v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseLock, Budget: 3 * time.Second}, false, false) assert.Equal(t, verdict.OutcomeRefused, v.Outcome) assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) assert.Equal(t, verdict.CauseLockBudget, v.Cause) assert.Equal(t, "billing.invoices", v.Table) + assert.False(t, v.Forced) assert.NotEmpty(t, v.Detail) }) t.Run("statement budget", func(t *testing.T) { - v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseStatement, Budget: 30 * time.Second}) + v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseStatement, Budget: 30 * time.Second}, false, false) assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) assert.Equal(t, verdict.CauseStatementBudget, v.Cause) assert.NotEmpty(t, v.Detail) }) + t.Run("statement budget on the online idiom advises a larger budget", func(t *testing.T) { + blind := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute}, false, false) + online := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute}, false, true) + assert.Equal(t, verdict.CauseStatementBudget, online.Cause) + assert.NotEqual(t, blind.Detail, online.Detail, + "a cancelled online idiom needs a larger budget, not the copy-and-swap advice") + }) + + t.Run("a forced refusal records the override", func(t *testing.T) { + v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute}, true, false) + assert.True(t, v.Forced, "the machine-readable audit record must survive a refusal") + }) + t.Run("unknown cause falls back to the error text", func(t *testing.T) { budgetErr := &executor.BudgetError{Budget: time.Second} - v := budgetVerdict(st, budgetErr) + v := budgetVerdict(st, budgetErr, false, false) assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) assert.Equal(t, verdict.CauseNone, v.Cause) assert.Equal(t, budgetErr.Error(), v.Detail) }) } + +func TestExecRefusal(t *testing.T) { + st, err := statement.ParseOne("CREATE INDEX CONCURRENTLY i ON billing.invoices (customer_id)") + require.NoError(t, err) + + t.Run("nil error is not a refusal", func(t *testing.T) { + _, refused := execRefusal(st, nil, false, false, false) + assert.False(t, refused) + }) + + t.Run("a budget-cancelled attempt is a refusal", func(t *testing.T) { + budgetErr := &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute} + v, refused := execRefusal(st, fmt.Errorf("wrapped: %w", budgetErr), false, true, true) + require.True(t, refused) + assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) + assert.Equal(t, verdict.CauseStatementBudget, v.Cause) + assert.True(t, v.Forced) + }) + + t.Run("a substituted sequence's budget failure is operational", func(t *testing.T) { + budgetErr := &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute} + _, refused := execRefusal(st, fmt.Errorf("wrapped: %w", budgetErr), true, false, false) + assert.False(t, refused, "a failed substituted step is an operational failure with a committed prefix") + }) + + t.Run("invalid-index debris is never a budget refusal", func(t *testing.T) { + // The exact chain a budget-cancelled concurrent build that left an + // invalid index produces: the buried *BudgetError must not map to a + // budget verdict that conceals the operator-recovery outcome. + invalid := &executor.InvalidIndexError{ + Schema: "billing", + Index: "i", + Build: &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute}, + Cleanup: executor.ErrBuildLeftInvalidIndex, + } + stepErr := &executor.SequenceStepError{Step: 1, Total: 1, Err: invalid} + _, refused := execRefusal(st, stepErr, false, false, true) + assert.False(t, refused, "invalid-index debris needs an operator, not a budget verdict") + }) + + t.Run("static admission refusals map to a typed refusal verdict", func(t *testing.T) { + for name, admissionErr := range map[string]error{ + "unsupported step": fmt.Errorf("sequence step 1 of 1: blocking CREATE INDEX: %w", executor.ErrUnsupportedSequenceStep), + "unnamed index": fmt.Errorf("sequence step 1 of 1: %w", executor.ErrUnnamedIndex), + "if not exists": fmt.Errorf("sequence step 1 of 1: %w", executor.ErrIfNotExistsUnsupported), + } { + t.Run(name, func(t *testing.T) { + v, refused := execRefusal(st, admissionErr, true, false, false) + require.True(t, refused, "an admission refusal decided before execution is a refusal verdict") + assert.Equal(t, verdict.ReasonUnsupportedStatement, v.Reason) + assert.NotEmpty(t, v.Detail) + }) + } + }) + + t.Run("an admission sentinel inside a step failure stays operational", func(t *testing.T) { + stepErr := &executor.SequenceStepError{Step: 2, Total: 3, Err: executor.ErrUnnamedIndex} + _, refused := execRefusal(st, stepErr, true, false, false) + assert.False(t, refused, "a step failure means execution started; the committed prefix must surface") + }) + + t.Run("an operational server error is not a refusal", func(t *testing.T) { + _, refused := execRefusal(st, fmt.Errorf("connection reset"), false, false, false) + assert.False(t, refused) + }) +} From 0d7e49111d3783e4d72dbf4e19eb8d7cf3526cf3 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Thu, 13 Aug 2026 20:29:25 +1000 Subject: [PATCH 3/3] Emit failed verdicts with executor codes and committed prefix; add changelog Review follow-ups on the substitution PR: a mid-sequence failure was only a typed error on stderr, so automation could not distinguish nothing committed from a scaffold left behind. Execution failures now end in a third verdict outcome (failed, still exit 1) carrying executor.OutcomeCode, the failed step, and the committed prefix. The changelog calls out the intended outcome changes for automation callers. --- CHANGELOG.md | 41 +++++++++++++++++ docs/architecture.md | 2 +- docs/low-level-design.md | 7 ++- internal/cli/migrate.go | 54 +++++++++++++++++++++-- internal/cli/migrate_integration_test.go | 45 +++++++++++++++++++ internal/cli/migrate_test.go | 48 ++++++++++++++++++++ pkg/executor/sequence.go | 9 ++-- pkg/verdict/verdict.go | 53 +++++++++++++++++----- pkg/verdict/verdict_test.go | 56 ++++++++++++++++++++++++ 9 files changed, 296 insertions(+), 19 deletions(-) create mode 100644 CHANGELOG.md diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..13c26bb --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,41 @@ +# Changelog + +Notable changes to pg-sprite, with emphasis on anything that changes what +automation observes: exit codes, verdict fields, and outcome vocabulary. +The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). + +## [Unreleased] + +### Changed — observable outcomes for automation callers + +- **A plain (blocking) `CREATE INDEX` now succeeds instead of refusing.** + The engine substitutes `CREATE INDEX CONCURRENTLY` and drives it to a + verified valid index: exit 0, with the substitution disclosed in the + verdict's `executed_sql`. Previously this refused with exit 2 and reason + `index-statement`. Anything branching on the old refusal will now see + success for the same input. +- **Rewrite-requiring changes (e.g. `ALTER COLUMN ... TYPE`) refuse up + front with reason `backend-unavailable`** instead of running a blind + bounded attempt that budget-cancels. No lock acquisition is attempted; + the refusal is decided from the classification alone. Previously the same + input ended in reason `not-native-safe-budget-exceeded` after a cancelled + attempt. +- **Blocking `ALTER TABLE` forms with a safer native sequence (e.g. + `SET NOT NULL`) run as that sequence by default**, disclosed in + `executed_sql`. The submitted form can still be forced with + `--force `, which is audited and recorded in the verdict's + `forced` field. + +### Added + +- **A third verdict outcome, `failed`,** for execution failures (still exit + 1 — refusals remain exit 2). The verdict carries the executor's stable + outcome code in `code`, and for a mid-sequence failure the 1-based + `failed_step`, its `failed_step_sql`, and the committed prefix in + `executed_sql` — an empty prefix means nothing committed, so automation + can distinguish "nothing happened" from "partial state left behind" + without parsing stderr. +- `--force `: an audited override that runs the submitted + form as one bounded attempt, unlocking only the substitution-override, + rewrite-required, and backend-unavailable paths. Planner refusals stay + unforceable. diff --git a/docs/architecture.md b/docs/architecture.md index 6149a2f..6531c4c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -128,7 +128,7 @@ different levels of commitment: | `pkg/dbconn` | Pool with bounded session timeouts, retries, RDS/Aurora auto-TLS (embedded CA bundle), terminate-blockers; advisory-lock mutual exclusion lands here | exists | | `pkg/statement` | `go-pgquery` (Wasm `libpg_query`) parse boundary, typed per-operation descriptors, and advisory rewrites (never hand-parse SQL); migration-time shadow DDL + fingerprints are derived by `pkg/schemadiff` via scratch-DB execute-and-introspect | exists | | `pkg/preflight` | Precondition verification and refusals before any write | exists (Phase 1: table-size guard); grows through Phase 2 | -| `pkg/verdict` | Structured outcome contract (executed / refused + reason + safer idiom), rendering, exit codes | exists (Phase 1) | +| `pkg/verdict` | Structured outcome contract (executed / refused / failed + reason, stable executor code, and safer idiom), rendering, exit codes | exists (Phase 1) | | `pkg/schemadiff` | Execute-and-introspect desired state, introspect the live catalog, and produce an ordered declarative diff | exists | | `pkg/planner` | Classify typed operations and emit safer native SQL | exists | | `pkg/lint` | Offline lint findings with typed codes: unsupported operations are errors; blocking idioms, rewrites, and destructive drops are warnings | exists (Phase 2.5) | diff --git a/docs/low-level-design.md b/docs/low-level-design.md index 05b1f8d..066eb5a 100644 --- a/docs/low-level-design.md +++ b/docs/low-level-design.md @@ -807,8 +807,11 @@ qualification burden off the user), substitutes and executes classifier-produced sequences by default with the guarded `--force` escape hatch, and renders the typed outcomes. At the library seam, each executor outcome maps to a stable string code (`executor.OutcomeCode`), the same treatment `pkg/lint` gave its findings, so an orchestrator -embedding `pkg/executor` branches on one vocabulary; the CLI's verdict JSON does not carry -the executor codes yet. Remaining Phase 3 work, roughly in order: +embedding `pkg/executor` branches on one vocabulary; the CLI's verdict JSON carries the same +codes — an execution failure ends in a `failed` verdict (exit 1, distinct from the refusal +exit 2) with the code, the failed step, and the committed prefix in `executed_sql`, so +automation can distinguish nothing-committed from partial state left behind. Remaining +Phase 3 work, roughly in order: - bound lock acquisition with timeout and retry for the blocking idioms, - progress reporting (`pg_stat_progress_create_index` by the build's backend PID, which the diff --git a/internal/cli/migrate.go b/internal/cli/migrate.go index 009460f..fd36525 100644 --- a/internal/cli/migrate.go +++ b/internal/cli/migrate.go @@ -24,7 +24,9 @@ import ( // exactly as dry-run would, execute the routed SQL — the planner's safer // native sequence by default when the submitted form blocks — and end in // exactly one verdict. Refusal verdicts are printed to out and returned as -// verdict.ErrRefused so the entry point maps them to the refusal exit code. +// verdict.ErrRefused so the entry point maps them to the refusal exit code; +// an execution failure prints a failed verdict — the stable executor code +// plus the committed prefix — and still returns the operational error. // --dry-run diverts to the classify-and-route plan instead. func (c *MigrateCmd) run(ctx context.Context, out io.Writer) error { if c.DryRun { @@ -209,10 +211,11 @@ func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.P } retry := c.retryPolicy() start := time.Now() + var rep executor.SequenceReport if forced { err = executor.ExecuteNative(ctx, pool, pt, st, budget.Brief, retry) } else { - _, err = executor.RunSequence(ctx, pool, pt, execSQL, budget, retry) + rep, err = executor.RunSequence(ctx, pool, pt, execSQL, budget, retry) } elapsed := time.Since(start) if v, refused := execRefusal(st, err, substituted, forced, onlineIdiomPlan(plan)); refused { @@ -224,7 +227,16 @@ func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.P // Everything else is an operational failure, not a refusal: the // typed *SequenceStepError names the failed step and the committed // prefix that remains, and an *InvalidIndexError carries the - // operator recovery guidance. + // operator recovery guidance. The failed verdict is the error's + // machine-readable twin on stdout — the stable executor code, the + // failed step, and the committed prefix — while the error itself + // still returns, so the process exits 1, not the refusal code. + v := failureVerdict(st, err, rep, forced) + logger.Debug("execution failed", + "code", v.Code, "failed_step", v.FailedStep, "committed_steps", len(v.ExecutedSQL), "elapsed", elapsed) + if emitErr := c.emit(out, v); emitErr != nil { + return emitErr + } return fmt.Errorf("run schema change on %s: %w", qualified(st), err) } logger.Debug("schema change committed", @@ -250,6 +262,42 @@ func (c *MigrateCmd) execute(ctx context.Context, out io.Writer, pool *pgxpool.P return c.emit(out, v) } +// failureVerdict maps an operational execution failure to its failed +// verdict: the executor's stable outcome code, and for a mid-sequence +// failure the failed step and the committed prefix whose state remains. +// It is the machine-readable twin of the returned error — automation +// branches on Code and ExecutedSQL instead of parsing stderr prose. A +// single bounded attempt (the submitted form, forced or not) rolls back on +// failure, so it carries no step and an empty committed prefix. +func failureVerdict(st statement.Statement, err error, + rep executor.SequenceReport, forced bool) verdict.Verdict { + v := verdict.Verdict{ + Outcome: verdict.OutcomeFailed, + Code: string(executor.OutcomeCode(err)), + Statement: st.SQL(), + Table: qualified(st), + Forced: forced, + Detail: "execution failed; nothing committed — a started bounded attempt rolls back", + } + var stepErr *executor.SequenceStepError + if !errors.As(err, &stepErr) { + return v + } + v.FailedStep = stepErr.Step + v.FailedStepSQL = stepErr.SQL + for _, s := range rep.Steps { + v.ExecutedSQL = append(v.ExecutedSQL, s.SQL) + } + if len(v.ExecutedSQL) > 0 { + v.Detail = fmt.Sprintf("sequence step %d of %d failed; the %d committed steps' state remains — the planner sequence's partial-failure contract says how a retry resumes", + stepErr.Step, stepErr.Total, len(v.ExecutedSQL)) + } else { + v.Detail = fmt.Sprintf("sequence step %d of %d failed; no earlier steps had committed — Code names the outcome and any state the failed step itself left", + stepErr.Step, stepErr.Total) + } + return v +} + // execRefusal maps an execution failure to its refusal verdict, when the // failure belongs to the refusal contract rather than the operational-error // exit: a static admission refusal (decided before anything executed), or a diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index 6244b84..0738113 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -19,6 +19,7 @@ import ( "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/preflight" "github.com/block/pg-sprite/pkg/verdict" ) @@ -209,6 +210,50 @@ func TestMigrateSubstitutesSetNotNullSequence(t *testing.T) { assert.Zero(t, scaffolds, "the scaffold CHECK constraint must be dropped") } +// A substituted sequence that fails mid-way emits a failed verdict whose +// typed fields — the stable executor code, the failed step, and the +// committed prefix — let automation distinguish "nothing happened" from +// "partial state left behind" without parsing stderr prose. The run still +// returns an operational error, not a refusal. +func TestMigrateFailedSequenceEmitsFailedVerdict(t *testing.T) { + url := testutil.StartPostgres(t) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + defer pool.Close() + schema := testutil.NewSchema(t, pool) + _, err = pool.Exec(t.Context(), fmt.Sprintf("CREATE TABLE %s.t (id int PRIMARY KEY, v text)", schema)) + require.NoError(t, err) + // The NULL row makes the sequence's VALIDATE CONSTRAINT step fail after + // the scaffold CHECK ... NOT VALID step has already committed. + _, err = pool.Exec(t.Context(), fmt.Sprintf("INSERT INTO %s.t VALUES (1, NULL)", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN v SET NOT NULL", schema)) + cmd.JSON = true + var out strings.Builder + err = cmd.run(t.Context(), &out) + require.Error(t, err, "an execution failure is an operational error") + require.NotErrorIs(t, err, verdict.ErrRefused, "a failure is not a refusal") + + var v verdict.Verdict + require.NoError(t, json.Unmarshal([]byte(out.String()), &v)) + assert.Equal(t, verdict.OutcomeFailed, v.Outcome) + assert.Equal(t, string(executor.CodeExecutionFailed), v.Code) + assert.Equal(t, 2, v.FailedStep, "the VALIDATE CONSTRAINT step is the second of the four") + assert.Len(t, v.ExecutedSQL, 1, "exactly the scaffold step committed before the failure") + assert.False(t, v.Forced) + + // The verdict's committed prefix must describe real state: the scaffold + // CHECK constraint survives the failed run. + var scaffolds int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) 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 = 't' AND con.contype = 'c'`, schema).Scan(&scaffolds)) + assert.Equal(t, 1, scaffolds, "the committed scaffold constraint must remain, per the partial-failure contract") +} + // A blocking CREATE INDEX is substituted with its concurrent build and // driven to a verified valid index. func TestMigrateSubstitutesBlockingCreateIndex(t *testing.T) { diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index 48eea8a..ca56bb8 100644 --- a/internal/cli/migrate_test.go +++ b/internal/cli/migrate_test.go @@ -101,6 +101,54 @@ func TestBudgetVerdict(t *testing.T) { }) } +func TestFailureVerdict(t *testing.T) { + st, err := statement.ParseOne("ALTER TABLE billing.invoices ALTER COLUMN status SET NOT NULL") + require.NoError(t, err) + + t.Run("a mid-sequence failure discloses the step and the committed prefix", func(t *testing.T) { + stepErr := &executor.SequenceStepError{ + Step: 2, + Total: 4, + Kind: executor.StepValidateConstraint, + SQL: "ALTER TABLE billing.invoices VALIDATE CONSTRAINT c", + Err: fmt.Errorf("server error"), + } + rep := executor.SequenceReport{Steps: []executor.StepReport{ + {SQL: "ALTER TABLE billing.invoices ADD CONSTRAINT c CHECK (status IS NOT NULL) NOT VALID", Kind: executor.StepBrief}, + }} + v := failureVerdict(st, fmt.Errorf("wrapped: %w", stepErr), rep, false) + assert.Equal(t, verdict.OutcomeFailed, v.Outcome) + assert.Equal(t, string(executor.CodeExecutionFailed), v.Code) + assert.Equal(t, 2, v.FailedStep) + assert.Equal(t, stepErr.SQL, v.FailedStepSQL) + assert.Equal(t, []string{rep.Steps[0].SQL}, v.ExecutedSQL, + "the committed prefix is what distinguishes partial state from nothing happened") + assert.Equal(t, "billing.invoices", v.Table) + assert.False(t, v.Forced) + }) + + t.Run("the failed step's typed cause maps to its own stable code", func(t *testing.T) { + stepErr := &executor.SequenceStepError{ + Step: 2, Total: 4, Kind: executor.StepValidateConstraint, + SQL: "ALTER TABLE billing.invoices VALIDATE CONSTRAINT c", + Err: &executor.BudgetError{Cause: executor.CauseStatement, Budget: time.Minute}, + } + v := failureVerdict(st, stepErr, executor.SequenceReport{}, false) + assert.Equal(t, string(executor.CodeBudgetStatementExceeded), v.Code) + assert.Empty(t, v.ExecutedSQL, "an empty committed prefix means nothing committed") + }) + + t.Run("a non-sequence failure carries no step and an empty prefix", func(t *testing.T) { + v := failureVerdict(st, fmt.Errorf("server error"), executor.SequenceReport{}, true) + assert.Equal(t, verdict.OutcomeFailed, v.Outcome) + assert.Equal(t, string(executor.CodeExecutionFailed), v.Code) + assert.Zero(t, v.FailedStep) + assert.Empty(t, v.FailedStepSQL) + assert.Empty(t, v.ExecutedSQL) + assert.True(t, v.Forced, "the machine-readable audit record must survive a failure") + }) +} + func TestExecRefusal(t *testing.T) { st, err := statement.ParseOne("CREATE INDEX CONCURRENTLY i ON billing.invoices (customer_id)") require.NoError(t, err) diff --git a/pkg/executor/sequence.go b/pkg/executor/sequence.go index 2bb58de..6595732 100644 --- a/pkg/executor/sequence.go +++ b/pkg/executor/sequence.go @@ -148,8 +148,10 @@ type StepReport struct { Index *IndexBuildReport `json:"index,omitempty"` } -// SequenceReport is the record of a completed sequence run: one report per -// step, in execution order. It is returned only when every step committed. +// SequenceReport is the record of a sequence run: one report per committed +// step, in execution order. On success it covers every step; alongside a +// *SequenceStepError it covers exactly the committed prefix, so a caller +// can disclose what already happened. type SequenceReport struct { // Steps are the per-step reports, in execution order. Steps []StepReport `json:"steps"` @@ -203,7 +205,8 @@ type sequenceStep struct { // 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. +// documented partial-failure contracts, and the returned report covers +// exactly that prefix. // // Like the concurrent build — and unlike a blind optimistic attempt — no // size-guard proof is required beyond the preflight itself: long scans on diff --git a/pkg/verdict/verdict.go b/pkg/verdict/verdict.go index 1352196..3a9bca8 100644 --- a/pkg/verdict/verdict.go +++ b/pkg/verdict/verdict.go @@ -1,8 +1,10 @@ // Package verdict is the engine's structured outcome contract: every migrate -// invocation ends in exactly one verdict — executed natively, or refused with -// a typed reason and, where one exists, a safer native idiom. Refusals use a -// distinct exit code from operational errors. This type is the seam a future -// orchestrator adapter maps onto SchemaBot's ExecutionModeBlocked. +// invocation ends in exactly one verdict — executed natively, refused with +// a typed reason and, where one exists, a safer native idiom, or failed +// during execution with the executor's stable outcome code and a disclosure +// of what committed before the failure. Refusals use a distinct exit code +// from operational errors. This type is the seam a future orchestrator +// adapter maps onto SchemaBot's ExecutionModeBlocked. package verdict import ( @@ -24,13 +26,20 @@ var ErrRefused = errors.New("refused") // Outcome is what happened to the submitted change. type Outcome string -// The two outcomes a migrate run can end in. +// The outcomes a migrate run can end in. const ( // OutcomeExecuted means the change ran and committed natively within // its budgets. OutcomeExecuted Outcome = "executed-natively" // OutcomeRefused means the change was not executed; Reason says why. OutcomeRefused Outcome = "refused" + // OutcomeFailed means execution was attempted and failed: an + // operational error, not a refusal — the process still exits 1. Code + // carries the executor's stable outcome code, and for a mid-sequence + // failure FailedStep and ExecutedSQL disclose the failed step and the + // committed prefix whose state remains, so automation can distinguish + // "nothing happened" from "partial state left behind". + OutcomeFailed Outcome = "failed" ) // Reason is the typed cause of a refusal. Reasons are flat kebab-case @@ -87,6 +96,20 @@ type Verdict struct { // Cause narrows a budget refusal to the budget that fired; empty // otherwise. Cause Cause `json:"cause,omitempty"` + // Code is the executor's stable outcome code (executor.OutcomeCode) + // carried by a failed verdict — flat kebab-case, part of the executor's + // report contract. It stays a plain string here so this contract + // package does not depend on the executor. Empty unless Outcome is + // OutcomeFailed. + Code string `json:"code,omitempty"` + // FailedStep is the 1-based position of the sequence step that failed, + // matching the numbering the planner's partial-failure contracts use; + // zero when the failure was not a mid-sequence one (a single-statement + // attempt rolls back and commits nothing). + FailedStep int `json:"failed_step,omitempty"` + // FailedStepSQL is the failed step's statement — the step the planner's + // partial-failure contract says a retry resumes from. + FailedStepSQL string `json:"failed_step_sql,omitempty"` // Attempts is how many bounded attempts ran before a lock-budget // refusal, so automation can tell an exhausted bounded retry from a // single cancelled attempt; zero for every other verdict. @@ -101,10 +124,11 @@ type Verdict struct { // SaferIdiom is a native alternative to the refused statement, when one // exists (e.g. CREATE INDEX CONCURRENTLY, ADD CONSTRAINT ... NOT VALID). SaferIdiom string `json:"safer_idiom,omitempty"` - // ExecutedSQL is the ordered SQL the engine actually ran when it - // substituted the planner's safer native sequence for the submitted - // form. Empty when the submitted form ran as-is — a non-empty value is - // what tells automation a substitution happened. + // ExecutedSQL is the ordered SQL the engine actually ran and committed. + // On an executed verdict it is the substituted safer native sequence + // (empty when the submitted form ran as-is — a non-empty value is what + // tells automation a substitution happened). On a failed verdict it is + // the committed prefix that remains: empty means nothing committed. ExecutedSQL []string `json:"executed_sql,omitempty"` // Forced reports that --force overrode the engine's routing: the // submitted form ran as-is instead of a safer substitution or a @@ -130,6 +154,8 @@ func (v Verdict) String() string { b.WriteString("executed natively") case OutcomeRefused: fmt.Fprintf(&b, "refused (%s)", v.Reason) + case OutcomeFailed: + fmt.Fprintf(&b, "failed (%s)", v.Code) default: fmt.Fprintf(&b, "unknown outcome %q", string(v.Outcome)) } @@ -149,8 +175,15 @@ func (v Verdict) String() string { if v.Forced { b.WriteString("\n forced: the submitted form ran as-is (--force)") } + if v.FailedStep > 0 { + fmt.Fprintf(&b, "\n failed at: step %d: %s", v.FailedStep, v.FailedStepSQL) + } if len(v.ExecutedSQL) > 0 { - b.WriteString("\n executed as:") + if v.Outcome == OutcomeFailed { + b.WriteString("\n committed before the failure (their state remains):") + } else { + b.WriteString("\n executed as:") + } for i, sql := range v.ExecutedSQL { fmt.Fprintf(&b, "\n %d. %s", i+1, sql) } diff --git a/pkg/verdict/verdict_test.go b/pkg/verdict/verdict_test.go index c4cd9fc..062e1ef 100644 --- a/pkg/verdict/verdict_test.go +++ b/pkg/verdict/verdict_test.go @@ -26,6 +26,44 @@ func TestJSONRoundTrip(t *testing.T) { assert.Equal(t, v, got) } +func TestJSONRoundTripFailed(t *testing.T) { + v := Verdict{ + Outcome: OutcomeFailed, + Code: "budget-statement-exceeded", + Statement: "ALTER TABLE t ALTER COLUMN v SET NOT NULL", + Table: "t", + FailedStep: 2, + FailedStepSQL: "ALTER TABLE t VALIDATE CONSTRAINT c", + ExecutedSQL: []string{"ALTER TABLE t ADD CONSTRAINT c CHECK (v IS NOT NULL) NOT VALID"}, + Detail: "step 2 of 4 failed; the committed step's state remains", + } + s, err := v.JSON() + require.NoError(t, err) + + var got Verdict + require.NoError(t, json.Unmarshal([]byte(s), &got)) + assert.Equal(t, v, got) +} + +// The failed verdict's JSON keys are the machine contract automation reads; +// renaming a Go field must not silently rename a key. +func TestFailedJSONKeysArePinned(t *testing.T) { + s, err := Verdict{ + Outcome: OutcomeFailed, + Code: "execution-failed", + Statement: "ALTER TABLE t ALTER COLUMN v SET NOT NULL", + FailedStep: 2, + FailedStepSQL: "ALTER TABLE t VALIDATE CONSTRAINT c", + ExecutedSQL: []string{"ALTER TABLE t ADD CONSTRAINT c CHECK (v IS NOT NULL) NOT VALID"}, + }.JSON() + require.NoError(t, err) + for _, key := range []string{ + `"outcome": "failed"`, `"code"`, `"failed_step"`, `"failed_step_sql"`, `"executed_sql"`, + } { + assert.Contains(t, s, key) + } +} + func TestJSONOmitsEmptyOptionalFields(t *testing.T) { s, err := Verdict{Outcome: OutcomeExecuted, Statement: "ALTER TABLE t ADD COLUMN x int"}.JSON() require.NoError(t, err) @@ -33,6 +71,8 @@ func TestJSONOmitsEmptyOptionalFields(t *testing.T) { assert.NotContains(t, s, "table") assert.NotContains(t, s, "safer_idiom") assert.NotContains(t, s, "attempts") + assert.NotContains(t, s, "code") + assert.NotContains(t, s, "failed_step") } // Reason and Cause values are the machine contract automation switches on: @@ -73,6 +113,22 @@ func TestStringRefusedIncludesReasonAndIdiom(t *testing.T) { assert.Contains(t, s, "CREATE INDEX CONCURRENTLY") } +func TestStringFailedIncludesCodeStepAndCommittedPrefix(t *testing.T) { + s := Verdict{ + Outcome: OutcomeFailed, + Code: "execution-failed", + Statement: "ALTER TABLE t ALTER COLUMN v SET NOT NULL", + Table: "t", + FailedStep: 2, + FailedStepSQL: "ALTER TABLE t VALIDATE CONSTRAINT c", + ExecutedSQL: []string{"ALTER TABLE t ADD CONSTRAINT c CHECK (v IS NOT NULL) NOT VALID"}, + }.String() + assert.Contains(t, s, "failed (execution-failed)") + assert.Contains(t, s, "failed at: step 2: ALTER TABLE t VALIDATE CONSTRAINT c") + assert.Contains(t, s, "committed before the failure") + assert.NotContains(t, s, "executed as:", "a committed prefix is not a completed substitution") +} + func TestStringIncludesAttemptsWhenSet(t *testing.T) { v := Verdict{ Outcome: OutcomeRefused,