From 2a9d5dee0a027def0c7045ad5bc6507e11e7649d Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Tue, 27 Oct 2026 13:34:15 +1100 Subject: [PATCH 1/3] migrate: require --quiesced to name the table it takes offline A quiesced run refuses every write to the table until the swap, so the declaration becomes a typed SCHEMA.TABLE acknowledgement like --restart, checked before anything executes and audited before the fence goes on. --- docs/cli-output-examples.md | 3 +- docs/limitations.md | 4 +-- docs/schemabot-integration.md | 8 +++-- integration/supabase/copy_swap_route_test.go | 13 ++++--- integration/supabase/schema_test.go | 2 +- internal/cli/cli.go | 2 +- internal/cli/migrate_integration_test.go | 4 +-- internal/cli/migrate_test.go | 26 +++++++++++--- internal/cli/restart_integration_test.go | 6 ++-- internal/cli/signal_integration_test.go | 2 +- .../front_door_integration_test.go | 4 ++- pkg/migrate/copy_swap.go | 22 ++++++++---- .../copy_swap_connection_integration_test.go | 14 ++++---- pkg/migrate/copy_swap_integration_test.go | 7 ++-- pkg/migrate/copy_swap_quiesced.go | 32 +++++++++++++++++ .../copy_swap_quiesced_integration_test.go | 34 +++++++++++++++++++ .../desired_copy_swap_integration_test.go | 14 ++++---- pkg/migrate/migrate.go | 5 +++ pkg/migrate/rename_index.go | 4 +++ pkg/migrate/run_integration_test.go | 2 +- 20 files changed, 160 insertions(+), 48 deletions(-) create mode 100644 pkg/migrate/copy_swap_quiesced.go create mode 100644 pkg/migrate/copy_swap_quiesced_integration_test.go diff --git a/docs/cli-output-examples.md b/docs/cli-output-examples.md index 2f744cb5..7a16cf52 100644 --- a/docs/cli-output-examples.md +++ b/docs/cli-output-examples.md @@ -374,7 +374,8 @@ $ pg-sprite migrate --alter 'ALTER TABLE users ALTER COLUMN id TYPE text' --dry- Without `--dry-run` the run needs `--free-disk` (the preflight refuses with `copy-and-swap-refused` / cause `copy-and-swap-disk-headroom` when the shadow and its indexes would not fit the declared space) and either logical -decoding on the server or `--quiesced`. The verdict's outcome is distinct +decoding on the server or `--quiesced SCHEMA.TABLE`, which names the table it +takes offline for writes until the swap. The verdict's outcome is distinct from `executed-natively`, and `detail` says what the verification pass covered and what became of the retained source: diff --git a/docs/limitations.md b/docs/limitations.md index 445e6e55..3dba2994 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -23,7 +23,7 @@ not escape hatches: | Retyping or dropping a column a view reads | PostgreSQL refuses `ALTER COLUMN TYPE` and `DROP COLUMN` on a column a view or materialized view reads, even for a type change it would relabel in place. Without `ONLY`, a type change reaches the column on every partition and inheritance child too, and a drop reaches every partition and each inheritance child that does not declare the column and inherits it only from tables the drop also reaches; a child that declares the column or also inherits it from a table outside the drop keeps its copy, while one that inherits it along two paths from the dropped table loses it. The planner reads the views on every copy the change reaches from the catalog and refuses the change with `dependent-view`, naming each view in the plan and in the verdict's `dependent_views`, instead of letting the statement fail at the server. `DROP COLUMN … CASCADE`, which would drop the views with the column, is refused with `cascading-drop`. pg-sprite does not drop and recreate the views for you. Drop or redefine them first, then re-plan. | | Dropping with `CASCADE` | `CASCADE` on `DROP COLUMN`, `DROP CONSTRAINT`, or `DROP INDEX` also drops whatever depends on the target: a view that reads the column, or another table's foreign key on the constraint. Those objects are not in the statement, so the plan cannot show them and the table's lock does not cover them. The planner refuses every such drop with `cascading-drop`. Drop each dependent object as its own change, then resubmit without `CASCADE`; PostgreSQL then refuses the drop while anything still depends on the target. | | `ADD COLUMN` on a hot standby, or by a role without `TEMPORARY` | Whether adding a column rewrites the table is the server's answer, read by adding the column to an empty temporary table. A server that accepts no writes, or a role whose `TEMPORARY` on the database was revoked, cannot create that table, so the plan fails with `ErrAddColumnProbeUnavailable` before anything runs. Plan against the primary, or grant `TEMPORARY` on the database to the engine's role. | -| Copy-and-swap | The copy-and-swap backend carries a general `ALTER COLUMN TYPE` (with or without `USING`): the changed shape is built as a shadow table, filled in chunks while writes stream through a logical replication slot, verified, and swapped in under the source's name. The run needs `--free-disk` (the preflight refuses `copy-and-swap-disk-headroom` when the shadow and its indexes would not fit) and either logical decoding (`wal_level = logical`, a free replication slot) or `--quiesced`; its preflight refuses table shapes it cannot run against — no primary key or usable replica identity, unlogged, partitioned, `FORCE ROW LEVEL SECURITY`, foreign keys in either direction, triggers, dependent views, membership in a publication, or a subscription on the database — each as a typed `copy-and-swap-refused` cause ([refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause)). The other rewrite-class operations — volatile-default `ADD COLUMN`, `ADD COLUMN` of a domain with a `CHECK` or `NOT NULL`, `GENERATED … STORED`, `SET TABLESPACE` — route to the backend but it does not carry them yet; they refuse `backend-unavailable`. pg-sprite never falls through to a blocking rewrite. A run that stops short of the swap keeps its hold on the source until it is resumed or abandoned — the write fence of a `--quiesced` run (every write is refused `55006`), or the replication slot of a live run, which pins WAL on the server — and the `failed` verdict's detail says which applies: a stop the next run clears (a lost connection, a cancelled run) resumes from the checkpoint row when the same statement is run again; a stop the data causes meets the same rows again, so it is not resumed: when the server rejects a row under the new type or a constraint (SQLSTATE class `22` or `23`) the run abandons itself — drops the shadow, releases the fence or slot, deletes the checkpoint row — and the verdict is `failed` with `code: data-fault`, so correct the data or the statement and run it again from the beginning; when the shadow and source diverge at verification (or the abandon itself failed), the row and the hold stay and the run has to be started over: correct the data or the statement, then run it with `--restart SCHEMA.TABLE` (`CopySwapOptions.Restart` as a library caller), which drops the shadow, releases the fence or slot, deletes the checkpoint row, and runs the change fresh under the run's own lock ([stopping, resuming, and starting over](schemabot-integration.md#stopping-resuming-and-starting-over)). | +| Copy-and-swap | The copy-and-swap backend carries a general `ALTER COLUMN TYPE` (with or without `USING`): the changed shape is built as a shadow table, filled in chunks while writes stream through a logical replication slot, verified, and swapped in under the source's name. The run needs `--free-disk` (the preflight refuses `copy-and-swap-disk-headroom` when the shadow and its indexes would not fit) and either logical decoding (`wal_level = logical`, a free replication slot) or `--quiesced SCHEMA.TABLE`, which names the table it takes offline for writes until the swap; its preflight refuses table shapes it cannot run against — no primary key or usable replica identity, unlogged, partitioned, `FORCE ROW LEVEL SECURITY`, foreign keys in either direction, triggers, dependent views, membership in a publication, or a subscription on the database — each as a typed `copy-and-swap-refused` cause ([refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause)). The other rewrite-class operations — volatile-default `ADD COLUMN`, `ADD COLUMN` of a domain with a `CHECK` or `NOT NULL`, `GENERATED … STORED`, `SET TABLESPACE` — route to the backend but it does not carry them yet; they refuse `backend-unavailable`. pg-sprite never falls through to a blocking rewrite. A run that stops short of the swap keeps its hold on the source until it is resumed or abandoned — the write fence of a quiesced run (every write is refused `55006`), or the replication slot of a live run, which pins WAL on the server — and the `failed` verdict's detail says which applies: a stop the next run clears (a lost connection, a cancelled run) resumes from the checkpoint row when the same statement is run again; a stop the data causes meets the same rows again, so it is not resumed: when the server rejects a row under the new type or a constraint (SQLSTATE class `22` or `23`) the run abandons itself — drops the shadow, releases the fence or slot, deletes the checkpoint row — and the verdict is `failed` with `code: data-fault`, so correct the data or the statement and run it again from the beginning; when the shadow and source diverge at verification (or the abandon itself failed), the row and the hold stay and the run has to be started over: correct the data or the statement, then run it with `--restart SCHEMA.TABLE` (`CopySwapOptions.Restart` as a library caller), which drops the shadow, releases the fence or slot, deletes the checkpoint row, and runs the change fresh under the run's own lock ([stopping, resuming, and starting over](schemabot-integration.md#stopping-resuming-and-starting-over)). | | `ALTER COLUMN TYPE ... USING ` the copy cannot reproduce | The copy-and-swap route (above) runs the retype, but the shadow builder admits a `USING` expression only when it can evaluate it again, identically, in the chunk copy, the verifier's recopy, and every change the catch-up flushes. Three rules, each refused as `shadow-statement-conversion` before any shadow is committed: the expression must be **immutable** by the server's own classification (no `now()`, `random()`, `current_setting()`, session-dependent casts such as `text` to `timestamptz`, or functions declared `STABLE` or `VOLATILE`; checked by trying it as a stored generated column's expression on the empty shadow inside the build's transaction, then rolling that attempt back), it must **name only the column it retypes**, bare and unqualified (no neighbouring column, no `t.col`, no whole-row `t.*`), and it must not retype the **primary key**, whose values every chunk range, watermark, and resume clear are expressed in. The expression is also refused when the deparser cannot prove it rendered the text faithfully. Rewrite the expression in terms of the retyped column alone and of `IMMUTABLE` functions, or retype without `USING` so the server's assignment cast converts the value; the refusal names the column, the expression, and that rewrite. The spellings most often refused, and what to write instead, are in [`USING` expressions the copy reproduces](#using-expressions-the-copy-reproduces) below. | | A schema change to the source table while a copy-and-swap runs | Unsupported: logical decoding does not stream DDL, and the copy does not lock it out — a chunk holds `ACCESS SHARE` on the source only for its own transaction, so `ALTER TABLE` on the source completes between chunks. The copy selects the source by column name, so what happens depends on whether every copied name still resolves. A copied column renamed or dropped stops the copy at its next chunk with the server's `undefined_column` error (SQLSTATE `42703`), which is not a typed refusal; that chunk stays in flight and nothing from it reaches the shadow. A change that leaves every copied name resolving to a column of a compatible type — two columns swapping names, a column renamed and re-added under the old name, a widened type — is not detected by the copy, which completes, and after a swap or a re-add the rows copied since hold another column's values or the new column's default. Once the replication stream has decoded a change to the table, a later relation description that differs stops it with `ErrSourceShapeChanged`. None of these reaches the swap: the verifier compares source and shadow rows by column name, and the cutover gate re-fingerprints the source and refuses any changed column as `cutover-schema-drift` ([refusal-classes.md](refusal-classes.md#shadow-operation-refusals-keyed-on-refusalcause)), after which the shadow is dropped and the change re-planned. | | A concurrent index build whose name is occupied by an invalid index, or whose table carries quarantined debris | The build refuses before running and never drops the occupant: an invalid entry is either abandoned debris or someone's build still in progress, and a drop by name cannot tell them apart. The refusal is typed by proof (`invalid-index-abandoned`, `invalid-index-build-in-flight`, `invalid-index-other-table`, `invalid-index-not-droppable`, `invalid-index-builder-unobservable`), and a table carrying an entry a previous recovery quarantined (`pgsprite_abandoned_`) but did not drop is refused the same way, so debris never goes quiet. The proven removal — `executor.RebuildAbandonedIndex`, or `executor.DropAbandonedIndex` to remove the leftover without running the build; both library-only today, no CLI verb — takes a bounded `SHARE UPDATE EXCLUSIVE` lock on the table (which any running concurrent index command holds), re-verifies the entry by OID, renames it to an identity-derived name and drops that, all within one shared budget; it refuses a visible in-flight build, another table's entry, or an entry the server will not drop concurrently (a partitioned table's index, an index partition, a constraint's index), and reports a lock budget rather than wait behind a build it cannot see. From the CLI, the [runbook](invalid-index-recovery.md) applies. | @@ -121,7 +121,7 @@ composition of the model boundaries above with those gates. At a glance: | Add a constraint (`UNIQUE`, `CHECK`); `SET NOT NULL` | Converges via the safer online sequence; not size-guarded either. | | Relax a `NOT NULL` | Refused, whole plan: dropping `NOT NULL` discards the same guarantee its constraint form would, so it is destructive. Run it deliberately through the imperative front door — it executes natively there — then rerun. | | Change an index or constraint definition | Refused, whole plan — the drop-and-recreate row above. | -| Narrow a column type (`varchar(255)` → `varchar(50)`) | Converges through copy-and-swap: the plan's disposition is `execute` with the statement routed to the `copy-and-swap` backend, and desired-state execution runs it with the same preflight gates and flags as `migrate --alter` (the Copy-and-swap row above). The route's preflight is read at admission, before any statement runs, so a table or server the route refuses — no `--free-disk`, no logical decoding without `--quiesced`, a trigger, a dependent view — refuses the whole plan with nothing executed, native statements included. The dry run and `diff` do not read the preflight — the plan report says `execute` on the route alone, and whether the route can run here, now, with these flags, is answered when the plan is executed. A plan that retypes two columns runs two copy-and-swap passes on the table, one per statement; the second replaces the done row the first leaves behind. | +| Narrow a column type (`varchar(255)` → `varchar(50)`) | Converges through copy-and-swap: the plan's disposition is `execute` with the statement routed to the `copy-and-swap` backend, and desired-state execution runs it with the same preflight gates and flags as `migrate --alter` (the Copy-and-swap row above). The route's preflight is read at admission, before any statement runs, so a table or server the route refuses — no `--free-disk`, no logical decoding without `--quiesced SCHEMA.TABLE`, a trigger, a dependent view — refuses the whole plan with nothing executed, native statements included. The dry run and `diff` do not read the preflight — the plan report says `execute` on the route alone, and whether the route can run here, now, with these flags, is answered when the plan is executed. A plan that retypes two columns runs two copy-and-swap passes on the table, one per statement; the second replaces the done row the first leaves behind. | | A bounded-attempt edit on a table above the size threshold | Refused at that statement (`not-native-safe-table-too-large`): with the default 1 GiB threshold, adding a column to a larger table refuses until the threshold is raised. | A destructive refusal is all-or-nothing: one destructive statement refuses diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 8a86c97b..08971a54 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -459,7 +459,11 @@ finds no lock to confirm. The check reads `pg_control_system()`, which PostgreSQ `PUBLIC`; a server that withholds it stops the route with an ordinary error), the free space on the table's volume for the preflight's disk-headroom check, whether to keep the retained source under its `_old` name, and whether the table is quiesced for the run — the fenced path that needs -no logical decoding — or live. Three further fields let the adapter size and watch the run: +no logical decoding but refuses every write to the table with `55006` until the swap — or +live. `Quiesced` is a typed acknowledgement, not a switch: it must name the resolved target +as `SCHEMA.TABLE`, the same way `Restart` does, or `Run` returns an error before anything +executes, and an acknowledged run records a warn-level audit naming the table before the +fence goes on. Three further fields let the adapter size and watch the run: `Copier` (`copier.Options`) sizes the copy — workers, chunk sizes, the chunk time they are throttled toward — with zero values taking the copier's defaults; `CheckpointInterval` is how often a copy in progress saves its landed watermark (zero takes the Runner's default); and `OnCheckpoint` is @@ -654,7 +658,7 @@ CLI and through `migrate.Run`, plus one the Runner performs on its own; an orche | Operator need | What to do | What pg-sprite does | | --- | --- | --- | -| **Start** a change | `pg-sprite migrate --url … 'ALTER TABLE s.t …'` (with `--free-disk`, and `--quiesced` when the table takes no writes) | Routes the statement; on the copy-and-swap backend it takes the table's lock session, writes the checkpoint row, builds and fills the shadow, verifies, swaps, and prints one verdict | +| **Start** a change | `pg-sprite migrate --url … 'ALTER TABLE s.t …'` (with `--free-disk`, and `--quiesced SCHEMA.TABLE` to take the table offline for writes instead of streaming them) | Routes the statement; on the copy-and-swap backend it takes the table's lock session, writes the checkpoint row, builds and fills the shadow, verifies, swaps, and prints one verdict | | **Stop** a running change | Send `SIGINT` (Ctrl-C) or `SIGTERM` once | The run's context is cancelled and the run unwinds through its own stop path: a copy in flight saves the watermark its committed chunks reached, the table lock session is released, the row stays at its last save with the run's hold on the source — the write fence of a quiesced run, or the replication slot of a live run, which retains WAL until the run is finished or discarded — and the verdict is `failed` with `code: cancelled-by-caller` (exit 1). A second `SIGINT`/`SIGTERM` ends the process at once with exit 130, before the stop path completes; the row then records whatever the last save landed, and the lock session is released by the server when its connection drops | | **Resume** a stopped change | Run the same statement again | The checkpoint row says where to continue: a stopped copy continues from its watermark, a run past its copy verifies again, a `cutover` row is answered by the catalog (a committed swap finishes from the old-table drop; a rolled-back swap goes back through verify and gate). Nothing is copied twice. A different statement on the same table is refused (`resume-row-mismatch`) | | **Start over** — discard a stopped change and run it from the beginning | Run the same statement with `--restart SCHEMA.TABLE`, where the value names the resolved target exactly | Under the run's own table lock session, discards what the stopped run left in the order a later restart could meet idempotently — the shadow, then the hold on the source (`DropWriteFence` for a quiesced row, `decode.DropSlot` for a live row), then the checkpoint row — and runs the change fresh. A leftover shadow that no row records is discarded the same way. The decision is written to the audit log before anything is discarded | diff --git a/integration/supabase/copy_swap_route_test.go b/integration/supabase/copy_swap_route_test.go index 318970ee..dcdfb80a 100644 --- a/integration/supabase/copy_swap_route_test.go +++ b/integration/supabase/copy_swap_route_test.go @@ -19,11 +19,14 @@ import ( // workers, or replication stream open, which is the fail-closed shape these // probes rely on to leave the fixture untouched. The only session opened // from the connection is the read-only one that proves it reaches the -// pool's database. Quiesced keeps the probe independent of the fixture -// role's replication access. -func copySwapOptions() migrate.Options { +// pool's database. Quiesced, acknowledged for st's table, keeps the probe +// independent of the fixture role's replication access. +func copySwapOptions(st statement.Statement) migrate.Options { opts := migrate.DefaultOptions() - opts.CopySwap = migrate.CopySwapOptions{Connection: dbconn.Config{URL: directURL}, Quiesced: true} + opts.CopySwap = migrate.CopySwapOptions{ + Connection: dbconn.Config{URL: directURL}, + Quiesced: st.Schema() + "." + st.Table(), + } return opts } @@ -37,7 +40,7 @@ func refuseCarriedCopySwap(t *testing.T, pool *pgxpool.Pool, table, sql string, before := snapshot(t, pool, table) st, err := statement.ParseOne(sql) require.NoError(t, err) - v, err := migrate.Run(t.Context(), pool, st, copySwapOptions()) + v, err := migrate.Run(t.Context(), pool, st, copySwapOptions(st)) require.NoError(t, err) require.Equal(t, verdict.OutcomeRefused, v.Outcome) assert.Equal(t, verdict.ReasonCopyAndSwapRefused, v.Reason) diff --git a/integration/supabase/schema_test.go b/integration/supabase/schema_test.go index 7a8a79be..15ebbc9d 100644 --- a/integration/supabase/schema_test.go +++ b/integration/supabase/schema_test.go @@ -100,7 +100,7 @@ func TestTextRewriteLeavesSchemaUnchanged(t *testing.T) { execSQL(t, pool, "INSERT INTO "+table+" VALUES (1,'00000000-0000-0000-0000-000000000001','123')") st, err := statement.ParseOne("ALTER TABLE " + table + " ALTER COLUMN body TYPE integer USING body::integer") require.NoError(t, err) - v, err := migrate.Run(t.Context(), pool, st, copySwapOptions()) + v, err := migrate.Run(t.Context(), pool, st, copySwapOptions(st)) require.NoError(t, err) require.Equal(t, verdict.OutcomeRefused, v.Outcome) assert.Equal(t, verdict.ReasonCopyAndSwapRefused, v.Reason) diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 3c989cc4..6a37c4fe 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -118,7 +118,7 @@ type MigrateCmd struct { LockBackoffMax time.Duration `help:"Maximum exponential backoff between lock-timeout attempts." default:"1s"` FreeDisk byteSize `help:"Free space on the volume the table lives on, for a statement the planner routes to copy-and-swap; the preflight refuses the run when the shadow table and its indexes would not fit (binary units: B, KiB, MiB, GiB, TiB). Unset means no space and refuses." name:"free-disk" placeholder:"SIZE"` KeepOld bool `help:"After a copy-and-swap, keep the retained source table under its _old name for a rollback window you end by hand, instead of dropping it before the run reports done." name:"keep-old"` - Quiesced bool `help:"Declare that nothing writes the table for the length of a copy-and-swap. The run then fences writes instead of streaming them through a replication slot, and needs no logical decoding from the server; a write that reaches the fence fails." name:"quiesced"` + Quiesced string `help:"Take the table offline for writes for the length of a copy-and-swap: the run fences writes instead of streaming them through a replication slot, and needs no logical decoding from the server, but every write that reaches the fence fails with SQLSTATE 55006 until the swap. The value is the typed acknowledgement: it must name the resolved schema-qualified target table exactly." name:"quiesced" placeholder:"SCHEMA.TABLE"` Restart string `help:"Start a copy-and-swap over instead of resuming the leftover of an earlier run: the leftover shadow table, checkpoint row, and write fence or replication slot are discarded under the run's lock, then the statement runs from the beginning. The value is the typed acknowledgement: it must name the resolved schema-qualified target table exactly. A leftover whose swap already committed is not discarded; run the statement without this flag to finish it. Cannot be combined with --force, --accept-blocking, or --dry-run." placeholder:"SCHEMA.TABLE"` DryRun bool `help:"Classify and route the statement, print the plan, and execute nothing."` JSON bool `help:"Emit the verdict (or dry-run plan) as JSON."` diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index 44a6823a..18519061 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -294,7 +294,7 @@ func TestMigrateRunsTypeRewriteViaCopyAndSwap(t *testing.T) { cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) cmd.FreeDisk = byteSize(testutil.UnlimitedDisk) - cmd.Quiesced = true + cmd.Quiesced = schema + ".t" cmd.JSON = true var out strings.Builder require.NoError(t, cmd.run(t.Context(), &out)) @@ -334,7 +334,7 @@ func TestMigrateCopyAndSwapRefusesWithoutFreeDisk(t *testing.T) { require.NoError(t, err) cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) - cmd.Quiesced = true + cmd.Quiesced = schema + ".t" cmd.JSON = true var out strings.Builder err = cmd.run(t.Context(), &out) diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index 15b2c688..082cbb37 100644 --- a/internal/cli/migrate_test.go +++ b/internal/cli/migrate_test.go @@ -106,22 +106,40 @@ func TestAcceptBlockingFlagGrammar(t *testing.T) { // executes nothing for them to govern. func TestCopySwapFlagsWireIntoEnginePolicy(t *testing.T) { t.Run("wires connection, free space, and choices", func(t *testing.T) { - c := parseMigrate(t, "--free-disk", "20GiB", "--keep-old", "--quiesced") + c := parseMigrate(t, "--free-disk", "20GiB", "--keep-old", "--quiesced", "app.orders") got := c.options(nil).CopySwap assert.Equal(t, c.Config(), got.Connection) assert.Equal(t, int64(20<<30), got.FreeDiskBytes) assert.True(t, got.KeepOld) - assert.True(t, got.Quiesced) + assert.Equal(t, "app.orders", got.Quiesced) }) t.Run("defaults to no free space and the live path", func(t *testing.T) { got := parseMigrate(t).options(nil).CopySwap assert.Zero(t, got.FreeDiskBytes) assert.False(t, got.KeepOld) - assert.False(t, got.Quiesced) + assert.Empty(t, got.Quiesced) }) - for _, flag := range []string{"--free-disk=1GiB", "--keep-old", "--quiesced"} { + // Taking a table offline for writes needs the table named: a bare + // --quiesced, as a script written for the old boolean flag passes it, + // is a parse error rather than a fence on whatever table the statement + // names. + t.Run("rejects --quiesced without the table", func(t *testing.T) { + c := New("test") + k, err := kong.New(c, kong.Vars{"version": "test"}) + require.NoError(t, err) + _, err = k.Parse([]string{ + "migrate", + "--url", "postgres://user@localhost:5432/app", + "--alter", "ALTER TABLE t ALTER COLUMN c TYPE bigint", + "--quiesced", + }) + require.Error(t, err) + assert.Contains(t, err.Error(), "--quiesced") + }) + + for _, flag := range []string{"--free-disk=1GiB", "--keep-old", "--quiesced=public.t"} { t.Run("rejects "+flag+" with --dry-run", func(t *testing.T) { c := New("test") k, err := kong.New(c, kong.Vars{"version": "test"}) diff --git a/internal/cli/restart_integration_test.go b/internal/cli/restart_integration_test.go index 287b593f..60e876f2 100644 --- a/internal/cli/restart_integration_test.go +++ b/internal/cli/restart_integration_test.go @@ -42,7 +42,7 @@ func TestMigrateRestartDiscardsALeftoverAndRunsTheChange(t *testing.T) { alter := fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema) plain := newMigrateCmd(url, alter) plain.FreeDisk = byteSize(testutil.UnlimitedDisk) - plain.Quiesced = true + plain.Quiesced = schema + ".t" plain.JSON = true var refusedOut strings.Builder require.ErrorIs(t, plain.run(t.Context(), &refusedOut), verdict.ErrRefused, "a refusal verdict exits with the refusal code") @@ -53,7 +53,7 @@ func TestMigrateRestartDiscardsALeftoverAndRunsTheChange(t *testing.T) { restart := newMigrateCmd(url, alter) restart.FreeDisk = byteSize(testutil.UnlimitedDisk) - restart.Quiesced = true + restart.Quiesced = schema + ".t" restart.JSON = true restart.Restart = schema + ".t" var out strings.Builder @@ -100,7 +100,7 @@ func TestMigrateRestartAckMismatchExecutesNothing(t *testing.T) { cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) cmd.FreeDisk = byteSize(testutil.UnlimitedDisk) - cmd.Quiesced = true + cmd.Quiesced = schema + ".t" cmd.Restart = "wrong.table" var out strings.Builder err = cmd.run(t.Context(), &out) diff --git a/internal/cli/signal_integration_test.go b/internal/cli/signal_integration_test.go index a5344732..b838d312 100644 --- a/internal/cli/signal_integration_test.go +++ b/internal/cli/signal_integration_test.go @@ -75,7 +75,7 @@ func TestMigrateSignalDuringCopyReportsCancelledByCaller(t *testing.T) { cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) cmd.FreeDisk = byteSize(testutil.UnlimitedDisk) - cmd.Quiesced = true + cmd.Quiesced = schema + ".t" cmd.JSON = true var out strings.Builder err = cmd.run(ctx, &out) diff --git a/pkg/capabilities/front_door_integration_test.go b/pkg/capabilities/front_door_integration_test.go index cf0bd2b2..6b5e7c17 100644 --- a/pkg/capabilities/front_door_integration_test.go +++ b/pkg/capabilities/front_door_integration_test.go @@ -251,7 +251,6 @@ func TestMigrateSupportedRowsExecute(t *testing.T) { opts.CopySwap = migrate.CopySwapOptions{ Connection: dbconn.Config{URL: url}, FreeDiskBytes: testutil.UnlimitedDisk, - Quiesced: true, } executed = verdict.OutcomeExecutedViaCopyAndSwap } @@ -259,6 +258,9 @@ func TestMigrateSupportedRowsExecute(t *testing.T) { sql = fmt.Sprintf(sql, schema) st, err := statement.ParseOne(sql) require.NoError(t, err) + if claim.copyAndSwap { + opts.CopySwap.Quiesced = st.Schema() + "." + st.Table() + } v, err := migrate.Run(t.Context(), pool, st, opts) require.NoError(t, err, sql) assert.Equal(t, executed, v.Outcome, diff --git a/pkg/migrate/copy_swap.go b/pkg/migrate/copy_swap.go index 5003da14..0175450f 100644 --- a/pkg/migrate/copy_swap.go +++ b/pkg/migrate/copy_swap.go @@ -65,11 +65,15 @@ type CopySwapOptions struct { // drops it before the run reports done. KeepOld bool - // Quiesced declares that nothing writes the table while the run lasts. - // The run then takes the write fence instead of a replication slot and - // needs no logical decoding from the server; a write that reaches the - // fence fails. False runs the live path: slot, stream, and catch-up. - Quiesced bool + // Quiesced, when set, declares that nothing writes the table while the + // run lasts. The run then takes the write fence instead of a + // replication slot and needs no logical decoding from the server; every + // write that reaches the fence fails, so the table is offline for + // writes until the swap. It must name the resolved target table as + // SCHEMA.TABLE, the same way Restart does, so the declaration cannot be + // left on a script and fence another table by accident. Empty runs the + // live path: slot, stream, and catch-up. + Quiesced string // Restart, when set, is the operator's acknowledgement that the run on // the table is to start over: whatever an earlier run left — its @@ -171,7 +175,7 @@ func (o CopySwapOptions) runOptions(logger *slog.Logger) schemachange.RunOptions // needs none of the server's logical-decoding facilities. func (o CopySwapOptions) environment() preflight.CopySwapEnvironment { return preflight.CopySwapEnvironment{ - LogicalDecoding: !o.Quiesced, + LogicalDecoding: o.Quiesced == "", FreeDiskBytes: o.FreeDiskBytes, } } @@ -209,8 +213,12 @@ func copyAndSwap(ctx context.Context, pool *pgxpool.Pool, st statement.Statement logger.Debug("copy-and-swap preflight passed", "decodes_wal", target.DecodesWAL(), "keep_old", opts.CopySwap.KeepOld) + quiesced := opts.CopySwap.Quiesced != "" + if quiesced { + auditQuiesced(opts.audit(), st) + } result, err := runner.Run(ctx, pool, target, st) - return copySwapOutcome(st, opts.CopySwap.Quiesced, result, err) + return copySwapOutcome(st, quiesced, result, err) } // copySwapOutcome folds what the Runner returned into Run's contract. A diff --git a/pkg/migrate/copy_swap_connection_integration_test.go b/pkg/migrate/copy_swap_connection_integration_test.go index 17c2e2a5..60f71a60 100644 --- a/pkg/migrate/copy_swap_connection_integration_test.go +++ b/pkg/migrate/copy_swap_connection_integration_test.go @@ -94,7 +94,7 @@ func TestCopyAndSwapRefusesAConnectionOnAnotherDatabase(t *testing.T) { schema := d.createOrders(t) st := parseOne(t, fmt.Sprintf("ALTER TABLE %s.orders ALTER COLUMN id TYPE bigint", schema)) - opts := copySwapRunOptions(d.otherURL) + opts := copySwapRunOptions(d.otherURL, schema+".orders") v, err := migrate.Run(t.Context(), d.pool, st, opts) var mismatch *migrate.CopySwapConnectionMismatchError require.ErrorAs(t, err, &mismatch) @@ -105,7 +105,7 @@ func TestCopyAndSwapRefusesAConnectionOnAnotherDatabase(t *testing.T) { assert.False(t, hasEngineSchema(t, d.poolURL), "no checkpoint row was written in the pool's database") assert.False(t, hasEngineSchema(t, d.otherURL), "nothing was written through the connection either") - v, err = migrate.Run(t.Context(), d.pool, st, copySwapRunOptions(d.poolURL)) + v, err = migrate.Run(t.Context(), d.pool, st, copySwapRunOptions(d.poolURL, schema+".orders")) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) assert.Equal(t, "bigint", d.columnType(t, schema)) @@ -127,7 +127,7 @@ func TestRunDesiredRefusesAConnectionOnAnotherDatabase(t *testing.T) { qty integer NOT NULL, note text )`), - }, copySwapRunOptions(d.otherURL)) + }, copySwapRunOptions(d.otherURL, schema+".orders")) var mismatch *migrate.CopySwapConnectionMismatchError require.ErrorAs(t, err, &mismatch) assert.Equal(t, mismatch.PoolCluster, mismatch.ConnectionCluster, "both databases are on the test server") @@ -163,7 +163,7 @@ func TestCopyAndSwapRefusesADatabaseOfTheSameNameOnAnotherCluster(t *testing.T) v, err := migrate.Run(t.Context(), d.pool, parseOne(t, fmt.Sprintf("ALTER TABLE %s.orders ALTER COLUMN id TYPE bigint", schema)), - copySwapRunOptions(sameName)) + copySwapRunOptions(sameName, schema+".orders")) var mismatch *migrate.CopySwapConnectionMismatchError require.ErrorAs(t, err, &mismatch) assert.Zero(t, v, "nothing was executed, so there is no verdict") @@ -184,7 +184,7 @@ func TestCopyAndSwapStopsWhenTheConnectionCannotBeOpened(t *testing.T) { schema := d.createOrders(t) missing := withDatabase(t, d.poolURL, "no_such_database") - assertStopsBeforeAnyStatement(t, d, d.pool, schema, copySwapRunOptions(missing)) + assertStopsBeforeAnyStatement(t, d, d.pool, schema, copySwapRunOptions(missing, schema+".orders")) } // The identity is read with pg_control_system(), which the server grants to @@ -234,14 +234,14 @@ func TestCopyAndSwapStopsWhenEitherSideCannotBeIdentified(t *testing.T) { roleURL := withUser(t, d.poolURL, role, "unidentified") t.Run("the connection", func(t *testing.T) { - assertStopsBeforeAnyStatement(t, d, d.pool, schema, copySwapRunOptions(roleURL)) + assertStopsBeforeAnyStatement(t, d, d.pool, schema, copySwapRunOptions(roleURL, schema+".orders")) }) t.Run("the pool", func(t *testing.T) { rolePool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: roleURL}) require.NoError(t, err) t.Cleanup(rolePool.Close) - assertStopsBeforeAnyStatement(t, d, rolePool, schema, copySwapRunOptions(d.poolURL)) + assertStopsBeforeAnyStatement(t, d, rolePool, schema, copySwapRunOptions(d.poolURL, schema+".orders")) }) } diff --git a/pkg/migrate/copy_swap_integration_test.go b/pkg/migrate/copy_swap_integration_test.go index 3f4d1fea..c7f1fb69 100644 --- a/pkg/migrate/copy_swap_integration_test.go +++ b/pkg/migrate/copy_swap_integration_test.go @@ -74,14 +74,15 @@ func (f copySwapFixture) createOrders(t *testing.T) { FROM generate_series(1, 2500) g`) } -// options is a quiesced copy-and-swap policy over the fixture's database -// with room on disk, the shape every test here starts from. +// options is a copy-and-swap policy over the fixture's database with room +// on disk, quiesced for the orders table, the shape every test here starts +// from. func (f copySwapFixture) options() Options { opts := DefaultOptions() opts.CopySwap = CopySwapOptions{ Connection: f.cfg, FreeDiskBytes: testutil.UnlimitedDisk, - Quiesced: true, + Quiesced: f.schema + ".orders", } return opts } diff --git a/pkg/migrate/copy_swap_quiesced.go b/pkg/migrate/copy_swap_quiesced.go new file mode 100644 index 00000000..0aebe4eb --- /dev/null +++ b/pkg/migrate/copy_swap_quiesced.go @@ -0,0 +1,32 @@ +package migrate + +import ( + "fmt" + "log/slog" + + "github.com/block/pg-sprite/pkg/schemachange" + "github.com/block/pg-sprite/pkg/statement" +) + +// checkQuiescedAck confirms that the quiesced acknowledgement names the +// resolved target table, exactly as the force and restart acknowledgements +// must. A quiesced run refuses every write to the table for its whole +// length, so a typed table name is what stops the declaration from being +// left on a script and taking the next table offline by accident. +func checkQuiescedAck(st statement.Statement, ack string) error { + if ack == qualified(st) { + return nil + } + return fmt.Errorf("the quiesced acknowledgement must name the resolved target table %q, got %q; nothing was executed", + qualified(st), ack) +} + +// auditQuiesced records, before the fence goes on, that the operator took +// the table offline for writes. The record is warn-level and +// unconditional, as the force and restart audits are: an audit trail must +// not depend on diagnostics being enabled. +func auditQuiesced(audit *slog.Logger, st statement.Statement) { + audit.Warn("quiesced run acknowledged; every write to the table will be refused until the swap", + "table", qualified(st), + "sqlstate", schemachange.WriteFenceSQLState) +} diff --git a/pkg/migrate/copy_swap_quiesced_integration_test.go b/pkg/migrate/copy_swap_quiesced_integration_test.go new file mode 100644 index 00000000..b67be247 --- /dev/null +++ b/pkg/migrate/copy_swap_quiesced_integration_test.go @@ -0,0 +1,34 @@ +package migrate + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/schemachange" + "github.com/block/pg-sprite/pkg/statement" + "github.com/block/pg-sprite/pkg/verdict" +) + +// A quiesced acknowledgement that names another table is an error before +// any verdict, and the table is never taken offline: no shadow is built, +// no fence goes on, so the source still takes writes, and its shape is +// unchanged. The acknowledgement is what stops a quiesced declaration left +// on a script from fencing the next table it meets. +func TestRunRejectsAMismatchedQuiescedAck(t *testing.T) { + f := newCopySwapFixture(t) + f.createOrders(t) + st, err := statement.ParseOne(f.inSchema(`ALTER TABLE {schema}.orders ALTER COLUMN qty TYPE bigint`)) + require.NoError(t, err) + + opts := f.options() + opts.CopySwap.Quiesced = f.schema + ".customers" + v, err := Run(t.Context(), f.pool, st, opts) + require.Error(t, err) + assert.Contains(t, err.Error(), "quiesced acknowledgement") + assert.Equal(t, verdict.Verdict{}, v, "an error before a verdict carries a zero verdict") + assert.False(t, f.relationExists(t, schemachange.ShadowName(f.schema, "orders")), "no shadow was built") + f.exec(t, `INSERT INTO {schema}.orders (qty, note, sku) VALUES (1, 'after', 'sku1')`) + assert.Equal(t, "integer", f.columnType(t, "orders", "qty"), "nothing was executed") +} diff --git a/pkg/migrate/desired_copy_swap_integration_test.go b/pkg/migrate/desired_copy_swap_integration_test.go index 45bb5e16..5f911a12 100644 --- a/pkg/migrate/desired_copy_swap_integration_test.go +++ b/pkg/migrate/desired_copy_swap_integration_test.go @@ -17,14 +17,14 @@ import ( // copySwapRunOptions are runOptions with the copy-and-swap route wired the // way the CLI wires it: the route's own connection, free disk the shadow -// will fit in, and the quiesced declaration so the test server needs no -// logical decoding. -func copySwapRunOptions(serverURL string) migrate.Options { +// will fit in, and the quiesced acknowledgement for table (SCHEMA.TABLE) so +// the test server needs no logical decoding. +func copySwapRunOptions(serverURL, table string) migrate.Options { opts := runOptions() opts.CopySwap = migrate.CopySwapOptions{ Connection: dbconn.Config{URL: serverURL}, FreeDiskBytes: testutil.UnlimitedDisk, - Quiesced: true, + Quiesced: table, } return opts } @@ -62,7 +62,7 @@ func TestRunDesiredCopyAndSwap(t *testing.T) { // instant, then a retype the router sends through copy-and-swap. // With no free disk declared the route's preflight refuses the // retype; admission reads that before the ADD COLUMN runs. - opts := copySwapRunOptions(serverURL) + opts := copySwapRunOptions(serverURL, schema+".t") opts.CopySwap.FreeDiskBytes = 0 res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{ Schema: schema, @@ -114,7 +114,7 @@ func TestRunDesiredCopyAndSwap(t *testing.T) { body varchar(10) )`), } - res, err := migrate.RunDesired(t.Context(), pool, req, copySwapRunOptions(serverURL)) + res, err := migrate.RunDesired(t.Context(), pool, req, copySwapRunOptions(serverURL, schema+".t")) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecuted, res.Outcome) require.Len(t, res.Plan.Statements, 2, "one retype per column") @@ -130,7 +130,7 @@ func TestRunDesiredCopyAndSwap(t *testing.T) { // The convergence oracle: a second run derives an empty plan and // runs nothing. - res, err = migrate.RunDesired(t.Context(), pool, req, copySwapRunOptions(serverURL)) + res, err = migrate.RunDesired(t.Context(), pool, req, copySwapRunOptions(serverURL, schema+".t")) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecuted, res.Outcome) assert.Empty(t, res.Plan.Statements, "the converged table plans no statements") diff --git a/pkg/migrate/migrate.go b/pkg/migrate/migrate.go index 8a6f877a..d4a64b9b 100644 --- a/pkg/migrate/migrate.go +++ b/pkg/migrate/migrate.go @@ -256,6 +256,11 @@ func run(ctx context.Context, pool *pgxpool.Pool, st statement.Statement, opts O return verdict.Verdict{}, err } } + if opts.CopySwap.Quiesced != "" { + if err := checkQuiescedAck(st, opts.CopySwap.Quiesced); err != nil { + return verdict.Verdict{}, err + } + } facts, err := LiveFacts(ctx, pool, st) if err != nil { diff --git a/pkg/migrate/rename_index.go b/pkg/migrate/rename_index.go index f97a3607..872705e2 100644 --- a/pkg/migrate/rename_index.go +++ b/pkg/migrate/rename_index.go @@ -90,6 +90,10 @@ func renameIndex(ctx context.Context, pool *pgxpool.Pool, st statement.Statement return verdict.Verdict{}, fmt.Errorf("%w: an index rename never runs through copy-and-swap; nothing was executed", ErrIndexRenameAcknowledgement) } + if opts.CopySwap.Quiesced != "" { + return verdict.Verdict{}, fmt.Errorf("%w: quiesced names a table to fence, and an index rename never runs through copy-and-swap; nothing was executed", + ErrIndexRenameAcknowledgement) + } facts, err := LiveFacts(ctx, pool, st) if err != nil { return verdict.Verdict{}, err diff --git a/pkg/migrate/run_integration_test.go b/pkg/migrate/run_integration_test.go index 9410bcf9..0afaa2c7 100644 --- a/pkg/migrate/run_integration_test.go +++ b/pkg/migrate/run_integration_test.go @@ -196,7 +196,7 @@ func TestRunDispatch(t *testing.T) { ALTER TABLE %s.t ALTER COLUMN id TYPE bigint, ALTER COLUMN v SET NOT NULL, - ADD CONSTRAINT t_qty_positive CHECK (qty > 0)`, schema)), copySwapRunOptions(url)) + ADD CONSTRAINT t_qty_positive CHECK (qty > 0)`, schema)), copySwapRunOptions(url, schema+".t")) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) From 8530c63128b70cd81251de8cdcf2f3dfef5668ad Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 28 Oct 2026 13:49:09 +1100 Subject: [PATCH 2/3] migrate: rename --quiesced to --block-writes The flag takes the table offline for writes, and "quiesced" read as a promise about the table rather than a warning about what the run does. --- docs/cli-output-examples.md | 4 ++-- docs/limitations.md | 4 ++-- docs/schemabot-integration.md | 8 +++---- integration/supabase/copy_swap_route_test.go | 6 +++--- internal/cli/cli.go | 4 ++-- internal/cli/migrate.go | 2 +- internal/cli/migrate_integration_test.go | 4 ++-- internal/cli/migrate_test.go | 21 +++++++++---------- internal/cli/restart_integration_test.go | 6 +++--- internal/cli/signal_integration_test.go | 2 +- .../front_door_integration_test.go | 2 +- pkg/migrate/copy_swap.go | 12 +++++------ ..._quiesced.go => copy_swap_block_writes.go} | 14 ++++++------- ...opy_swap_block_writes_integration_test.go} | 10 ++++----- pkg/migrate/copy_swap_integration_test.go | 2 +- pkg/migrate/copy_swap_test.go | 2 +- .../desired_copy_swap_integration_test.go | 2 +- pkg/migrate/migrate.go | 4 ++-- pkg/migrate/rename_index.go | 4 ++-- 19 files changed, 56 insertions(+), 57 deletions(-) rename pkg/migrate/{copy_swap_quiesced.go => copy_swap_block_writes.go} (53%) rename pkg/migrate/{copy_swap_quiesced_integration_test.go => copy_swap_block_writes_integration_test.go} (76%) diff --git a/docs/cli-output-examples.md b/docs/cli-output-examples.md index 7a16cf52..7b92bfe4 100644 --- a/docs/cli-output-examples.md +++ b/docs/cli-output-examples.md @@ -374,8 +374,8 @@ $ pg-sprite migrate --alter 'ALTER TABLE users ALTER COLUMN id TYPE text' --dry- Without `--dry-run` the run needs `--free-disk` (the preflight refuses with `copy-and-swap-refused` / cause `copy-and-swap-disk-headroom` when the shadow and its indexes would not fit the declared space) and either logical -decoding on the server or `--quiesced SCHEMA.TABLE`, which names the table it -takes offline for writes until the swap. The verdict's outcome is distinct +decoding on the server or `--block-writes SCHEMA.TABLE`, which names the table +whose writes it blocks until the swap. The verdict's outcome is distinct from `executed-natively`, and `detail` says what the verification pass covered and what became of the retained source: diff --git a/docs/limitations.md b/docs/limitations.md index 3dba2994..18ca7260 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -23,7 +23,7 @@ not escape hatches: | Retyping or dropping a column a view reads | PostgreSQL refuses `ALTER COLUMN TYPE` and `DROP COLUMN` on a column a view or materialized view reads, even for a type change it would relabel in place. Without `ONLY`, a type change reaches the column on every partition and inheritance child too, and a drop reaches every partition and each inheritance child that does not declare the column and inherits it only from tables the drop also reaches; a child that declares the column or also inherits it from a table outside the drop keeps its copy, while one that inherits it along two paths from the dropped table loses it. The planner reads the views on every copy the change reaches from the catalog and refuses the change with `dependent-view`, naming each view in the plan and in the verdict's `dependent_views`, instead of letting the statement fail at the server. `DROP COLUMN … CASCADE`, which would drop the views with the column, is refused with `cascading-drop`. pg-sprite does not drop and recreate the views for you. Drop or redefine them first, then re-plan. | | Dropping with `CASCADE` | `CASCADE` on `DROP COLUMN`, `DROP CONSTRAINT`, or `DROP INDEX` also drops whatever depends on the target: a view that reads the column, or another table's foreign key on the constraint. Those objects are not in the statement, so the plan cannot show them and the table's lock does not cover them. The planner refuses every such drop with `cascading-drop`. Drop each dependent object as its own change, then resubmit without `CASCADE`; PostgreSQL then refuses the drop while anything still depends on the target. | | `ADD COLUMN` on a hot standby, or by a role without `TEMPORARY` | Whether adding a column rewrites the table is the server's answer, read by adding the column to an empty temporary table. A server that accepts no writes, or a role whose `TEMPORARY` on the database was revoked, cannot create that table, so the plan fails with `ErrAddColumnProbeUnavailable` before anything runs. Plan against the primary, or grant `TEMPORARY` on the database to the engine's role. | -| Copy-and-swap | The copy-and-swap backend carries a general `ALTER COLUMN TYPE` (with or without `USING`): the changed shape is built as a shadow table, filled in chunks while writes stream through a logical replication slot, verified, and swapped in under the source's name. The run needs `--free-disk` (the preflight refuses `copy-and-swap-disk-headroom` when the shadow and its indexes would not fit) and either logical decoding (`wal_level = logical`, a free replication slot) or `--quiesced SCHEMA.TABLE`, which names the table it takes offline for writes until the swap; its preflight refuses table shapes it cannot run against — no primary key or usable replica identity, unlogged, partitioned, `FORCE ROW LEVEL SECURITY`, foreign keys in either direction, triggers, dependent views, membership in a publication, or a subscription on the database — each as a typed `copy-and-swap-refused` cause ([refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause)). The other rewrite-class operations — volatile-default `ADD COLUMN`, `ADD COLUMN` of a domain with a `CHECK` or `NOT NULL`, `GENERATED … STORED`, `SET TABLESPACE` — route to the backend but it does not carry them yet; they refuse `backend-unavailable`. pg-sprite never falls through to a blocking rewrite. A run that stops short of the swap keeps its hold on the source until it is resumed or abandoned — the write fence of a quiesced run (every write is refused `55006`), or the replication slot of a live run, which pins WAL on the server — and the `failed` verdict's detail says which applies: a stop the next run clears (a lost connection, a cancelled run) resumes from the checkpoint row when the same statement is run again; a stop the data causes meets the same rows again, so it is not resumed: when the server rejects a row under the new type or a constraint (SQLSTATE class `22` or `23`) the run abandons itself — drops the shadow, releases the fence or slot, deletes the checkpoint row — and the verdict is `failed` with `code: data-fault`, so correct the data or the statement and run it again from the beginning; when the shadow and source diverge at verification (or the abandon itself failed), the row and the hold stay and the run has to be started over: correct the data or the statement, then run it with `--restart SCHEMA.TABLE` (`CopySwapOptions.Restart` as a library caller), which drops the shadow, releases the fence or slot, deletes the checkpoint row, and runs the change fresh under the run's own lock ([stopping, resuming, and starting over](schemabot-integration.md#stopping-resuming-and-starting-over)). | +| Copy-and-swap | The copy-and-swap backend carries a general `ALTER COLUMN TYPE` (with or without `USING`): the changed shape is built as a shadow table, filled in chunks while writes stream through a logical replication slot, verified, and swapped in under the source's name. The run needs `--free-disk` (the preflight refuses `copy-and-swap-disk-headroom` when the shadow and its indexes would not fit) and either logical decoding (`wal_level = logical`, a free replication slot) or `--block-writes SCHEMA.TABLE`, which names the table whose writes it blocks until the swap; its preflight refuses table shapes it cannot run against — no primary key or usable replica identity, unlogged, partitioned, `FORCE ROW LEVEL SECURITY`, foreign keys in either direction, triggers, dependent views, membership in a publication, or a subscription on the database — each as a typed `copy-and-swap-refused` cause ([refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause)). The other rewrite-class operations — volatile-default `ADD COLUMN`, `ADD COLUMN` of a domain with a `CHECK` or `NOT NULL`, `GENERATED … STORED`, `SET TABLESPACE` — route to the backend but it does not carry them yet; they refuse `backend-unavailable`. pg-sprite never falls through to a blocking rewrite. A run that stops short of the swap keeps its hold on the source until it is resumed or abandoned — the write fence of a `--block-writes` run (every write is refused `55006`), or the replication slot of a live run, which pins WAL on the server — and the `failed` verdict's detail says which applies: a stop the next run clears (a lost connection, a cancelled run) resumes from the checkpoint row when the same statement is run again; a stop the data causes meets the same rows again, so it is not resumed: when the server rejects a row under the new type or a constraint (SQLSTATE class `22` or `23`) the run abandons itself — drops the shadow, releases the fence or slot, deletes the checkpoint row — and the verdict is `failed` with `code: data-fault`, so correct the data or the statement and run it again from the beginning; when the shadow and source diverge at verification (or the abandon itself failed), the row and the hold stay and the run has to be started over: correct the data or the statement, then run it with `--restart SCHEMA.TABLE` (`CopySwapOptions.Restart` as a library caller), which drops the shadow, releases the fence or slot, deletes the checkpoint row, and runs the change fresh under the run's own lock ([stopping, resuming, and starting over](schemabot-integration.md#stopping-resuming-and-starting-over)). | | `ALTER COLUMN TYPE ... USING ` the copy cannot reproduce | The copy-and-swap route (above) runs the retype, but the shadow builder admits a `USING` expression only when it can evaluate it again, identically, in the chunk copy, the verifier's recopy, and every change the catch-up flushes. Three rules, each refused as `shadow-statement-conversion` before any shadow is committed: the expression must be **immutable** by the server's own classification (no `now()`, `random()`, `current_setting()`, session-dependent casts such as `text` to `timestamptz`, or functions declared `STABLE` or `VOLATILE`; checked by trying it as a stored generated column's expression on the empty shadow inside the build's transaction, then rolling that attempt back), it must **name only the column it retypes**, bare and unqualified (no neighbouring column, no `t.col`, no whole-row `t.*`), and it must not retype the **primary key**, whose values every chunk range, watermark, and resume clear are expressed in. The expression is also refused when the deparser cannot prove it rendered the text faithfully. Rewrite the expression in terms of the retyped column alone and of `IMMUTABLE` functions, or retype without `USING` so the server's assignment cast converts the value; the refusal names the column, the expression, and that rewrite. The spellings most often refused, and what to write instead, are in [`USING` expressions the copy reproduces](#using-expressions-the-copy-reproduces) below. | | A schema change to the source table while a copy-and-swap runs | Unsupported: logical decoding does not stream DDL, and the copy does not lock it out — a chunk holds `ACCESS SHARE` on the source only for its own transaction, so `ALTER TABLE` on the source completes between chunks. The copy selects the source by column name, so what happens depends on whether every copied name still resolves. A copied column renamed or dropped stops the copy at its next chunk with the server's `undefined_column` error (SQLSTATE `42703`), which is not a typed refusal; that chunk stays in flight and nothing from it reaches the shadow. A change that leaves every copied name resolving to a column of a compatible type — two columns swapping names, a column renamed and re-added under the old name, a widened type — is not detected by the copy, which completes, and after a swap or a re-add the rows copied since hold another column's values or the new column's default. Once the replication stream has decoded a change to the table, a later relation description that differs stops it with `ErrSourceShapeChanged`. None of these reaches the swap: the verifier compares source and shadow rows by column name, and the cutover gate re-fingerprints the source and refuses any changed column as `cutover-schema-drift` ([refusal-classes.md](refusal-classes.md#shadow-operation-refusals-keyed-on-refusalcause)), after which the shadow is dropped and the change re-planned. | | A concurrent index build whose name is occupied by an invalid index, or whose table carries quarantined debris | The build refuses before running and never drops the occupant: an invalid entry is either abandoned debris or someone's build still in progress, and a drop by name cannot tell them apart. The refusal is typed by proof (`invalid-index-abandoned`, `invalid-index-build-in-flight`, `invalid-index-other-table`, `invalid-index-not-droppable`, `invalid-index-builder-unobservable`), and a table carrying an entry a previous recovery quarantined (`pgsprite_abandoned_`) but did not drop is refused the same way, so debris never goes quiet. The proven removal — `executor.RebuildAbandonedIndex`, or `executor.DropAbandonedIndex` to remove the leftover without running the build; both library-only today, no CLI verb — takes a bounded `SHARE UPDATE EXCLUSIVE` lock on the table (which any running concurrent index command holds), re-verifies the entry by OID, renames it to an identity-derived name and drops that, all within one shared budget; it refuses a visible in-flight build, another table's entry, or an entry the server will not drop concurrently (a partitioned table's index, an index partition, a constraint's index), and reports a lock budget rather than wait behind a build it cannot see. From the CLI, the [runbook](invalid-index-recovery.md) applies. | @@ -121,7 +121,7 @@ composition of the model boundaries above with those gates. At a glance: | Add a constraint (`UNIQUE`, `CHECK`); `SET NOT NULL` | Converges via the safer online sequence; not size-guarded either. | | Relax a `NOT NULL` | Refused, whole plan: dropping `NOT NULL` discards the same guarantee its constraint form would, so it is destructive. Run it deliberately through the imperative front door — it executes natively there — then rerun. | | Change an index or constraint definition | Refused, whole plan — the drop-and-recreate row above. | -| Narrow a column type (`varchar(255)` → `varchar(50)`) | Converges through copy-and-swap: the plan's disposition is `execute` with the statement routed to the `copy-and-swap` backend, and desired-state execution runs it with the same preflight gates and flags as `migrate --alter` (the Copy-and-swap row above). The route's preflight is read at admission, before any statement runs, so a table or server the route refuses — no `--free-disk`, no logical decoding without `--quiesced SCHEMA.TABLE`, a trigger, a dependent view — refuses the whole plan with nothing executed, native statements included. The dry run and `diff` do not read the preflight — the plan report says `execute` on the route alone, and whether the route can run here, now, with these flags, is answered when the plan is executed. A plan that retypes two columns runs two copy-and-swap passes on the table, one per statement; the second replaces the done row the first leaves behind. | +| Narrow a column type (`varchar(255)` → `varchar(50)`) | Converges through copy-and-swap: the plan's disposition is `execute` with the statement routed to the `copy-and-swap` backend, and desired-state execution runs it with the same preflight gates and flags as `migrate --alter` (the Copy-and-swap row above). The route's preflight is read at admission, before any statement runs, so a table or server the route refuses — no `--free-disk`, no logical decoding without `--block-writes SCHEMA.TABLE`, a trigger, a dependent view — refuses the whole plan with nothing executed, native statements included. The dry run and `diff` do not read the preflight — the plan report says `execute` on the route alone, and whether the route can run here, now, with these flags, is answered when the plan is executed. A plan that retypes two columns runs two copy-and-swap passes on the table, one per statement; the second replaces the done row the first leaves behind. | | A bounded-attempt edit on a table above the size threshold | Refused at that statement (`not-native-safe-table-too-large`): with the default 1 GiB threshold, adding a column to a larger table refuses until the threshold is raised. | A destructive refusal is all-or-nothing: one destructive statement refuses diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 08971a54..a18ce263 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -458,9 +458,9 @@ the environmental `shadow-lock-unconfirmed` refusal, when its first working tran finds no lock to confirm. The check reads `pg_control_system()`, which PostgreSQL grants to `PUBLIC`; a server that withholds it stops the route with an ordinary error), the free space on the table's volume for the preflight's disk-headroom check, whether to keep the retained source under -its `_old` name, and whether the table is quiesced for the run — the fenced path that needs -no logical decoding but refuses every write to the table with `55006` until the swap — or -live. `Quiesced` is a typed acknowledgement, not a switch: it must name the resolved target +its `_old` name, and whether the run blocks the table's writes — offline mode, the fenced path +(a quiesced target, in the engine's terms) that needs no logical decoding but refuses every +write to the table with `55006` until the swap — or is live. `BlockWrites` is a typed acknowledgement, not a switch: it must name the resolved target as `SCHEMA.TABLE`, the same way `Restart` does, or `Run` returns an error before anything executes, and an acknowledged run records a warn-level audit naming the table before the fence goes on. Three further fields let the adapter size and watch the run: @@ -658,7 +658,7 @@ CLI and through `migrate.Run`, plus one the Runner performs on its own; an orche | Operator need | What to do | What pg-sprite does | | --- | --- | --- | -| **Start** a change | `pg-sprite migrate --url … 'ALTER TABLE s.t …'` (with `--free-disk`, and `--quiesced SCHEMA.TABLE` to take the table offline for writes instead of streaming them) | Routes the statement; on the copy-and-swap backend it takes the table's lock session, writes the checkpoint row, builds and fills the shadow, verifies, swaps, and prints one verdict | +| **Start** a change | `pg-sprite migrate --url … 'ALTER TABLE s.t …'` (with `--free-disk`, and `--block-writes SCHEMA.TABLE` to take the table offline for writes instead of streaming them) | Routes the statement; on the copy-and-swap backend it takes the table's lock session, writes the checkpoint row, builds and fills the shadow, verifies, swaps, and prints one verdict | | **Stop** a running change | Send `SIGINT` (Ctrl-C) or `SIGTERM` once | The run's context is cancelled and the run unwinds through its own stop path: a copy in flight saves the watermark its committed chunks reached, the table lock session is released, the row stays at its last save with the run's hold on the source — the write fence of a quiesced run, or the replication slot of a live run, which retains WAL until the run is finished or discarded — and the verdict is `failed` with `code: cancelled-by-caller` (exit 1). A second `SIGINT`/`SIGTERM` ends the process at once with exit 130, before the stop path completes; the row then records whatever the last save landed, and the lock session is released by the server when its connection drops | | **Resume** a stopped change | Run the same statement again | The checkpoint row says where to continue: a stopped copy continues from its watermark, a run past its copy verifies again, a `cutover` row is answered by the catalog (a committed swap finishes from the old-table drop; a rolled-back swap goes back through verify and gate). Nothing is copied twice. A different statement on the same table is refused (`resume-row-mismatch`) | | **Start over** — discard a stopped change and run it from the beginning | Run the same statement with `--restart SCHEMA.TABLE`, where the value names the resolved target exactly | Under the run's own table lock session, discards what the stopped run left in the order a later restart could meet idempotently — the shadow, then the hold on the source (`DropWriteFence` for a quiesced row, `decode.DropSlot` for a live row), then the checkpoint row — and runs the change fresh. A leftover shadow that no row records is discarded the same way. The decision is written to the audit log before anything is discarded | diff --git a/integration/supabase/copy_swap_route_test.go b/integration/supabase/copy_swap_route_test.go index dcdfb80a..bcd25e83 100644 --- a/integration/supabase/copy_swap_route_test.go +++ b/integration/supabase/copy_swap_route_test.go @@ -19,13 +19,13 @@ import ( // workers, or replication stream open, which is the fail-closed shape these // probes rely on to leave the fixture untouched. The only session opened // from the connection is the read-only one that proves it reaches the -// pool's database. Quiesced, acknowledged for st's table, keeps the probe +// pool's database. BlockWrites, acknowledged for st's table, keeps the probe // independent of the fixture role's replication access. func copySwapOptions(st statement.Statement) migrate.Options { opts := migrate.DefaultOptions() opts.CopySwap = migrate.CopySwapOptions{ - Connection: dbconn.Config{URL: directURL}, - Quiesced: st.Schema() + "." + st.Table(), + Connection: dbconn.Config{URL: directURL}, + BlockWrites: st.Schema() + "." + st.Table(), } return opts } diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 6a37c4fe..82011674 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -118,7 +118,7 @@ type MigrateCmd struct { LockBackoffMax time.Duration `help:"Maximum exponential backoff between lock-timeout attempts." default:"1s"` FreeDisk byteSize `help:"Free space on the volume the table lives on, for a statement the planner routes to copy-and-swap; the preflight refuses the run when the shadow table and its indexes would not fit (binary units: B, KiB, MiB, GiB, TiB). Unset means no space and refuses." name:"free-disk" placeholder:"SIZE"` KeepOld bool `help:"After a copy-and-swap, keep the retained source table under its _old name for a rollback window you end by hand, instead of dropping it before the run reports done." name:"keep-old"` - Quiesced string `help:"Take the table offline for writes for the length of a copy-and-swap: the run fences writes instead of streaming them through a replication slot, and needs no logical decoding from the server, but every write that reaches the fence fails with SQLSTATE 55006 until the swap. The value is the typed acknowledgement: it must name the resolved schema-qualified target table exactly." name:"quiesced" placeholder:"SCHEMA.TABLE"` + BlockWrites string `help:"Offline mode: block the table's writes for the length of a copy-and-swap: the run fences writes instead of streaming them through a replication slot, and needs no logical decoding from the server, but every write that reaches the fence fails with SQLSTATE 55006 until the swap. The value is the typed acknowledgement: it must name the resolved schema-qualified target table exactly." name:"block-writes" placeholder:"SCHEMA.TABLE"` Restart string `help:"Start a copy-and-swap over instead of resuming the leftover of an earlier run: the leftover shadow table, checkpoint row, and write fence or replication slot are discarded under the run's lock, then the statement runs from the beginning. The value is the typed acknowledgement: it must name the resolved schema-qualified target table exactly. A leftover whose swap already committed is not discarded; run the statement without this flag to finish it. Cannot be combined with --force, --accept-blocking, or --dry-run." placeholder:"SCHEMA.TABLE"` DryRun bool `help:"Classify and route the statement, print the plan, and execute nothing."` JSON bool `help:"Emit the verdict (or dry-run plan) as JSON."` @@ -170,7 +170,7 @@ func (c *MigrateCmd) Validate(kctx *kong.Context) error { return errors.New("--accept-blocking requires --statement-timeout on the same command line: it is the outage bound you accept after the lock is granted, and the default is not an acknowledgement") } if c.DryRun { - for _, name := range []string{"free-disk", "keep-old", "quiesced"} { + for _, name := range []string{"free-disk", "keep-old", "block-writes"} { if flagSupplied(kctx, name) { return fmt.Errorf("--%s cannot be combined with --dry-run: the dry run executes nothing, so a copy-and-swap policy has nothing to govern", name) } diff --git a/internal/cli/migrate.go b/internal/cli/migrate.go index b7025a00..974c835a 100644 --- a/internal/cli/migrate.go +++ b/internal/cli/migrate.go @@ -94,7 +94,7 @@ func (c *MigrateCmd) options(logger *slog.Logger) migrate.Options { Connection: c.Config(), FreeDiskBytes: int64(c.FreeDisk), KeepOld: c.KeepOld, - Quiesced: c.Quiesced, + BlockWrites: c.BlockWrites, Restart: c.Restart, }, } diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index 18519061..b143d6c5 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -294,7 +294,7 @@ func TestMigrateRunsTypeRewriteViaCopyAndSwap(t *testing.T) { cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) cmd.FreeDisk = byteSize(testutil.UnlimitedDisk) - cmd.Quiesced = schema + ".t" + cmd.BlockWrites = schema + ".t" cmd.JSON = true var out strings.Builder require.NoError(t, cmd.run(t.Context(), &out)) @@ -334,7 +334,7 @@ func TestMigrateCopyAndSwapRefusesWithoutFreeDisk(t *testing.T) { require.NoError(t, err) cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) - cmd.Quiesced = schema + ".t" + cmd.BlockWrites = schema + ".t" cmd.JSON = true var out strings.Builder err = cmd.run(t.Context(), &out) diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index 082cbb37..6dadda57 100644 --- a/internal/cli/migrate_test.go +++ b/internal/cli/migrate_test.go @@ -106,26 +106,25 @@ func TestAcceptBlockingFlagGrammar(t *testing.T) { // executes nothing for them to govern. func TestCopySwapFlagsWireIntoEnginePolicy(t *testing.T) { t.Run("wires connection, free space, and choices", func(t *testing.T) { - c := parseMigrate(t, "--free-disk", "20GiB", "--keep-old", "--quiesced", "app.orders") + c := parseMigrate(t, "--free-disk", "20GiB", "--keep-old", "--block-writes", "app.orders") got := c.options(nil).CopySwap assert.Equal(t, c.Config(), got.Connection) assert.Equal(t, int64(20<<30), got.FreeDiskBytes) assert.True(t, got.KeepOld) - assert.Equal(t, "app.orders", got.Quiesced) + assert.Equal(t, "app.orders", got.BlockWrites) }) t.Run("defaults to no free space and the live path", func(t *testing.T) { got := parseMigrate(t).options(nil).CopySwap assert.Zero(t, got.FreeDiskBytes) assert.False(t, got.KeepOld) - assert.Empty(t, got.Quiesced) + assert.Empty(t, got.BlockWrites) }) - // Taking a table offline for writes needs the table named: a bare - // --quiesced, as a script written for the old boolean flag passes it, - // is a parse error rather than a fence on whatever table the statement - // names. - t.Run("rejects --quiesced without the table", func(t *testing.T) { + // Blocking a table's writes needs the table named: a bare + // --block-writes is a parse error rather than a fence on whatever table + // the statement names. + t.Run("rejects --block-writes without the table", func(t *testing.T) { c := New("test") k, err := kong.New(c, kong.Vars{"version": "test"}) require.NoError(t, err) @@ -133,13 +132,13 @@ func TestCopySwapFlagsWireIntoEnginePolicy(t *testing.T) { "migrate", "--url", "postgres://user@localhost:5432/app", "--alter", "ALTER TABLE t ALTER COLUMN c TYPE bigint", - "--quiesced", + "--block-writes", }) require.Error(t, err) - assert.Contains(t, err.Error(), "--quiesced") + assert.Contains(t, err.Error(), "--block-writes") }) - for _, flag := range []string{"--free-disk=1GiB", "--keep-old", "--quiesced=public.t"} { + for _, flag := range []string{"--free-disk=1GiB", "--keep-old", "--block-writes=public.t"} { t.Run("rejects "+flag+" with --dry-run", func(t *testing.T) { c := New("test") k, err := kong.New(c, kong.Vars{"version": "test"}) diff --git a/internal/cli/restart_integration_test.go b/internal/cli/restart_integration_test.go index 60e876f2..54492d3b 100644 --- a/internal/cli/restart_integration_test.go +++ b/internal/cli/restart_integration_test.go @@ -42,7 +42,7 @@ func TestMigrateRestartDiscardsALeftoverAndRunsTheChange(t *testing.T) { alter := fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema) plain := newMigrateCmd(url, alter) plain.FreeDisk = byteSize(testutil.UnlimitedDisk) - plain.Quiesced = schema + ".t" + plain.BlockWrites = schema + ".t" plain.JSON = true var refusedOut strings.Builder require.ErrorIs(t, plain.run(t.Context(), &refusedOut), verdict.ErrRefused, "a refusal verdict exits with the refusal code") @@ -53,7 +53,7 @@ func TestMigrateRestartDiscardsALeftoverAndRunsTheChange(t *testing.T) { restart := newMigrateCmd(url, alter) restart.FreeDisk = byteSize(testutil.UnlimitedDisk) - restart.Quiesced = schema + ".t" + restart.BlockWrites = schema + ".t" restart.JSON = true restart.Restart = schema + ".t" var out strings.Builder @@ -100,7 +100,7 @@ func TestMigrateRestartAckMismatchExecutesNothing(t *testing.T) { cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) cmd.FreeDisk = byteSize(testutil.UnlimitedDisk) - cmd.Quiesced = schema + ".t" + cmd.BlockWrites = schema + ".t" cmd.Restart = "wrong.table" var out strings.Builder err = cmd.run(t.Context(), &out) diff --git a/internal/cli/signal_integration_test.go b/internal/cli/signal_integration_test.go index b838d312..563f16c9 100644 --- a/internal/cli/signal_integration_test.go +++ b/internal/cli/signal_integration_test.go @@ -75,7 +75,7 @@ func TestMigrateSignalDuringCopyReportsCancelledByCaller(t *testing.T) { cmd := newMigrateCmd(url, fmt.Sprintf("ALTER TABLE %s.t ALTER COLUMN id TYPE bigint", schema)) cmd.FreeDisk = byteSize(testutil.UnlimitedDisk) - cmd.Quiesced = schema + ".t" + cmd.BlockWrites = schema + ".t" cmd.JSON = true var out strings.Builder err = cmd.run(ctx, &out) diff --git a/pkg/capabilities/front_door_integration_test.go b/pkg/capabilities/front_door_integration_test.go index 6b5e7c17..9c04ea0b 100644 --- a/pkg/capabilities/front_door_integration_test.go +++ b/pkg/capabilities/front_door_integration_test.go @@ -259,7 +259,7 @@ func TestMigrateSupportedRowsExecute(t *testing.T) { st, err := statement.ParseOne(sql) require.NoError(t, err) if claim.copyAndSwap { - opts.CopySwap.Quiesced = st.Schema() + "." + st.Table() + opts.CopySwap.BlockWrites = st.Schema() + "." + st.Table() } v, err := migrate.Run(t.Context(), pool, st, opts) require.NoError(t, err, sql) diff --git a/pkg/migrate/copy_swap.go b/pkg/migrate/copy_swap.go index 0175450f..21d60494 100644 --- a/pkg/migrate/copy_swap.go +++ b/pkg/migrate/copy_swap.go @@ -65,7 +65,7 @@ type CopySwapOptions struct { // drops it before the run reports done. KeepOld bool - // Quiesced, when set, declares that nothing writes the table while the + // BlockWrites, when set, declares that nothing writes the table while the // run lasts. The run then takes the write fence instead of a // replication slot and needs no logical decoding from the server; every // write that reaches the fence fails, so the table is offline for @@ -73,7 +73,7 @@ type CopySwapOptions struct { // SCHEMA.TABLE, the same way Restart does, so the declaration cannot be // left on a script and fence another table by accident. Empty runs the // live path: slot, stream, and catch-up. - Quiesced string + BlockWrites string // Restart, when set, is the operator's acknowledgement that the run on // the table is to start over: whatever an earlier run left — its @@ -175,7 +175,7 @@ func (o CopySwapOptions) runOptions(logger *slog.Logger) schemachange.RunOptions // needs none of the server's logical-decoding facilities. func (o CopySwapOptions) environment() preflight.CopySwapEnvironment { return preflight.CopySwapEnvironment{ - LogicalDecoding: o.Quiesced == "", + LogicalDecoding: o.BlockWrites == "", FreeDiskBytes: o.FreeDiskBytes, } } @@ -213,9 +213,9 @@ func copyAndSwap(ctx context.Context, pool *pgxpool.Pool, st statement.Statement logger.Debug("copy-and-swap preflight passed", "decodes_wal", target.DecodesWAL(), "keep_old", opts.CopySwap.KeepOld) - quiesced := opts.CopySwap.Quiesced != "" + quiesced := opts.CopySwap.BlockWrites != "" if quiesced { - auditQuiesced(opts.audit(), st) + auditBlockWrites(opts.audit(), st) } result, err := runner.Run(ctx, pool, target, st) return copySwapOutcome(st, quiesced, result, err) @@ -399,7 +399,7 @@ func copySwapDetail(result schemachange.Result, quiesced bool) string { s += "; the run reconciled a lost replication slot before catching up" } if quiesced { - s += "; the table was quiesced: writes were fenced for the length of the run" + s += "; writes to the table were blocked for the length of the run" } return s } diff --git a/pkg/migrate/copy_swap_quiesced.go b/pkg/migrate/copy_swap_block_writes.go similarity index 53% rename from pkg/migrate/copy_swap_quiesced.go rename to pkg/migrate/copy_swap_block_writes.go index 0aebe4eb..9d2bc49c 100644 --- a/pkg/migrate/copy_swap_quiesced.go +++ b/pkg/migrate/copy_swap_block_writes.go @@ -8,25 +8,25 @@ import ( "github.com/block/pg-sprite/pkg/statement" ) -// checkQuiescedAck confirms that the quiesced acknowledgement names the +// checkBlockWritesAck confirms that the block-writes acknowledgement names the // resolved target table, exactly as the force and restart acknowledgements -// must. A quiesced run refuses every write to the table for its whole +// must. A block-writes run refuses every write to the table for its whole // length, so a typed table name is what stops the declaration from being // left on a script and taking the next table offline by accident. -func checkQuiescedAck(st statement.Statement, ack string) error { +func checkBlockWritesAck(st statement.Statement, ack string) error { if ack == qualified(st) { return nil } - return fmt.Errorf("the quiesced acknowledgement must name the resolved target table %q, got %q; nothing was executed", + return fmt.Errorf("the block-writes acknowledgement must name the resolved target table %q, got %q; nothing was executed", qualified(st), ack) } -// auditQuiesced records, before the fence goes on, that the operator took +// auditBlockWrites records, before the fence goes on, that the operator took // the table offline for writes. The record is warn-level and // unconditional, as the force and restart audits are: an audit trail must // not depend on diagnostics being enabled. -func auditQuiesced(audit *slog.Logger, st statement.Statement) { - audit.Warn("quiesced run acknowledged; every write to the table will be refused until the swap", +func auditBlockWrites(audit *slog.Logger, st statement.Statement) { + audit.Warn("block-writes run acknowledged; every write to the table will be refused until the swap", "table", qualified(st), "sqlstate", schemachange.WriteFenceSQLState) } diff --git a/pkg/migrate/copy_swap_quiesced_integration_test.go b/pkg/migrate/copy_swap_block_writes_integration_test.go similarity index 76% rename from pkg/migrate/copy_swap_quiesced_integration_test.go rename to pkg/migrate/copy_swap_block_writes_integration_test.go index b67be247..3b66b18d 100644 --- a/pkg/migrate/copy_swap_quiesced_integration_test.go +++ b/pkg/migrate/copy_swap_block_writes_integration_test.go @@ -11,22 +11,22 @@ import ( "github.com/block/pg-sprite/pkg/verdict" ) -// A quiesced acknowledgement that names another table is an error before +// A block-writes acknowledgement that names another table is an error before // any verdict, and the table is never taken offline: no shadow is built, // no fence goes on, so the source still takes writes, and its shape is -// unchanged. The acknowledgement is what stops a quiesced declaration left +// unchanged. The acknowledgement is what stops a block-writes declaration left // on a script from fencing the next table it meets. -func TestRunRejectsAMismatchedQuiescedAck(t *testing.T) { +func TestRunRejectsAMismatchedBlockWritesAck(t *testing.T) { f := newCopySwapFixture(t) f.createOrders(t) st, err := statement.ParseOne(f.inSchema(`ALTER TABLE {schema}.orders ALTER COLUMN qty TYPE bigint`)) require.NoError(t, err) opts := f.options() - opts.CopySwap.Quiesced = f.schema + ".customers" + opts.CopySwap.BlockWrites = f.schema + ".customers" v, err := Run(t.Context(), f.pool, st, opts) require.Error(t, err) - assert.Contains(t, err.Error(), "quiesced acknowledgement") + assert.Contains(t, err.Error(), "block-writes acknowledgement") assert.Equal(t, verdict.Verdict{}, v, "an error before a verdict carries a zero verdict") assert.False(t, f.relationExists(t, schemachange.ShadowName(f.schema, "orders")), "no shadow was built") f.exec(t, `INSERT INTO {schema}.orders (qty, note, sku) VALUES (1, 'after', 'sku1')`) diff --git a/pkg/migrate/copy_swap_integration_test.go b/pkg/migrate/copy_swap_integration_test.go index c7f1fb69..d6432af5 100644 --- a/pkg/migrate/copy_swap_integration_test.go +++ b/pkg/migrate/copy_swap_integration_test.go @@ -82,7 +82,7 @@ func (f copySwapFixture) options() Options { opts.CopySwap = CopySwapOptions{ Connection: f.cfg, FreeDiskBytes: testutil.UnlimitedDisk, - Quiesced: f.schema + ".orders", + BlockWrites: f.schema + ".orders", } return opts } diff --git a/pkg/migrate/copy_swap_test.go b/pkg/migrate/copy_swap_test.go index 2e49fc48..36a5974a 100644 --- a/pkg/migrate/copy_swap_test.go +++ b/pkg/migrate/copy_swap_test.go @@ -311,5 +311,5 @@ func TestCopySwapDetail(t *testing.T) { OldTableDropped: true, }, true) assert.Equal(t, "verified 4 rows in 1 chunks; the retained source is dropped; "+ - "the table was quiesced: writes were fenced for the length of the run", quiesced) + "writes to the table were blocked for the length of the run", quiesced) } diff --git a/pkg/migrate/desired_copy_swap_integration_test.go b/pkg/migrate/desired_copy_swap_integration_test.go index 5f911a12..0f41228f 100644 --- a/pkg/migrate/desired_copy_swap_integration_test.go +++ b/pkg/migrate/desired_copy_swap_integration_test.go @@ -24,7 +24,7 @@ func copySwapRunOptions(serverURL, table string) migrate.Options { opts.CopySwap = migrate.CopySwapOptions{ Connection: dbconn.Config{URL: serverURL}, FreeDiskBytes: testutil.UnlimitedDisk, - Quiesced: table, + BlockWrites: table, } return opts } diff --git a/pkg/migrate/migrate.go b/pkg/migrate/migrate.go index d4a64b9b..1d4b3722 100644 --- a/pkg/migrate/migrate.go +++ b/pkg/migrate/migrate.go @@ -256,8 +256,8 @@ func run(ctx context.Context, pool *pgxpool.Pool, st statement.Statement, opts O return verdict.Verdict{}, err } } - if opts.CopySwap.Quiesced != "" { - if err := checkQuiescedAck(st, opts.CopySwap.Quiesced); err != nil { + if opts.CopySwap.BlockWrites != "" { + if err := checkBlockWritesAck(st, opts.CopySwap.BlockWrites); err != nil { return verdict.Verdict{}, err } } diff --git a/pkg/migrate/rename_index.go b/pkg/migrate/rename_index.go index 872705e2..59e3d4dd 100644 --- a/pkg/migrate/rename_index.go +++ b/pkg/migrate/rename_index.go @@ -90,8 +90,8 @@ func renameIndex(ctx context.Context, pool *pgxpool.Pool, st statement.Statement return verdict.Verdict{}, fmt.Errorf("%w: an index rename never runs through copy-and-swap; nothing was executed", ErrIndexRenameAcknowledgement) } - if opts.CopySwap.Quiesced != "" { - return verdict.Verdict{}, fmt.Errorf("%w: quiesced names a table to fence, and an index rename never runs through copy-and-swap; nothing was executed", + if opts.CopySwap.BlockWrites != "" { + return verdict.Verdict{}, fmt.Errorf("%w: block-writes names a table to fence, and an index rename never runs through copy-and-swap; nothing was executed", ErrIndexRenameAcknowledgement) } facts, err := LiveFacts(ctx, pool, st) From 8e0a1c33f3842913ce7e4e63769bd4c23b0594e2 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 28 Oct 2026 14:39:15 +1100 Subject: [PATCH 3/3] migrate: cover the block-writes acknowledgement in a plan and on rename A desired-state plan with a mismatched acknowledgement runs neither its native statement nor its retype, and an index rename refuses it. The resume row in the SchemaBot guide now says a run that blocked writes resumes only with the same acknowledgement. --- docs/schemabot-integration.md | 2 +- internal/cli/migrate_integration_test.go | 2 +- internal/cli/restart_integration_test.go | 4 +- .../desired_copy_swap_integration_test.go | 43 ++++++++++++++++++- pkg/migrate/rename_index_integration_test.go | 15 +++++++ 5 files changed, 61 insertions(+), 5 deletions(-) diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 751588e3..1075dbac 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -678,7 +678,7 @@ CLI and through `migrate.Run`, plus one the Runner performs on its own; an orche | --- | --- | --- | | **Start** a change | `pg-sprite migrate --url … 'ALTER TABLE s.t …'` (with `--free-disk`, and `--block-writes SCHEMA.TABLE` to take the table offline for writes instead of streaming them) | Routes the statement; on the copy-and-swap backend it takes the table's lock session, writes the checkpoint row, builds and fills the shadow, verifies, swaps, and prints one verdict | | **Stop** a running change | Send `SIGINT` (Ctrl-C) or `SIGTERM` once | The run's context is cancelled and the run unwinds through its own stop path: a copy in flight saves the watermark its committed chunks reached, the table lock session is released, the row stays at its last save with the run's hold on the source — the write fence of a quiesced run, or the replication slot of a live run, which retains WAL until the run is finished or discarded — and the verdict is `failed` with `code: cancelled-by-caller` (exit 1). A second `SIGINT`/`SIGTERM` ends the process at once with exit 130, before the stop path completes; the row then records whatever the last save landed, and the lock session is released by the server when its connection drops | -| **Resume** a stopped change | Run the same statement again | The checkpoint row says where to continue: a stopped copy continues from its watermark, a run past its copy verifies again, a `cutover` row is answered by the catalog (a committed swap finishes from the old-table drop; a rolled-back swap goes back through verify and gate). Nothing is copied twice. A different statement on the same table is refused (`resume-row-mismatch`) | +| **Resume** a stopped change | Run the same statement again, with the same `--block-writes SCHEMA.TABLE` if the stopped run blocked writes | The checkpoint row says where to continue: a stopped copy continues from its watermark, a run past its copy verifies again, a `cutover` row is answered by the catalog (a committed swap finishes from the old-table drop; a rolled-back swap goes back through verify and gate). Nothing is copied twice. A different statement on the same table is refused (`resume-row-mismatch`). A stopped run that blocked writes resumes only with the acknowledgement, and a stopped live run only without it; a run of the other kind is refused (`resume-row-decoding`) and leaves the stopped run's hold as it is | | **Start over** — discard a stopped change and run it from the beginning | Run the same statement with `--restart SCHEMA.TABLE`, where the value names the resolved target exactly | Under the run's own table lock session, discards what the stopped run left in the order a later restart could meet idempotently — the shadow, then the hold on the source (`DropWriteFence` for a quiesced row, `decode.DropSlot` for a live row), then the checkpoint row — and runs the change fresh. A leftover shadow that no row records is discarded the same way. The decision is written to the audit log before anything is discarded | | **Abandon** a change the data refuses | Nothing — the Runner does it | A run whose fill or swap stops on a data fault (a `*pgconn.PgError` in SQLSTATE class `22` or `23`: a value the new type will not hold, a constraint the copied rows break) would stop the same way on every resume, so the Runner discards it the way `--restart` would — shadow, hold on the source, checkpoint row — under the lock session it still holds, and returns `*schemachange.AbandonedError` wrapping the server's error. The verdict is `failed` with `code: data-fault` and nothing is left on the source. Correct the data or the statement, then run it again; the run starts from the beginning | diff --git a/internal/cli/migrate_integration_test.go b/internal/cli/migrate_integration_test.go index b143d6c5..244c01ea 100644 --- a/internal/cli/migrate_integration_test.go +++ b/internal/cli/migrate_integration_test.go @@ -274,7 +274,7 @@ func TestMigrateRenameColumnRefusedUnlessForced(t *testing.T) { // A type rewrite routes to the copy-and-swap backend and runs end to end // from this front door: the shadow table takes the new type, every row is // carried, the retained source is dropped, and the one verdict says so. -// --quiesced keeps the run to the write fence, so the test server needs no +// --block-writes keeps the run to the write fence, so the test server needs no // logical decoding. func TestMigrateRunsTypeRewriteViaCopyAndSwap(t *testing.T) { url := testutil.StartPostgres(t) diff --git a/internal/cli/restart_integration_test.go b/internal/cli/restart_integration_test.go index 54492d3b..ea11d496 100644 --- a/internal/cli/restart_integration_test.go +++ b/internal/cli/restart_integration_test.go @@ -18,8 +18,8 @@ import ( // --restart from the front door: a relation under the shadow's name that // no checkpoint row records is refused by a plain run, and discarded by a // run that carries the acknowledgement — which then carries the change to -// its copy-and-swap verdict on a shadow of its own. --quiesced keeps the -// run to the write fence, so the test server needs no logical decoding. +// its copy-and-swap verdict on a shadow of its own. --block-writes keeps +// the run to the write fence, so the test server needs no logical decoding. func TestMigrateRestartDiscardsALeftoverAndRunsTheChange(t *testing.T) { url := testutil.StartPostgres(t) pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) diff --git a/pkg/migrate/desired_copy_swap_integration_test.go b/pkg/migrate/desired_copy_swap_integration_test.go index 0f41228f..26f59446 100644 --- a/pkg/migrate/desired_copy_swap_integration_test.go +++ b/pkg/migrate/desired_copy_swap_integration_test.go @@ -17,7 +17,7 @@ import ( // copySwapRunOptions are runOptions with the copy-and-swap route wired the // way the CLI wires it: the route's own connection, free disk the shadow -// will fit in, and the quiesced acknowledgement for table (SCHEMA.TABLE) so +// will fit in, and the block-writes acknowledgement for table (SCHEMA.TABLE) so // the test server needs no logical decoding. func copySwapRunOptions(serverURL, table string) migrate.Options { opts := runOptions() @@ -90,6 +90,47 @@ func TestRunDesiredCopyAndSwap(t *testing.T) { assert.Equal(t, "integer", columnType(t, schema, "owner_id")) }) + t.Run("a block-writes acknowledgement naming another table runs nothing", func(t *testing.T) { + schema := testutil.NewSchema(t, pool) + _, err := pool.Exec(t.Context(), fmt.Sprintf(` + CREATE TABLE %s.t ( + id int PRIMARY KEY, + owner_id int NOT NULL + )`, schema)) + require.NoError(t, err) + + // The same plan as above, a native ADD COLUMN ahead of a retype the + // router sends through copy-and-swap, with every gate open except + // the acknowledgement, which names another table. Each planned + // statement checks it before anything runs, so the first one, the + // native ADD COLUMN, already stops the plan: neither statement runs + // and no table's writes are blocked. + opts := copySwapRunOptions(serverURL, schema+".other") + res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{ + Schema: schema, + Desired: parseDesired(t, `CREATE TABLE t ( + id int PRIMARY KEY, + owner_id bigint NOT NULL, + note text + )`), + }, opts) + require.Error(t, err) + assert.Contains(t, err.Error(), "block-writes acknowledgement") + assert.Equal(t, verdict.OutcomeFailed, res.Outcome) + assert.Empty(t, res.Verdicts, "no statement reached a verdict") + require.Len(t, res.Plan.Statements, 2, "the plan carried the native statement and the retype") + assert.Equal(t, router.BackendCopyAndSwap, res.Plan.Statements[1].Backend) + + var columns int + require.NoError(t, pool.QueryRow(t.Context(), + `SELECT count(*) FROM information_schema.columns + WHERE table_schema = $1 AND table_name = 't'`, schema).Scan(&columns)) + assert.Equal(t, 2, columns, "the native ADD COLUMN did not commit") + assert.Equal(t, "integer", columnType(t, schema, "owner_id"), "the retype did not run") + _, err = pool.Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s.t (id, owner_id) VALUES (1, 1)`, schema)) + assert.NoError(t, err, "the table still takes writes") + }) + t.Run("two retypes of one table converge and re-run as a no-op", func(t *testing.T) { schema := testutil.NewSchema(t, pool) _, err := pool.Exec(t.Context(), fmt.Sprintf(` diff --git a/pkg/migrate/rename_index_integration_test.go b/pkg/migrate/rename_index_integration_test.go index c661b172..35f8b617 100644 --- a/pkg/migrate/rename_index_integration_test.go +++ b/pkg/migrate/rename_index_integration_test.go @@ -127,3 +127,18 @@ func TestRenameIndexRejectsTheForceAcknowledgement(t *testing.T) { require.ErrorIs(t, err, migrate.ErrIndexRenameAcknowledgement) assert.True(t, relationNamed(t, pool, schema, "users_email_idx"), "nothing must have executed") } + +// --block-writes names a table whose writes a copy-and-swap blocks, and an +// index rename never runs through copy-and-swap: even naming the index's +// own table, the acknowledgement has nothing to apply to, so it is a usage +// error and nothing executes. +func TestRenameIndexRejectsTheBlockWritesAcknowledgement(t *testing.T) { + pool, schema, _ := renameIndexFixture(t) + sql := fmt.Sprintf(`ALTER INDEX %s.users_email_idx RENAME TO users_contact_email_idx`, schema) + opts := runOptions() + opts.CopySwap.BlockWrites = schema + ".users" + + _, err := migrate.Run(t.Context(), pool, parseOne(t, sql), opts) + require.ErrorIs(t, err, migrate.ErrIndexRenameAcknowledgement) + assert.True(t, relationNamed(t, pool, schema, "users_email_idx"), "nothing must have executed") +}