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/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..6531c4c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -128,14 +128,14 @@ 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) | | `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..066eb5a 100644 --- a/docs/low-level-design.md +++ b/docs/low-level-design.md @@ -337,35 +337,38 @@ 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 — 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. ### 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 +799,23 @@ 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. +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 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, -- 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 1b06c34..294c805 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -74,19 +74,36 @@ 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."` - LockAttempts int `help:"Maximum bounded attempts when native DDL exceeds lock_timeout; 1 disables retry." default:"3"` - LockBackoff time.Duration `help:"Initial exponential backoff between lock-timeout attempts." default:"100ms"` - LockBackoffMax time.Duration `help:"Maximum exponential backoff between lock-timeout attempts." default:"1s"` + 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"` + LockAttempts int `help:"Maximum bounded attempts when native DDL exceeds lock_timeout; 1 disables retry." default:"3"` + LockBackoff time.Duration `help:"Initial exponential backoff between lock-timeout attempts." default:"100ms"` + LockBackoffMax time.Duration `help:"Maximum exponential backoff between lock-timeout attempts." default:"1s"` + 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 21648e6..fd36525 100644 --- a/internal/cli/migrate.go +++ b/internal/cli/migrate.go @@ -5,20 +5,29 @@ 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; +// 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 { return c.runDryRun(ctx, out) @@ -39,39 +48,320 @@ 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. 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 { + 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)) + return c.emit(out, sizeGuardVerdict(st, sizeErr, forced)) } if err != nil { 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}, + } retry := c.retryPolicy() start := time.Now() - err = executor.ExecuteNative(ctx, pool, pt, st, budget, retry) + var rep executor.SequenceReport + if forced { + err = executor.ExecuteNative(ctx, pool, pt, st, budget.Brief, retry) + } else { + rep, err = executor.RunSequence(ctx, pool, pt, execSQL, budget, retry) + } elapsed := time.Since(start) - var budgetErr *executor.BudgetError - if errors.As(err, &budgetErr) { - logger.Debug("optimistic attempt cancelled", - "cause", budgetErr.Cause, "budget", budgetErr.Budget, "attempts", budgetErr.Attempts, "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), "attempts", v.Attempts, "elapsed", elapsed) + return c.emit(out, v) } if err != nil { - return err + // 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. 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("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) +} + +// 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 +// 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 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) } func (c *MigrateCmd) retryPolicy() executor.RetryPolicy { @@ -103,15 +393,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: @@ -120,21 +413,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: @@ -144,14 +436,66 @@ 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 { +// 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 "+ @@ -160,14 +504,35 @@ 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: @@ -180,13 +545,19 @@ func budgetVerdict(st statement.Statement, budgetErr *executor.BudgetError) verd break } 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 a8889d2..0738113 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -19,6 +19,8 @@ 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" ) @@ -30,8 +32,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 +94,459 @@ 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) + 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, 1000) g", schema)) + require.NoError(t, err) + + cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", 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.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 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, 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 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) { + 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) + 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 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) @@ -105,6 +559,7 @@ func TestMigrateRefusesRewriteWithBudgetVerdict(t *testing.T) { 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 @@ -116,17 +571,159 @@ func TestMigrateRefusesRewriteWithBudgetVerdict(t *testing.T) { assert.Equal(t, verdict.OutcomeRefused, v.Outcome) assert.Equal(t, verdict.ReasonBudgetExceeded, v.Reason) assert.Equal(t, verdict.CauseStatementBudget, v.Cause) - assert.Equal(t, schema+".t", v.Table) + 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") - var count int +} + +// 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(), - fmt.Sprintf("SELECT count(*) FROM %s.t", schema)).Scan(&count)) - assert.Equal(t, 300000, count, "the cancelled attempt must not change the data") + `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. +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 +769,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/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index 6d7fac9..ca56bb8 100644 --- a/internal/cli/migrate_test.go +++ b/internal/cli/migrate_test.go @@ -1,6 +1,7 @@ package cli import ( + "fmt" "testing" "time" @@ -61,27 +62,155 @@ 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, Attempts: 3}) + v := budgetVerdict(st, &executor.BudgetError{Cause: executor.CauseLock, Budget: 3 * time.Second, Attempts: 3}, 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, 3, v.Attempts, "the exhausted attempt count must reach the verdict") 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 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) + + 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) + }) +} 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 ae5c6a1..6595732 100644 --- a/pkg/executor/sequence.go +++ b/pkg/executor/sequence.go @@ -137,22 +137,24 @@ 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. +// 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 + Steps []StepReport `json:"steps"` } // SequenceStepError reports that a step failed and the run stopped there. @@ -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/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 75306df..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 @@ -51,6 +60,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 @@ -78,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. @@ -92,6 +124,17 @@ 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 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 + // 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. @@ -111,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)) } @@ -127,5 +172,21 @@ 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 v.FailedStep > 0 { + fmt.Fprintf(&b, "\n failed at: step %d: %s", v.FailedStep, v.FailedStepSQL) + } + if len(v.ExecutedSQL) > 0 { + 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) + } + } return b.String() } 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,