Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion docs/copy-and-swap.md
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down
2 changes: 1 addition & 1 deletion docs/limitations.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion docs/schemabot-integration.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion internal/cli/cli.go
Original file line number Diff line number Diff line change
Expand Up @@ -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."`
Expand Down
3 changes: 3 additions & 0 deletions internal/cli/migrate_desired.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down
10 changes: 10 additions & 0 deletions internal/cli/migrate_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
11 changes: 11 additions & 0 deletions pkg/migrate/copy_swap_block_writes.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 6 additions & 2 deletions pkg/migrate/copy_swap_connection_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -261,14 +261,18 @@ 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 (
id bigint PRIMARY KEY,
qty integer NOT NULL,
note text
)`),
}, opts)
}, desiredOpts)
require.Error(t, err)
assert.NotErrorAs(t, err, &mismatch)
assert.Equal(t, verdict.OutcomeFailed, res.Outcome)
Expand Down
10 changes: 7 additions & 3 deletions pkg/migrate/desired.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
56 changes: 11 additions & 45 deletions pkg/migrate/desired_copy_swap_integration_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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,
Expand Down Expand Up @@ -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(`
Expand All @@ -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")
Expand All @@ -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")
Expand Down
12 changes: 12 additions & 0 deletions pkg/migrate/desired_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading