diff --git a/docs/copy-and-swap.md b/docs/copy-and-swap.md index 103f83f0..590f3e93 100644 --- a/docs/copy-and-swap.md +++ b/docs/copy-and-swap.md @@ -40,7 +40,9 @@ logical decoding enabled. `--block-writes` is your declaration that nothing needs to write the table. Its value must name the table the statement changes, exactly as `SCHEMA.TABLE`; a value naming any other table stops the run before anything executes, so a flag left in a script cannot take the next -table offline. pg-sprite does not take the declaration on trust: the fence refuses any write +table offline. It applies to `migrate --alter` only: `migrate --desired` rejects it, because +there the derived plan, not you, decides how many statements take a copy and so how long +writes fail. Run each statement with `migrate --alter --block-writes` instead. pg-sprite does not take the declaration on trust: the fence refuses any write that does arrive, so nothing is lost, but the application sees errors for the whole run. The fence and why it also covers replication apply workers are described in [engine-role.md](engine-role.md#the-write-fence). diff --git a/docs/limitations.md b/docs/limitations.md index d617d6d7..9c31f7c1 100644 --- a/docs/limitations.md +++ b/docs/limitations.md @@ -163,7 +163,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 once `--allow-destructive` names the table — 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 `--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. | +| 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 (desired-state execution rejects `--block-writes`, so it never takes the table offline for writes), 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 e8334918..711cf370 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -463,7 +463,9 @@ its `_old` name, and whether the run blocks the table's writes — offline mode, 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: +fence goes on. `RunDesired` rejects `BlockWrites` (`ErrBlockWritesNotSupported`): the derived +plan, not the operator, would decide how many fenced copies run, so a value left on the +adapter's configuration cannot take the table offline for a later plan. 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 diff --git a/internal/cli/cli.go b/internal/cli/cli.go index f7fa9e43..cf17bda8 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -119,7 +119,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"` - 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"` + 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. Cannot be combined with --desired." 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."` diff --git a/internal/cli/migrate_desired.go b/internal/cli/migrate_desired.go index a0d010f6..f86c7831 100644 --- a/internal/cli/migrate_desired.go +++ b/internal/cli/migrate_desired.go @@ -33,6 +33,9 @@ func (c *MigrateCmd) validateInput() error { if c.AllowDestructive != "" { return errors.New("--allow-destructive cannot be combined with --desired: desired-state execution refuses a destructive plan whole") } + if c.BlockWrites != "" { + return errors.New("--block-writes cannot be combined with --desired: the plan, not the operator, decides how many copies would block the table's writes; run each statement with migrate --alter --block-writes instead") + } } return nil } diff --git a/internal/cli/migrate_test.go b/internal/cli/migrate_test.go index 6dadda57..06a8ea50 100644 --- a/internal/cli/migrate_test.go +++ b/internal/cli/migrate_test.go @@ -138,6 +138,16 @@ func TestCopySwapFlagsWireIntoEnginePolicy(t *testing.T) { assert.Contains(t, err.Error(), "--block-writes") }) + // The declarative front door never takes a table offline for writes: + // the derived plan, not the operator, would decide how many fenced + // copies run, so the acknowledgement is a usage error there. + t.Run("rejects --block-writes with --desired", func(t *testing.T) { + c := &MigrateCmd{Desired: "orders.sql", BlockWrites: "public.orders"} + err := c.validateInput() + require.Error(t, err) + assert.Contains(t, err.Error(), "--block-writes cannot be combined with --desired") + }) + 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") diff --git a/pkg/migrate/copy_swap_block_writes.go b/pkg/migrate/copy_swap_block_writes.go index 4becca09..4eb4f5e4 100644 --- a/pkg/migrate/copy_swap_block_writes.go +++ b/pkg/migrate/copy_swap_block_writes.go @@ -64,6 +64,17 @@ func fenceAfterStop(ctx context.Context, store *checkpoint.Store, st statement.S } } +// ErrBlockWritesNotSupported is returned by [RunDesired] when +// Options.CopySwap.BlockWrites is set. The acknowledgement names a table, +// but on the declarative front door the derived plan, not the operator, +// decides which statements take copy-and-swap and so how many fenced copies +// refuse the table's writes and for how long; a value left on an embedder's +// configuration would take the table offline for every later plan. Offline +// mode belongs on the imperative front door, one statement the operator +// wrote at a time. +var ErrBlockWritesNotSupported = errors.New( + "the block-writes acknowledgement applies to the imperative front door only; desired-state execution never takes a table offline for writes") + // checkBlockWritesAck confirms that the block-writes acknowledgement names the // resolved target table, exactly as the force and restart acknowledgements // must. A block-writes run refuses every write to the table for its whole diff --git a/pkg/migrate/copy_swap_connection_integration_test.go b/pkg/migrate/copy_swap_connection_integration_test.go index 60f71a60..32d195c5 100644 --- a/pkg/migrate/copy_swap_connection_integration_test.go +++ b/pkg/migrate/copy_swap_connection_integration_test.go @@ -127,7 +127,7 @@ func TestRunDesiredRefusesAConnectionOnAnotherDatabase(t *testing.T) { qty integer NOT NULL, note text )`), - }, copySwapRunOptions(d.otherURL, schema+".orders")) + }, desiredCopySwapOptions(d.otherURL)) var mismatch *migrate.CopySwapConnectionMismatchError require.ErrorAs(t, err, &mismatch) assert.Equal(t, mismatch.PoolCluster, mismatch.ConnectionCluster, "both databases are on the test server") @@ -261,6 +261,10 @@ func assertStopsBeforeAnyStatement(t *testing.T, d twoDatabases, runPool *pgxpoo assert.NotErrorAs(t, err, &mismatch) assert.Zero(t, v, "nothing was executed, so there is no verdict") + // Desired-state execution rejects the block-writes acknowledgement, so + // the desired run takes the same wiring without it. + desiredOpts := opts + desiredOpts.CopySwap.BlockWrites = "" res, err := migrate.RunDesired(t.Context(), runPool, migrate.DesiredRequest{ Schema: schema, Desired: parseDesired(t, `CREATE TABLE orders ( @@ -268,7 +272,7 @@ func assertStopsBeforeAnyStatement(t *testing.T, d twoDatabases, runPool *pgxpoo qty integer NOT NULL, note text )`), - }, opts) + }, desiredOpts) require.Error(t, err) assert.NotErrorAs(t, err, &mismatch) assert.Equal(t, verdict.OutcomeFailed, res.Outcome) diff --git a/pkg/migrate/desired.go b/pkg/migrate/desired.go index 8421b93b..15e38e39 100644 --- a/pkg/migrate/desired.go +++ b/pkg/migrate/desired.go @@ -183,9 +183,10 @@ func (r DesiredResult) refused(ref verdict.Refusal, detail string) DesiredResult // the operational error. An error with a zero result means the pipeline // stopped before planning or executing anything. // -// Options.Force and Options.AllowDestructive are rejected: the declarative -// front door never runs a submitted form blind and refuses a destructive -// plan whole. A destructive or force-worthy change belongs on the +// Options.Force, Options.AllowDestructive and Options.CopySwap.BlockWrites +// are rejected: the declarative front door never runs a submitted form +// blind, refuses a destructive plan whole, and never takes a table offline +// for writes. A destructive, force-worthy or offline change belongs on the // imperative front door where the operator states it explicitly. // // Every planned statement runs back through [Run], so Run's note on the @@ -202,6 +203,9 @@ func RunDesired(ctx context.Context, pool *pgxpool.Pool, req DesiredRequest, opt if opts.AllowDestructive != "" { return DesiredResult{}, ErrAllowDestructiveNotSupported } + if opts.CopySwap.BlockWrites != "" { + return DesiredResult{}, ErrBlockWritesNotSupported + } report, err := diffplan.Plan(ctx, pool, diffplan.Request{Schema: req.Schema, Desired: req.Desired}) if ref, ok := desiredShapeRefusal(err); ok { return DesiredResult{}.refused(ref, fmt.Sprintf("%v; nothing was planned or executed", err)), nil diff --git a/pkg/migrate/desired_copy_swap_integration_test.go b/pkg/migrate/desired_copy_swap_integration_test.go index 26f59446..89c45d40 100644 --- a/pkg/migrate/desired_copy_swap_integration_test.go +++ b/pkg/migrate/desired_copy_swap_integration_test.go @@ -29,13 +29,20 @@ func copySwapRunOptions(serverURL, table string) migrate.Options { return opts } +// desiredCopySwapOptions are copySwapRunOptions without the block-writes +// acknowledgement, which desired-state execution rejects: the route takes +// the live path, so a plan that reaches a copy needs logical decoding. +func desiredCopySwapOptions(serverURL string) migrate.Options { + return copySwapRunOptions(serverURL, "") +} + // Desired-state execution runs a plan through the copy-and-swap backend as // a unit: a plan whose preflight would refuse is refused whole before the // native statements ahead of the retype commit, and a plan that retypes two // columns of one table runs both — the second run replaces the done row the // first left behind — and converges. func TestRunDesiredCopyAndSwap(t *testing.T) { - serverURL := testutil.StartPostgres(t) + serverURL := testutil.StartPostgresWithSettings(t, "wal_level=logical", "wal_sender_timeout=2s") pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: serverURL}) require.NoError(t, err) defer pool.Close() @@ -62,7 +69,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, schema+".t") + opts := desiredCopySwapOptions(serverURL) opts.CopySwap.FreeDiskBytes = 0 res, err := migrate.RunDesired(t.Context(), pool, migrate.DesiredRequest{ Schema: schema, @@ -90,47 +97,6 @@ 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(` @@ -155,7 +121,7 @@ func TestRunDesiredCopyAndSwap(t *testing.T) { body varchar(10) )`), } - res, err := migrate.RunDesired(t.Context(), pool, req, copySwapRunOptions(serverURL, schema+".t")) + res, err := migrate.RunDesired(t.Context(), pool, req, desiredCopySwapOptions(serverURL)) require.NoError(t, err) assert.Equal(t, verdict.OutcomeExecuted, res.Outcome) require.Len(t, res.Plan.Statements, 2, "one retype per column") @@ -171,7 +137,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, schema+".t")) + res, err = migrate.RunDesired(t.Context(), pool, req, desiredCopySwapOptions(serverURL)) 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/desired_test.go b/pkg/migrate/desired_test.go index 162b8e56..906f5a15 100644 --- a/pkg/migrate/desired_test.go +++ b/pkg/migrate/desired_test.go @@ -39,6 +39,18 @@ func TestRunDesiredRejectsForce(t *testing.T) { assert.Equal(t, DesiredResult{}, res) } +// Desired-state execution never takes a table offline for writes: the +// derived plan, not the operator, would decide how many fenced copies run, +// so the acknowledgement is rejected as a typed sentinel before any +// database work, even when it names the right table. +func TestRunDesiredRejectsBlockWrites(t *testing.T) { + opts := DefaultOptions() + opts.CopySwap.BlockWrites = "public.t" + res, err := RunDesired(t.Context(), nil, DesiredRequest{Schema: "public"}, opts) + require.ErrorIs(t, err, ErrBlockWritesNotSupported) + assert.Equal(t, DesiredResult{}, res) +} + // A greenfield create-shape refusal explains itself through the typed // cause the executor stamped on the statement; a refusal without one // falls back to the generic wording rather than inventing a cause.