From 41e2fb1b317470f939a277e37413e2a6dc54d78d Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Thu, 29 Oct 2026 14:37:26 +1100 Subject: [PATCH 1/2] copier, applier: a cancel with a cause still reads as a cancellation A caller that stopped a run through context.WithCancelCause got its cause back bare, so the verdict coded the stop execution-failed instead of cancelled-by-caller. The copier and the catch-up now report the cause wrapped together with ctx.Err(), so errors.Is answers for both. --- docs/copy-and-swap-design.md | 2 +- docs/schemabot-integration.md | 4 +- pkg/applier/catchup.go | 15 +++++- pkg/applier/catchup_test.go | 25 +++++++++- pkg/applier/pause.go | 2 +- pkg/copier/copier.go | 33 +++++++++++-- pkg/copier/copier_integration_test.go | 12 +++-- pkg/copier/copier_test.go | 38 +++++++++++++++ pkg/migrate/copy_swap_integration_test.go | 45 ++++++++++++++++++ .../runner_crash_integration_test.go | 12 +++-- .../runner_live_integration_test.go | 47 +++++++++++++++++++ 11 files changed, 219 insertions(+), 16 deletions(-) diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index f122a10a..b8700926 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -669,7 +669,7 @@ next pass's to catch. | `pkg/copier` | Produces `Chunk` and `Watermark`; `Chunker` (built only from a `CopySwapTarget`) cuts consecutive chunks that tile the whole int64 key space — first open below, last open above — so every key a row can carry belongs to exactly one chunk and a watermark at the largest value means the copy is complete. `Copier` (built from a `CopySwapTarget`, a `Shadow` — the shape `schemachange.BuiltShadow` satisfies — and the table's `TableLockSession`) copies chunks with several workers, each in its own bounded transaction under the owner's role that takes `ACCESS SHARE` on the source and `ROW EXCLUSIVE` on the shadow, confirms the lock, and confirms both relation OIDs before one frozen never-overwriting insert — the chunk statement `pkg/internal/chunksql` holds for the copier and the verifier's repair alike, and so does its `search_path`: the catalog alone for a copy without conversions, and for a copy with them the shadow build's own path (catalog first, then the source schema, then `public`), so a USING expression resolves its unqualified names in the copy as it did when the statement ran on the shadow — and the guard recomputes the `Shadow` proof's `ConversionFingerprint` (`chunksql.ResolutionFingerprint`: what those names resolve to across that path, function bodies, strictness, settings, and argument defaults included) inside each chunk transaction, before the insert and again after it before the commit, and refuses as `copy-conversion-drift` when it differs, so under `READ COMMITTED`, which every chunk transaction pins whatever the session default, a chunk that a function or operator created on the path after the build changed rolls back unless the change was undone again before the second check, which the verifier then catches; the proof check refuses conversions without a fingerprint and a fingerprint without conversions — and the rendering settings a text conversion depends on (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), pinned in every chunk transaction so a float or `bytea` converted to `text` lands as the flush and the digest render it; a resumed copy first deletes every shadow row above the watermark in batches of the same guarded shape — the whole shadow after a zero watermark — so the resumed cut frontier and the shadow agree on what is uncut, and its first batch's guard takes the shadow in `SHARE ROW EXCLUSIVE` as its one lock on it (two fences queue rather than deadlock), so a chunk transaction of the earlier run still committing into it, or paused before its insert after confirming the lock, ends before the clear reads; the proof check refuses a shadow that is the source by name or by OID, since the clear deletes from one and the copy reads the other; the copy selects each column the `Shadow` proof names in `Conversions` as that column's USING expression — read from the statement by `statement.ColumnConversions`, since the catalog keeps no record of a USING once `ALTER COLUMN TYPE` has run — and every other column bare, so a retyped column without USING takes the server's assignment cast, and the proof check refuses a conversion of a column the copy does not carry, with no expression, or of the primary key, whose values every chunk range, the watermark, the resume clear, and the conflict target are expressed in; `Position` snapshots the cut frontier and landed watermark as two `Watermark` values plus the in-flight chunks, and `Position.Classify` is the applier's uncut / in-flight / landed rule. While `Run` is in progress the `Copier` is the `progress.Tracker`'s `WorkSource` (`Options.Tracker`): `rows_copied` is the rows this run's committed chunks inserted, `rows_total` the source's catalog row count read once at the start of the run, and `bytes_copied` / `bytes_total` the shadow's and the source's `pg_table_size` measured at each poll — every counter measured, none projected; on a resumed run the row ratio is that run's share, not the copy's completion. Each measurement, and each chunk cut, runs in a read-only transaction of its own under the chunk budgets with the catalog alone on its `search_path`, so a poll or a cut on a caller-built pool is still bounded (LK-2) and still resolves to the real catalog (CO-9); the copier stops being the source before `Run` returns. Every refusal the copier returns is a `*copier.RefusalError` carrying a `copier.RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#copier-refusals-keyed-on-copierrefusalcause), fail-closed under `ErrInvariantViolation`; a chunk whose insert collides with a shadow row on a unique index or exclusion constraint (SQLSTATE `23505` / `23P01`) — a value the catch-up has moved between rows and the flush has yet to rewrite — is run again after a bounded linear wait (`Options.CollisionAttempts`, `Options.CollisionBackoff`; `Collisions()` counts them) and, when every attempt collides, the server's `*pgconn.PgError` is returned wrapped; any other chunk the shadow itself rejects is that error as it stands, and neither carries a cause. | CO-4, CO-9, LK-1, LK-2, LK-3 | | `pkg/checksum` | `Verifier` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) compares the source with its shadow up to the copier's landed watermark and returns a `Report` of the chunks that differ. It cuts its own chunks with a `copier.Chunker`, cutting and digesting each chunk in one read-only `REPEATABLE READ` transaction, so the cut and the chunk's two digests describe one snapshot, no snapshot outlives one chunk, and the cut's lock wait is bounded like the reads (LK-2); each transaction runs the copier's guard — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, so a repair's recopy resolves a USING expression as the copy did, and the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, and recomputed once more after a recopy before its commit, so the source side of a digest and a recopy run the resolution the build recorded), `ACCESS SHARE` on both relations taken before the snapshot, lock confirmation, relation-OID check — and pins the rendering settings the copy, the flush, and the digest share (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), since the digest hashes each row's text rendering and a database or role configured otherwise would render two floats that differ only in their last digits the same, or a `bytea` converted to `text` in the escape form the copy never wrote. Both sides run the same frozen statement — row count plus hex `sha256` of the key-ordered concatenation of each row's `sha256(convert_to(ROW(col::shadow_type, …)::text, 'UTF8'))` over `pk BETWEEN $1 AND $2` (SHA-256 so a FIPS-mode OpenSSL, which refuses `md5()`, does not fail the pass) — so only the data can differ, and the cast on every column is D7: a retyped column hashes as the value the shadow holds, and a column the statement retyped with `USING` hashes on the source side as that expression cast to the shadow's type, the one place the two statements differ. Only the copy columns are compared; generated columns present on both sides are not yet hashed. A `Report` proves nothing. `Check` runs the same pass under a `DivergencePolicy` the caller states every time (the zero value is refused): `abort` returns a `DivergenceError` carrying the report with the shadow untouched; `repair` replaces every differing chunk inside one guarded read-write transaction — delete the shadow's rows over every chunk's key range, then the copy statement itself for every chunk, so a unique value the source moved from one differing chunk to another lands instead of colliding with the stale row — and digests each chunk again in a fresh snapshot, returning a `RepairError` for the first that still differs alongside the `Outcome` listing every repair that committed. A repair pass assumes nothing else writes the shadow and the source rows it recopies hold still until the rereads; a write inside the pass reads as a `RepairError`, never as a second repair; `recopy` is the same recopy without the rereads, for a source that is written while the pass runs, and reports what it recopied the same way; a recopy that collides with a shadow row because the source moved a unique value from a chunk the comparison read equal into a differing chunk (SQLSTATE `23505` or `23P01`) is rolled back, the pass compares again, and the recopy cut from that comparison runs, up to `Options.RecopyAttempts` recopies, with every collision counted in `Outcome.Collisions` and the last one's error ending the pass when all collide. `ParseDivergencePolicy` lets a caller refuse a configured policy before any pass; it accepts `abort` and `repair` and refuses `recopy`, which no setting may name. `CheckLive` is the pass for the catch-up stage, where the source is written and the stream is the shadow's only other writer; it is described under [Live verification](#live-verification). Only a pass that found nothing and repaired nothing mints the proofs, whose constructors are private to the package: a `CleanWatermark` at the watermark it read through, and a `VerifiedShadow` only when that watermark is complete; `Outcome.Clean` is true only when the clean watermark was minted. A pass with repairs returns its `Repair`s and no proof; the next pass mints. While `Verify` or `Check` runs — through the repair phase as well — the verifier is the tracker's `progress.WorkSource` (`Options.Tracker`): `chunks_compared` and `rows_hashed` count the comparison's two-sided digests that have committed and the source rows they covered (every comparison of the pass, so under `recopy` a collision's re-compare counts on top of the first), `chunks_mismatched` the chunks the comparison found differing, `chunks_repaired` the chunks whose one recopy transaction has committed, and `chunks_reread` the chunks digested again after the comparison — the repaired chunks under `Check`, every settling reread under `CheckLive` — every counter read from the pass's memory, none from the database, and all reset when a pass starts; a pass started while another runs on the same verifier is refused. | CO-1, CO-2, CO-3, CO-9, LK-1, LK-2 | | `pkg/decode` | Creates, drops, and inspects the route's slot and publication under the derived name (`CreateSlot` with exported snapshot and consistent point; `DropSlot` waiting out a holder, idempotent, never crossing clusters or databases, dropping only the route's own publication; `InspectSlot` with `wal_status` and retained WAL, the latter unknown rather than zero on a server in recovery; `SlotStatus.WithinLagCeiling` judging a slot against a ceiling in bytes — a `lost` slot is `SlotLostError` before any measure, an unknown measure is `ErrSlotLagUnknown` and never within the ceiling, over it is `SlotLagExceededError` carrying both figures; `DefaultSlotLagCeiling` is one GiB; `IsEngineSlotName` tells the engine's name shape from a name merely wearing its prefix; `InspectSlot` also reads the server's account of a lost slot — `invalidation_reason`, `conflicting`, `inactive_since` — where the server version carries it, one statement across every supported major, and `SlotStatus.LostState` renders it into the `SlotLostError` so an operator reads *why* from the error). `OpenStream` decodes the slot with pgoutput from a position — the consistent point, or the applied position a resume checkpointed — on a dedicated replication connection proven to be a session of the pool's cluster on the target's database — the same table on another cluster derives the same slot name — into one `ChangeEvent` per committed row change: every column carried with its presence, so the unchanged-TOAST marker is an absent `Column` and NULL a present one — built by `PresentColumn` and `AbsentColumn`, read through `Get`, which returns the value together with its presence, so a reader takes the presence with the value — and `OldKey` for an UPDATE that moved the primary key under either admitted replica identity; every event also carries the xid of the transaction that committed it (`Xid`, taken from the transaction's BEGIN), which is how the applier tells a delivered commit the server has not yet made visible from one it has. A relation other than the target — another name, or the target's name under an OID other than the one `CopySwapTarget` was proven for, which is a table dropped and recreated under that name — a tuple that does not line up with it, or a change outside a transaction fails closed; a table whose shape changed under the stream is `ErrSourceShapeChanged`, a TRUNCATE is `ErrUnsupportedChange`, and a warning the server sends in copy-both mode — the walsender saying it will withhold changes, as from a publication it skipped loading — stops the stream fail-closed before a keepalive can deliver a position past them; the replication connection asks for warnings itself (`SET client_min_messages = warning` on the open session, the last word on the setting), so no role, database, or server setting, login event trigger, URL parameter, or connection pooler can keep them from it. The stream keeps `Delivered` — moved by a commit or by a keepalive between transactions, never past a transaction not yet yielded in full — apart from `Confirmed`, the one position it reports to the server: `Confirm` refuses a regression or a position beyond `Delivered`, and a keepalive reply carries only what the caller confirmed, so the slot's resume point moves on the caller's word alone. A wait that runs out in `Next` sends the same report, so the stream is never silent for longer than the caller's wait (`Next`'s doc says why that matters to the walsender). Both positions order transactions by their commit, so a change's own LSN can lie below either; each `ChangeEvent` carries the `Delivered` it arrived with, which is what a caller may confirm while the change is unapplied, and the server replays by commit position, so a resume never skips a replayed change for lying below its checkpoint. `ServerWALEnd` is the walsender's send position from its last keepalive and never below `Delivered`; a change does not move it, since its message carries only the change's own position. `ServerWALEnd − Delivered` is a lower bound on lag — the walsender reports how far it has decoded, not how far the server has written — so lag against the server's WAL end is measured on the pool with `pg_current_wal_lsn()`. The server ending replication with the slot intact is `ErrStreamEnded`, the clean restart ST-4 keeps apart from slot loss. | ST-3, ST-4, CO-4, CO-8 | -| `pkg/applier` | `Buffer` merges events into one `Entry` per primary key — each carrying the xid of the latest event merged into it — and `Drain` judges every key against the copier's `Position` and a source `Snapshot` (`ReadSnapshot`, the server's `pg_current_snapshot()`): an entry whose transaction the snapshot does not yet show as committed — the server makes a commit visible only after its WAL flush and any synchronous-standby wait, while the walsender delivers it as soon as the WAL is flushed — is treated as in flight and stays buffered (`Batch.Invisible`), so no chunk read or completion read that cannot see the change decides its fate; the zero `Snapshot` shows nothing visible, `AllVisible` is for a drain under a lock every writer has already released. `Drain` returns a `Batch` whose `CompleteFirst` names the moved, marker-bearing images the flush must complete before it writes. `Flusher` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) applies one batch in one guarded read-write transaction — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, with the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, before the writes and again after them before the commit, so a USING expression in a flush resolves as the copy's did), `ACCESS SHARE` on both relations, lock confirmation, relation-OID check, the copier's guard — bound to the lock session's context so a lost lock cancels the flush in flight. A column the statement retyped with `USING` is written through that expression (D7): the flush reads the column's source type from the catalog inside its transaction and binds the decoded value as `(SELECT FROM (SELECT $n:: AS "col") AS "source")`, so the value lands as the copy would have landed it, while a completion read from the shadow is already in the shadow's form and is bound bare; the flusher refuses, as ST-6, a proof whose conversions name the primary key, a column the copy does not carry, or an empty expression, and a converted column the source no longer carries. It first reads a completion for every `CompleteFirst` image: from the old key's shadow row, read `FOR UPDATE` so the batch's own delete of that key cannot race the read; from the source row under `Key` when `OldKeyReused` or when the old key's shadow row is absent (a move captured while the old key was uncut left no shadow row). Neither row is known to be the stream's, so the image is not written: it returns in `Result.Held` as a `HeldImage` with its completion and the `pg_current_wal_insert_lsn()` the read took in the same statement, and `Buffer.Hold` takes it back with the completion pending — an event that replaces the key's entry or reuses its old key drops the completion, an UPDATE keeps it pending behind its own values, and `Buffer.Release(passed)` fills the markers still left once the stream has delivered every change committed through the read's position, so the next `Drain` flushes the whole image (D13). An image neither row can complete is written as a delete marker at `Key`, since the source holds no row there, and its key named in `Result.Skipped`. A marker-bearing image of a row that did not move whose shadow row is absent — the source deleted the row before its chunk was read and the DELETE is still on its way — is held the same way, with no completion and `Entry.AbsenceAwaited` set, and flushed once the stream has passed the position of the check that found the row absent; an image that comes back flagged for a row still absent is a CO-8 `ErrInvariantViolation` (D13). The primary path then applies the batch column-wise inside a savepoint, every delete marker before any image: a delete marker is a `DELETE`, a whole image an `INSERT … ON CONFLICT (pk) DO UPDATE`, and a marker-bearing image of a row that did not move an `UPDATE` of only its present columns — zero rows updated is a CO-8 `ErrInvariantViolation`, never an insert that invents the omitted value. On SQLSTATE `23505` or `23P01` (a unique index or an exclusion constraint) the flush rolls back to the savepoint and reapplies the batch whole-row (D13): every remaining marker is completed from the shadow row under its own key `FOR UPDATE` (absent is a CO-8 refusal), every key in the batch is deleted in one statement, and every surviving image is inserted plain; a second collision is `ErrBatchDeferred`, and `Buffer.Requeue` puts the batch back — refusing with a CO-5 `ErrInvariantViolation` if any of its keys is held, since nothing may `Add` between `Drain` and the flush's outcome — so the caller drains again once the copier has landed the chunk the rows wait on. `Result` counts the images and deletes written and the completions, carries the held images and the skipped keys, and says whether the fallback ran. The flush pins the same rendering settings as the copy and the digest (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`) so a float or a `bytea` the decoder rendered lands, through a `USING` conversion to `text`, as the text the copy wrote and the verifier will hash, and the completion reads carry the same pin so a held value lands as the shadow renders it. `Catchup` owns the loop: it gathers the stream into the buffer, `Release`s held images at the delivered position, drains against the copier's `Position` read *after* the last `Add` (so a key the copier cuts between the read and the drain is judged in flight, never discarded), flushes, and on `ErrBatchDeferred` requeues the batch for the next pass while a chunk is in flight — once the copy has landed every key, a refusal is one no chunk can resolve, and `Run` ends with `ErrBatchDeferred` rather than requeue it forever. It confirms the lesser of the delivered position and `Buffer.OldestPending` — each keyed on the `Delivered` the event arrived with, never the event's own LSN — so the slot never advances past a change that is still buffered, held, or deferred, and never falls below what the stream had already confirmed (ST-4); the confirmation follows the requeue or hold, so it never reads a buffer emptier than the one the next drain sees. The whole loop runs under the lock session's `Bind` context, so a lost lock ends a quiet wait on the stream as it ends a flush, reported as `ErrInvariantViolation` wrapping `dbconn.ErrTableLockLost`. `Run` never returns nil: it runs once and returns the context's cause when its owner stops it, carrying any cycle error that is not the stop itself alongside. It is the `progress.WorkSource` of a catch-up step: `changes_applied`, `changes_buffered`, and `lag_bytes` measured against `pg_current_wal_lsn()` read on the pool as each cycle ends, never against the walsender's send position. Each cycle then inspects the stream's slot on the pool and judges it against `CatchupOptions.SlotLagCeiling` (zero takes `decode.DefaultSlotLagCeiling`; below one byte is refused, so the ceiling is always in force): a slot that vanished or that the server reports `lost` ends `Run` with `decode.SlotLostError`, and one retaining more WAL than the ceiling ends it with `decode.SlotLagExceededError` — the table lock stays held and the slot stays in place, so the operator reads what the slot retained and the resume path decides what follows (D11, ST-3). | CO-4, CO-5, CO-6, CO-8, LK-1, LK-2, LK-3, ST-3 | +| `pkg/applier` | `Buffer` merges events into one `Entry` per primary key — each carrying the xid of the latest event merged into it — and `Drain` judges every key against the copier's `Position` and a source `Snapshot` (`ReadSnapshot`, the server's `pg_current_snapshot()`): an entry whose transaction the snapshot does not yet show as committed — the server makes a commit visible only after its WAL flush and any synchronous-standby wait, while the walsender delivers it as soon as the WAL is flushed — is treated as in flight and stays buffered (`Batch.Invisible`), so no chunk read or completion read that cannot see the change decides its fate; the zero `Snapshot` shows nothing visible, `AllVisible` is for a drain under a lock every writer has already released. `Drain` returns a `Batch` whose `CompleteFirst` names the moved, marker-bearing images the flush must complete before it writes. `Flusher` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) applies one batch in one guarded read-write transaction — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, with the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, before the writes and again after them before the commit, so a USING expression in a flush resolves as the copy's did), `ACCESS SHARE` on both relations, lock confirmation, relation-OID check, the copier's guard — bound to the lock session's context so a lost lock cancels the flush in flight. A column the statement retyped with `USING` is written through that expression (D7): the flush reads the column's source type from the catalog inside its transaction and binds the decoded value as `(SELECT FROM (SELECT $n:: AS "col") AS "source")`, so the value lands as the copy would have landed it, while a completion read from the shadow is already in the shadow's form and is bound bare; the flusher refuses, as ST-6, a proof whose conversions name the primary key, a column the copy does not carry, or an empty expression, and a converted column the source no longer carries. It first reads a completion for every `CompleteFirst` image: from the old key's shadow row, read `FOR UPDATE` so the batch's own delete of that key cannot race the read; from the source row under `Key` when `OldKeyReused` or when the old key's shadow row is absent (a move captured while the old key was uncut left no shadow row). Neither row is known to be the stream's, so the image is not written: it returns in `Result.Held` as a `HeldImage` with its completion and the `pg_current_wal_insert_lsn()` the read took in the same statement, and `Buffer.Hold` takes it back with the completion pending — an event that replaces the key's entry or reuses its old key drops the completion, an UPDATE keeps it pending behind its own values, and `Buffer.Release(passed)` fills the markers still left once the stream has delivered every change committed through the read's position, so the next `Drain` flushes the whole image (D13). An image neither row can complete is written as a delete marker at `Key`, since the source holds no row there, and its key named in `Result.Skipped`. A marker-bearing image of a row that did not move whose shadow row is absent — the source deleted the row before its chunk was read and the DELETE is still on its way — is held the same way, with no completion and `Entry.AbsenceAwaited` set, and flushed once the stream has passed the position of the check that found the row absent; an image that comes back flagged for a row still absent is a CO-8 `ErrInvariantViolation` (D13). The primary path then applies the batch column-wise inside a savepoint, every delete marker before any image: a delete marker is a `DELETE`, a whole image an `INSERT … ON CONFLICT (pk) DO UPDATE`, and a marker-bearing image of a row that did not move an `UPDATE` of only its present columns — zero rows updated is a CO-8 `ErrInvariantViolation`, never an insert that invents the omitted value. On SQLSTATE `23505` or `23P01` (a unique index or an exclusion constraint) the flush rolls back to the savepoint and reapplies the batch whole-row (D13): every remaining marker is completed from the shadow row under its own key `FOR UPDATE` (absent is a CO-8 refusal), every key in the batch is deleted in one statement, and every surviving image is inserted plain; a second collision is `ErrBatchDeferred`, and `Buffer.Requeue` puts the batch back — refusing with a CO-5 `ErrInvariantViolation` if any of its keys is held, since nothing may `Add` between `Drain` and the flush's outcome — so the caller drains again once the copier has landed the chunk the rows wait on. `Result` counts the images and deletes written and the completions, carries the held images and the skipped keys, and says whether the fallback ran. The flush pins the same rendering settings as the copy and the digest (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`) so a float or a `bytea` the decoder rendered lands, through a `USING` conversion to `text`, as the text the copy wrote and the verifier will hash, and the completion reads carry the same pin so a held value lands as the shadow renders it. `Catchup` owns the loop: it gathers the stream into the buffer, `Release`s held images at the delivered position, drains against the copier's `Position` read *after* the last `Add` (so a key the copier cuts between the read and the drain is judged in flight, never discarded), flushes, and on `ErrBatchDeferred` requeues the batch for the next pass while a chunk is in flight — once the copy has landed every key, a refusal is one no chunk can resolve, and `Run` ends with `ErrBatchDeferred` rather than requeue it forever. It confirms the lesser of the delivered position and `Buffer.OldestPending` — each keyed on the `Delivered` the event arrived with, never the event's own LSN — so the slot never advances past a change that is still buffered, held, or deferred, and never falls below what the stream had already confirmed (ST-4); the confirmation follows the requeue or hold, so it never reads a buffer emptier than the one the next drain sees. The whole loop runs under the lock session's `Bind` context, so a lost lock ends a quiet wait on the stream as it ends a flush, reported as `ErrInvariantViolation` wrapping `dbconn.ErrTableLockLost`. `Run` never returns nil: it runs once and, when its owner stops it, returns the context's cause wrapped together with the cancellation — a stop with a cause of the owner's own still answers `errors.Is` for `context.Canceled` — carrying any cycle error that is not the stop itself alongside. It is the `progress.WorkSource` of a catch-up step: `changes_applied`, `changes_buffered`, and `lag_bytes` measured against `pg_current_wal_lsn()` read on the pool as each cycle ends, never against the walsender's send position. Each cycle then inspects the stream's slot on the pool and judges it against `CatchupOptions.SlotLagCeiling` (zero takes `decode.DefaultSlotLagCeiling`; below one byte is refused, so the ceiling is always in force): a slot that vanished or that the server reports `lost` ends `Run` with `decode.SlotLostError`, and one retaining more WAL than the ceiling ends it with `decode.SlotLagExceededError` — the table lock stays held and the slot stays in place, so the operator reads what the slot retained and the resume path decides what follows (D11, ST-3). | CO-4, CO-5, CO-6, CO-8, LK-1, LK-2, LK-3, ST-3 | | `pkg/checkpoint` | Produces `Checkpoint` and persists it: `Store` (over a `pkg/dbconn` pool) creates `pgsprite.pgsprite_checkpoint` on first use (`Ensure`, serialized under the engine's advisory key so concurrent first users never race the create; it creates only what is absent, so a pre-provisioned schema and table the engine owns need no database `CREATE`, refuses with `ErrForeignObject` a schema or table another role owns, brings a table an older engine created up to this engine's shape by adding the columns it lacks — the shape is derived from the table's columns, and every column added after the first shape is nullable so rows survive — and refuses with `*TableNewerError` — `ErrTableNewer` to `errors.Is`, naming the columns — running no DDL, a table carrying columns this engine does not know; a read or write that reaches an older table before `Ensure` has evolved it is the typed `ErrTableBehind`, the server's undefined-column error wrapped), `Save` takes the target's `TableLockSession`, confirms it from the write's own transaction, and writes the target's one row in one `INSERT … ON CONFLICT DO UPDATE` whose update is guarded on the row's format version and fingerprints and on a phase this engine knows — zero rows updated is an `IncompatibleError`, never a silent overwrite — and treats the shadow proof as write-once: a `Save` without a proof keeps the one the row holds, the same proof is accepted, and a different proof is `ErrInvariantViolation` that changes nothing, since a different shadow is a fresh start and goes through `Delete`; `Load` returns the row for a run with the same `Fingerprints`, `ErrNotFound` for no row, `ErrTableMissing` when `Ensure` never ran, an `IncompatibleError` naming the first disagreeing field (format, then source, then target fingerprint) and carrying the row's whole `Identity`, or, after bounded retries through transient errors (`dbconn.Retryable` plus a session the server ended from outside it, `57P01`/`57P02`/`57P03`) with an injected sleep, an error that is none of those; `Delete`, under the same lock, is the explicit fresh start and removes only the row whose `Identity` the caller was shown; `Inspect` is the resume's first read, before the run knows which statement it is resuming — the same outcomes as `Load`, guarded on the row format and the phase alone — a row in a phase this engine does not know is an `IncompatibleError` on `MismatchPhase`, since the phase set grows within a format without a version bump — so the row's fingerprints, phase, and shadow proof come back for the caller to judge, while every write stays guarded on the whole identity. The watermark column is `NULL` while nothing has landed, the LSN is a `pg_lsn`, the phase is stored by its stable name, and `shadow_proof` is the JSON proof of the shadow the run built (`Checkpoint.ShadowProof`, `NULL` until there is one), which a resume after the swap needs because `InspectShadow` cannot re-mint it once the shadow no longer bears its name; `PhaseReconciling`, between `catching_up` and `verifying`, is the ST-4 marker a reconcile writes before it touches the slot, so a reconcile that dies restarts as one. `ReapOrphanSlots` is the ST-3 reaper, placed here because the checkpoint row is what makes a slot owned: it lists the logical, engine-prefixed slots of `current_database()` and then reads the non-terminal rows that name them — a run makes its slot and commits the row that names it under `WithSlotLock`, the engine's advisory lock on the slot's name, transaction-scoped on a connection dialled outside the store's pool so the holder's own work on the pool never waits on the seat the lock sits in, so a slot with no row was left by a run that ended without dropping it, and a slot a done or failed row names by one whose cleanup did not finish — and re-reads each orphan's row right before dropping it, both under the same lock, so a run that claimed the slot since the pass began keeps it and one claiming it now waits for the drop and creates afresh; the pass drops each orphan through `decode.DropIdleSlot`, publication and all, reporting it in `Reaped.Dropped`; a slot a live row names (`Live`), a slot a walsender or backend holds when the drop comes round (`Active`), and a name wearing the prefix without the engine's shape (`Foreign`) are reported and left as found; slots of other databases are never read; the drops stop at the first failure, reporting what was dropped before it; a database where `Ensure` never ran is refused with `ErrTableMissing` rather than read as all orphans; and two reapers over one database take turns at each slot, the one that waited finding the slot gone. | ST-1, ST-2, ST-3, ST-4, LK-1 | | `pkg/schemachange` | Shadow builder (`BuildShadow` produces `BuiltShadow`: source and shadow OIDs, fingerprints, identity handoff, copy columns, the statement's `USING` conversions, the conversion fingerprint (`chunksql.ResolutionFingerprint` over the conversions' unqualified names on the build's `search_path`, read in the build's own transaction and empty when the statement converts nothing), fidelity snapshot; before the retargeted statement runs it tries each conversion as a stored generated column's expression on the empty shadow — the one server-side check that admits only an immutable expression — in a save point it always rolls back, and refuses a non-immutable one as `shadow-statement-conversion` wrapping `ErrConversionNotImmutable`, since the copy, the recopy, and the catch-up evaluate the expression again at other times and must get the same answer; `SourceOfDerivedName` maps a derived name back to its table), orchestrator, and cutover. `BuildShadow`, `DropShadow`, and `InspectShadow` each take the `*dbconn.TableLockSession` for the table — a dedicated direct server session, distinct from the working pool, that `dbconn.AcquireTableLock` refuses to open through a transaction-pooling proxy: the operation runs under the session's `Bind` context so a lost lock cancels the statement in flight, and its transaction re-asserts from its own connection that the session's backend holds the lock before the first write, since the lock and the work are deliberately on different sessions. `InspectShadow` is the resume path `ErrShadowExists` points at: it re-derives the `BuiltShadow` proof from the catalog for the caller to compare with its checkpoint — `BuiltShadow.Proof()` is the plain, JSON-encodable view of that proof (`BuiltShadow` marshals as it), and nothing decodes back into a `BuiltShadow`, so only the builder and the inspection mint one; `DropShadow` is the D5 cleanup, dropping only a plain table the source's owner owns, without `CASCADE`. `GateCutover` is the ST-5 fidelity gate: handed a `BuiltShadow` and a `checksum.VerifiedShadow` for the same table, it re-reads both relations under the lock, refuses on a moved OID, a fingerprint or fidelity snapshot that drifted from the build's record, an invalid shadow index, or a swap name already taken, and otherwise mints `CutoverReady` — the pairing of every source index and extended statistics object with its shadow counterpart by catalog definition (not by name, since `LIKE` renames them; the default operator class and the column's own collation of a key column the statement retyped are set aside, since the server re-derives them for the new type; a written operator class or collation still has to agree, and a relaxed definition two indexes share pairs neither), and the sequences the swap must re-own; the swap itself consumes that proof. `Cutover` is the swap: one transaction under the lock session that takes `ACCESS EXCLUSIVE` on the source and the shadow under `lock_timeout` with bounded retry and backoff, runs the caller's `DrainFunc` (nil for a quiesced table) once per attempt once writers are excluded — a rolled-back attempt takes the drain's work with it, so the next attempt drains again from the same state — re-runs the copy-and-swap shape check (RF-2) and the gate's checklist at `READ COMMITTED`, whatever the session default, so the pairing it renames by is as fresh as the lock, performs the D8 renames (every source dependent to its derived `_old` name, the source to `_old`, the shadow to the source's name, each paired shadow dependent to its partner's name), completes the D5 handoff (re-owning shared sequences; recreating each identity column with the source sequence's declared options — bounds following a widened column's type — under the source sequence's name, carrying the source sequence's grants, and `setval` to its `(last_value, is_called)`), re-reads the catalog to confirm every rename and handoff before committing, and mints `SwappedTable`; an attempt whose connection broke or whose context ended is resolved by following its backend to its exit and then reading from a fresh connection which OID bears the source name, never assumed (LK-4), and every attempt the catalog shows rolled back is reported wrapping `ErrCutoverRolledBack`. `DropOldTable` is the D9 drop: a separate bounded transaction that drops only the relation whose OID the `SwappedTable` recorded as the source, without `CASCADE`. `InspectSwapped` is the resume path between `Cutover` and `DropOldTable`, and the one home of LK-4's committed branch: handed the `Proof` the caller checkpointed before the cutover, it re-derives the `SwappedTable` from the catalog in one bounded read transaction under the lock session — the source's name must be borne by the proof's shadow OID and the derived `_old` name by the proof's source OID (ST-6), the dependent pairing is re-derived from the names the swap gave (a live dependent pairs with the old-table dependent under its derived name, so the re-derived pairs carry no shadow names), and every identity column the proof handed off must be an identity on the live table drawing from the sequence recreated under the source sequence's name; `SwappedTable.Proof()` is the plain, JSON-encodable `SwapProof` view of it (`SwappedTable` marshals as it), and nothing decodes back into a `SwappedTable`, so only the cutover and the inspection mint one. Both relations must still be owned by the role the build recorded, since the drop runs as it. The catalog states the proof does not describe each get their own answer: the source still live is `ErrNotSwapped` (run `Cutover`), the `_old` name borne by nothing while the source's OID is gone from the catalog is `ErrOldTableNotFound` (the drop already committed; the change is complete), the `_old` name borne by another relation — or by nothing while the retained source still exists under another name — is refused as `cutover-relation-replaced`, a live or old table owned by another role is refused as `cutover-swap-mismatch`, the source's name borne by neither relation is refused as `cutover-outcome-ambiguous` (LK-4), and so is a read while another backend holds either relation exclusively, since a swap whose `COMMIT` is still deciding holds both until its outcome is visible and the same call answers once it has let go. `SwapProof.SameSwap` is the resume's comparison: the two relations by OID and name, the owner, the re-owned sequences, and the identity columns, leaving out the dependent pairing and the attempt count a re-derived proof cannot carry. `DropOldTable` locks whatever bears the `_old` name before it reads the OID it compares, so the relation it proves is the relation it drops. `InspectResume` is the resume path's first read: handed a non-terminal checkpoint row for the target, it refuses one it cannot act on before any slot is touched — a terminal phase (`resume-row-terminal`), a row in `cutover`, which only `InspectSwapped` can answer (`resume-row-cutover`), another table or another slot name (`resume-row-mismatch`) — and otherwise answers `ResumeClean` when the slot is present, the server still serves it, and nothing holds it, or `ResumeReconcile` when the row names no slot, the slot is gone, the server reports it lost (with the server's cause on the `decode.SlotLostError` it carries), or the row is already in `reconciling`; a slot of the derived name that is physical or another database's is refused as `resume-slot-foreign` in every phase (it also answers `errors.Is` for `decode.ErrForeignDecodingState`), and the engine's own slot while another backend holds it as `resume-slot-held`, to be inspected again once the holder lets go; a target proven without logical decoding is always clean, having no slot to lose. `Reconciler.Reconcile` is the ST-4 reconcile mode (D11, D14), run whole under the lock session's `Bind` context so a lost lock cancels the slot command in flight — a drop still waiting out a holder included — and a session that has lost the table never drops or makes a slot a run holding the lock since may own: it applies `InspectResume`'s row checks, refuses a row whose shadow name or fingerprints are not the `BuiltShadow`'s, confirms from a transaction of its own that the source and the shadow are still the relations the build recorded (`cutover-relation-replaced` otherwise, before anything is written), writes the row to `reconciling` naming the derived slot and publication so the slot it is about to make is the row's from the moment it exists, replaces whatever slot bears the derived name with a fresh one through `decode.ReplaceSlot`, keeping the publication in place for the live source's writers, runs one `checksum` pass at or below the watermark under `recopy` when anything landed — recopying only the chunks the gap touched, with no reread, since the new slot carries every write that lands after the recopy — and writes the row back to `copying` or `catching_up` by the watermark, naming the new slot and its consistent point as the applied position the stream reopens from; the shadow is kept and the watermark never moves, and the open slot is returned for the caller to decode from or close. `Runner.Run` is the orchestrator, routed by the target's proof — a quiesced run for a target proven without logical decoding, a live run for one proven for it: it acquires the table lock session itself, makes sure the checkpoint table exists, reads the target's row through `Store.Inspect`, and dispatches on it — no row builds the shadow, fences the source, and writes `copying` with the statement's text (a relation already under the shadow's name with no row is refused as `shadow-unrecorded`; a first save whose row does not land drops the shadow and the fence again, so nothing unrecorded is left); a `copying` or `verifying` row is held against the statement it records — compared as parsed, the same operations on the same table, so spelling and schema qualification do not matter (`resume-row-mismatch` otherwise, and a row without a statement, or one whose statement names another table, likewise) — and against the shadow `InspectShadow` re-derives, names, fingerprints, and JSON-encoded proof alike (`resume-row-mismatch` otherwise, except a proof that carries a conversion fingerprint and differs in that alone, which is `copy-conversion-drift`), and installs over the fence the stopped run left; a row whose swap has not committed — a `cutover` row whose swap rolled back included — and whose shadow no relation bears is refused as `resume-shadow-gone`, the counterpart of `shadow-unrecorded`, writing nothing, so the row and the slot a live row names are left for the operator to drop and delete; a `cutover` row is answered by `InspectSwapped`, a committed swap continuing at the old table drop and `ErrNotSwapped` back through verify and gate; a terminal row is refused as `resume-row-terminal` and left for the caller to delete; each kind of run refuses the other kind's row as `resume-row-decoding` (the quiesced run a row that names a slot or stands in `catching_up` or `reconciling`, since it has no slot to continue it with; the live run a row that names no slot, since the source behind it was fenced, not captured) — then runs the phases in order. The quiesced run: the copier from the row's watermark, the landed watermark saved every `CheckpointInterval` and once more when the copy ends, `verifying` with the complete watermark, one `checksum` pass under `DivergenceAbort`, `GateCutover`, the row to `cutover` before `Cutover` is called with no drain, `DropOldTable` unless `KeepOld` — under which the fence, which rode the rename to `_old`, is taken off the retained table — and the row to `done`. The live run: no row creates the slot and writes `copying` naming it, with the slot's consistent point as the applied position, under `Store.WithSlotLock` so the reaper never meets the slot before its row (a first save that does not land drops the slot and the shadow again; a slot of the derived name the run did not make is reported as the `*decode.SlotExistsError` it is and left as found); a resumed row is read through `InspectResume` right before the stream opens, and a slot that is gone or lost is replaced by `Reconciler.Reconcile`, whose outcome `Result.Reconciled` and `Result.Reconciliation` report (a `cutover` row whose swap `InspectSwapped` shows rolled back is written back to `catching_up` first, since the inspection reads a `cutover` row as a swap still deciding); the stream is then opened at the row's position and applied to the shadow by a capture (`applier.Catchup` over a `Flusher`) that runs beside the fill — the copier while the row is in `copying`, its chunk positions handed to the capture so the stream's changes to a chunk land only once the chunk has; `catching_up` until the stream has confirmed past a marker written after the copy (a resumed `verifying` row catches up again before its pass); then `verifying`, one `checksum.CheckLive` pass under `DivergenceAbort` with the capture's `Catchup` as its confirmed source — then the capture is ended, the row records the position it confirmed through, `GateCutover`, the row to `cutover`, and `Cutover` with a drain that under the swap's lock writes a marker, reopens the stream from the row's position, gathers until the marker is delivered, closes the stream, waits for the slot to go idle — all on one connection dialled outside the pool for the attempt, since the swap holds a pool connection and a drain waiting on the pool for a second would wait on every other swap doing the same — and flushes everything gathered through the same `Flusher` (a batch the flusher defers is refused as `cutover-drain-deferred`; images the flusher held back are released and flushed once more, and a hold that persists, or anything left unapplied, is `ErrInvariantViolation`, since no writer can change the shadow under the swap's lock); the slot is dropped once the swap has committed, before `DropOldTable` or the `KeepOld` settle, and the row goes to `done`. The fence is the quiescence: a statement-level `BEFORE INSERT / UPDATE / DELETE / TRUNCATE` trigger on the source, created as the engine role, enabled `ALWAYS` so a session in `session_replication_role = replica` is refused too, and calling a function in the engine's schema, that refuses every write with SQLSTATE `55006` while the run is in progress or stopped between two saves (a resumed run installs over it, which enables it `ALWAYS` again); the quiesced `Runner`'s swap re-reads the fence under its `ACCESS EXCLUSIVE` lock and refuses as `cutover-unfenced` (CO-1) when the engine's trigger is not on the source, enabled `ALWAYS`, leaving the row in `cutover` for the next `Run` to install again and verify again (a live run installs no fence, so its swap does not check one); abandoning a stopped run is `DropShadow`, `DropWriteFence` (quiesced) or `decode.DropSlot` (live), and `Store.Delete` under the table lock. `PhaseFailed` is never written: a run that stops leaves the row at its last save and the next `Run` resumes from it. Every refusal the ten operations return is a `*RefusalError` carrying a `RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#shadow-operation-refusals-keyed-on-refusalcause), so an importer routes on the cause rather than on message text; `Runner.Run` re-mints every `copier.RefusalCause` as a shadow cause too, wrapping the copier's refusal, so a caller reads one vocabulary whether the refusal came from a shadow operation or from the copy. | LK-1, LK-2, LK-4, CO-1, ST-4, ST-5, ST-6, ST-7 | | `pkg/refusal` | `Class`, the routing class a typed refusal cause names — `capability-boundary`, `by-design`, `environmental`, or `invariant-violation`, the classes the cause tables of [refusal-classes.md](refusal-classes.md) use — answered by `copier.RefusalCause.Class()`, `schemachange.RefusalCause.Class()`, `preflight.PartitionRefusalCause.Class()`, `preflight.CopySwapRefusalCause.Class()`, and `executor.CreateShapeCause.Class()` from each cause's `class` cell, so an importer triages on the class without reading the cause's text; the per-cause row still decides the action, and a class never licenses a retry on its own. `verdict.Class` declares its tokens from this package, so the periphery converts with a plain conversion and the core never imports the verdict renderer. | — | diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 272acb01..99021f0c 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -340,7 +340,7 @@ the slot under the row (decoding target): | `ErrCutoverRolledBack` / `ErrLockRetriesExhausted` | The swap did not commit; the source is still live and the row is in `cutover` | Retry `Run`: the row is answered by `InspectSwapped`, which sends the run back through verify and gate to a fresh swap attempt | | `ErrTableLockLost` (`shadow-lock-lost`) | The lock session dropped mid-run; whatever was in flight was cancelled with it | Retry `Run` once the database is reachable; the row says where to continue | | An error wrapped `after the checkpoint in phase …` | The row saved; `AfterCheckpoint` returned an error after it | The row and the run are consistent; retry `Run` when the hook's cause is cleared. A hook that fails at the first save leaves the shadow, since the row records it. A hook that fails at the done save stopped nothing: `Run` returns the completed `Result` beside the error, with `Result.Checkpoint.Phase` in `PhaseDone`, and a retry would start a new change | -| `context.Canceled` / `DeadlineExceeded` | The caller stopped the run; a cancelled copy saves what its committed chunks landed before returning | Retry `Run`; the copy continues from the saved watermark | +| `context.Canceled` / `DeadlineExceeded` | The caller stopped the run; a cancelled copy saves what its committed chunks landed before returning. A caller that cancelled with a cause of its own (`context.WithCancelCause`) gets both: the error answers `errors.Is` for the cause and for `context.Canceled` | Retry `Run`; the copy continues from the saved watermark | A refusal from either read leaves the row and the slot as they were. A `Reconcile` that fails after its marker leaves the row in `reconciling`, and `InspectResume` answers that row as @@ -433,7 +433,7 @@ backend ends in exactly one of three verdicts: | --- | --- | --- | | `outcome: executed-via-copy-and-swap` | The shadow was built, filled, verified, and swapped in under the source's name; `detail` says how many rows and chunks the verification pass covered, whether the retained source was dropped or kept under its `_old` name, whether the run first reconciled a lost slot, and — for a run that blocked writes (`--block-writes`) — that the source's writes were fenced for the run's length, since exit 0 otherwise reads as online-safe; `writes_blocked: true` says the same for a machine, on this outcome and on a failed run whose fence went on: one that left its fence standing, or that was abandoned after the fence went on. A blocked-writes run that stopped before its fence went on, for example at a table lock another session holds, reports `false`, since the table kept taking writes; its detail says so and that running the statement again starts from the beginning. A run whose done row is saved is this outcome even when what came after failed — `OnCheckpoint` handed the done row, or the release of the lock session; the detail then names that error. The error is nil | Done. Exit 0 at the CLI; the outcome token is distinct from `executed-natively` so an adapter can tell the two paths apart in its record | | `outcome: refused`, `reason: copy-and-swap-refused` | The route's own gates declined: the preflight found a table shape or an environment it will not run against, or a shadow operation refused mid-run to keep an invariant. `cause` is the route's token unchanged — a `copy-and-swap-…` cause from the preflight, a `shadow-…`, `cutover-…`, `resume-…`, or `copy-…` cause from the shadow operations — and `class` is its routing class. The error is nil | Route on `cause` with the two tables above and [refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause). A preflight refusal touched nothing. A refusal raised after the row was saved — a `cutover-…` or `resume-…` cause — leaves the row, the shadow, and the run's hold on the source in place: the write fence of a run that blocked writes, or the replication slot of a live run, which pins WAL until it is dropped; the cause says whether the next run clears it or the run is abandoned first. A missing privilege is not this reason: it is the same `insufficient-privileges` refusal the native route reports, with the exact `GRANT` in `detail` | -| `outcome: failed`, with `code` | The run stopped on an error that is not a refusal; the error is returned alongside the verdict. `code` is `cancelled-by-caller` when the caller's context ended, `invariant-violation` when the engine caught a breach of its own contract, `data-fault` when the server rejected the rows as the statement would shape them — a `*pgconn.PgError` in SQLSTATE class `22` or `23`: a value the new type will not hold, a constraint the copied rows break — and `execution-failed` for everything else — a divergence, an unsettled chunk, a rolled-back swap, a lost lock session, exhausted lock retries, a slot past its lag ceiling | `detail` says what the stop left and what to do with it. A data fault would stop the same way on every resume, so the Runner abandons the run itself before returning (`*schemachange.AbandonedError`): the shadow is dropped, the fence is taken off the source or the slot is dropped, and the checkpoint row is deleted — nothing is held; correct the data or the statement, then run it again without any acknowledgement, and the run starts from the beginning. Every other stop leaves the checkpoint row at its last save with the run's hold on the source — the fence or the slot: a divergence (`*checksum.DivergenceError`) is wedged in `verifying`, and a data fault whose abandon itself failed (the detail says the run "could not be abandoned") is held the same way; for both the detail says to correct the data or the statement and then start the run over with the restart acknowledgement (`--restart SCHEMA.TABLE`; `CopySwapOptions.Restart`), which drops the shadow, takes the fence off the source or drops the slot, deletes the row, and runs the change from the start. Any other stop says that running the same statement again resumes from the row. The `Run` outcome table above is the finer map. Match the wrapped error with `errors.Is`/`errors.As` for the finer routing — the verdict's `code` is the coarse class | +| `outcome: failed`, with `code` | The run stopped on an error that is not a refusal; the error is returned alongside the verdict. `code` is `cancelled-by-caller` when the caller's context ended — whatever cause the caller cancelled it with, `invariant-violation` when the engine caught a breach of its own contract, `data-fault` when the server rejected the rows as the statement would shape them — a `*pgconn.PgError` in SQLSTATE class `22` or `23`: a value the new type will not hold, a constraint the copied rows break — and `execution-failed` for everything else — a divergence, an unsettled chunk, a rolled-back swap, a lost lock session, exhausted lock retries, a slot past its lag ceiling | `detail` says what the stop left and what to do with it. A data fault would stop the same way on every resume, so the Runner abandons the run itself before returning (`*schemachange.AbandonedError`): the shadow is dropped, the fence is taken off the source or the slot is dropped, and the checkpoint row is deleted — nothing is held; correct the data or the statement, then run it again without any acknowledgement, and the run starts from the beginning. Every other stop leaves the checkpoint row at its last save with the run's hold on the source — the fence or the slot: a divergence (`*checksum.DivergenceError`) is wedged in `verifying`, and a data fault whose abandon itself failed (the detail says the run "could not be abandoned") is held the same way; for both the detail says to correct the data or the statement and then start the run over with the restart acknowledgement (`--restart SCHEMA.TABLE`; `CopySwapOptions.Restart`), which drops the shadow, takes the fence off the source or drops the slot, deletes the row, and runs the change from the start. Any other stop says that running the same statement again resumes from the row. The `Run` outcome table above is the finer map. Match the wrapped error with `errors.Is`/`errors.As` for the finer routing — the verdict's `code` is the coarse class | The adapter supplies the route's policy on `migrate.Options.CopySwap`: the connection the pool was dialed with (the route opens dedicated sessions — the table lock, the copy workers, diff --git a/pkg/applier/catchup.go b/pkg/applier/catchup.go index 3f0f71f2..bcf13724 100644 --- a/pkg/applier/catchup.go +++ b/pkg/applier/catchup.go @@ -308,13 +308,26 @@ func stopped(ctx context.Context, schema, table string, err error) error { if ctx.Err() == nil { return err } - stop := fmt.Errorf("catch-up on %s.%s stopped: %w", schema, table, context.Cause(ctx)) + stop := fmt.Errorf("catch-up on %s.%s stopped: %w", schema, table, endedWith(ctx)) if errors.Is(err, ctx.Err()) { return stop } return fmt.Errorf("%w; the cycle it stopped in failed: %w", stop, err) } +// endedWith is why an ended ctx ended. A caller that cancelled with a cause +// of its own (context.WithCancelCause) gets that cause wrapped together +// with ctx.Err(), so the error reads as the cause and as a cancellation +// (errors.Is(err, context.Canceled)) alike; a caller that cancelled +// plainly, or whose deadline passed, gets ctx.Err() as it stands. +func endedWith(ctx context.Context) error { + cause := context.Cause(ctx) + if errors.Is(cause, ctx.Err()) { + return cause + } + return fmt.Errorf("%w (%w)", cause, ctx.Err()) +} + // cycle is one gather, drain, flush, and confirm. Its status is recorded // once, at the end, so a poll never pairs this cycle's counters with the // last cycle's positions. diff --git a/pkg/applier/catchup_test.go b/pkg/applier/catchup_test.go index fabcdc33..e15262be 100644 --- a/pkg/applier/catchup_test.go +++ b/pkg/applier/catchup_test.go @@ -94,7 +94,8 @@ func TestStoppedKeepsACycleErrorThatIsNotTheStop(t *testing.T) { receive := fmt.Errorf("receive from slot: %w", context.Canceled) err := stopped(ctx, "s", "t", receive) assert.ErrorIs(t, err, cause) - assert.EqualError(t, err, "catch-up on s.t stopped: owner is ready to cut over", "the receive error is the stop itself") + assert.ErrorIs(t, err, context.Canceled, "a stop with the owner's own cause is still a cancellation") + assert.EqualError(t, err, "catch-up on s.t stopped: owner is ready to cut over (context canceled)", "the receive error is the stop itself") err = stopped(ctx, "s", "t", breach) assert.ErrorIs(t, err, cause, "the stop is still reported") @@ -102,6 +103,28 @@ func TestStoppedKeepsACycleErrorThatIsNotTheStop(t *testing.T) { assert.ErrorIs(t, err, breach) } +// An ended ctx is reported by its cause; a cause the owner chose is wrapped +// with ctx.Err() so the error is also the cancellation it is, while a plain +// cancel or a passed deadline is reported as it stands, not doubled. +func TestEndedWithKeepsACancellationReadableAsOne(t *testing.T) { + plain, cancel := context.WithCancel(t.Context()) + cancel() + assert.Equal(t, context.Canceled, endedWith(plain)) + + expired, cancelExpired := context.WithDeadline(t.Context(), time.Unix(0, 0)) + defer cancelExpired() + <-expired.Done() + assert.Equal(t, context.DeadlineExceeded, endedWith(expired)) + + caused, cancelCaused := context.WithCancelCause(t.Context()) + cause := errors.New("owner is ready to cut over") + cancelCaused(cause) + err := endedWith(caused) + assert.ErrorIs(t, err, cause) + assert.ErrorIs(t, err, context.Canceled) + assert.EqualError(t, err, "owner is ready to cut over (context canceled)") +} + // The copy has landed when its cut frontier is complete and no claimed // chunk is unlanded; a complete frontier with a chunk in flight, or a // frontier short of the end, still has a chunk that can resolve a refusal. diff --git a/pkg/applier/pause.go b/pkg/applier/pause.go index 5f8bde53..02e5c34e 100644 --- a/pkg/applier/pause.go +++ b/pkg/applier/pause.go @@ -61,7 +61,7 @@ func (c *Catchup) holdFlushes(ctx context.Context) error { case <-c.flushes: return nil case <-ctx.Done(): - return fmt.Errorf("pause the catch-up on %s.%s: %w", c.flusher.target.Schema(), c.flusher.target.Table(), context.Cause(ctx)) + return fmt.Errorf("pause the catch-up on %s.%s: %w", c.flusher.target.Schema(), c.flusher.target.Table(), endedWith(ctx)) } } diff --git a/pkg/copier/copier.go b/pkg/copier/copier.go index 43259352..5b6ae6fc 100644 --- a/pkg/copier/copier.go +++ b/pkg/copier/copier.go @@ -271,20 +271,47 @@ func (c *Copier) Run(ctx context.Context, pool *pgxpool.Pool) error { if err := c.clearAbove(ctx, pool, c.Position().Watermark); err != nil { return c.finish(err) } - ctx, stop := context.WithCancelCause(ctx) + workers, stop := context.WithCancelCause(ctx) defer stop(nil) var wg sync.WaitGroup for range c.opts.Workers { wg.Go(func() { - if err := c.work(ctx, pool); err != nil { + if err := c.work(workers, pool); err != nil { stop(err) } }) } wg.Wait() // INV: LK-3 - return c.finish(context.Cause(ctx)) + return c.finish(stopReason(ctx, workers)) +} + +// stopReason is why the workers stopped: the first worker's error, or the +// caller's ending of ctx. A worker's error stands as it is, even when the +// caller cancelled after it, since that error is what stopped the copy. +func stopReason(ctx, workers context.Context) error { + cause := context.Cause(workers) + if ctx.Err() == nil { + return cause + } + if !errors.Is(cause, context.Cause(ctx)) { + return cause + } + return endedWith(ctx) +} + +// endedWith is why an ended ctx ended. A caller that cancelled with a cause +// of its own (context.WithCancelCause) gets that cause wrapped together +// with ctx.Err(), so the error reads as the cause and as a cancellation +// (errors.Is(err, context.Canceled)) alike; a caller that cancelled +// plainly, or whose deadline passed, gets ctx.Err() as it stands. +func endedWith(ctx context.Context) error { + cause := context.Cause(ctx) + if errors.Is(cause, ctx.Err()) { + return cause + } + return fmt.Errorf("%w (%w)", cause, ctx.Err()) } func (c *Copier) start() error { diff --git a/pkg/copier/copier_integration_test.go b/pkg/copier/copier_integration_test.go index 07e7ee52..e5ab3904 100644 --- a/pkg/copier/copier_integration_test.go +++ b/pkg/copier/copier_integration_test.go @@ -2,6 +2,7 @@ package copier_test import ( "context" + "errors" "fmt" "math" "strconv" @@ -333,8 +334,12 @@ func TestCopierCancellationKeepsTheCancelledChunkInFlight(t *testing.T) { Clock: pin.clock, }) require.NoError(t, err) - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() + // The owner cancels with a cause of its own, as an embedder does; the + // copy must still read as the cancellation it is. + ctx, cancelWith := context.WithCancelCause(t.Context()) + defer cancelWith(nil) + ownerStop := errors.New("owner is ready to cut over") + cancel := func() { cancelWith(ownerStop) } results := make(chan error, 1) var wg sync.WaitGroup wg.Go(func() { results <- c.Run(ctx, f.pool) }) @@ -355,7 +360,8 @@ func TestCopierCancellationKeepsTheCancelledChunkInFlight(t *testing.T) { const stopDeadline = 15 * time.Second select { case err := <-results: - assert.ErrorIs(t, err, context.Canceled) + assert.ErrorIs(t, err, context.Canceled, "a cancel with the owner's own cause is still a cancellation") + assert.ErrorIs(t, err, ownerStop, "and carries the owner's cause") case <-time.After(stopDeadline): t.Fatalf("copy did not stop within %s of cancellation", stopDeadline) } diff --git a/pkg/copier/copier_test.go b/pkg/copier/copier_test.go index 54cc0024..1ebcfb5d 100644 --- a/pkg/copier/copier_test.go +++ b/pkg/copier/copier_test.go @@ -1,6 +1,7 @@ package copier import ( + "context" "errors" "math" "testing" @@ -93,3 +94,40 @@ func TestCopierOptionsValidateAsTheCopyWould(t *testing.T) { assert.ErrorIs(t, err, ErrInvalidChunkerOptions) assert.NotErrorIs(t, err, ErrInvalidOptions) } + +// Run reports why its workers stopped. A worker's own error is reported as +// it stands, whether or not the caller cancels afterwards; the caller's +// cancellation is reported as ctx.Err() when it was plain, and as the +// caller's own cause wrapped with ctx.Err() when it was cancelled with +// one, so a cause-cancelled copy still reads as a cancellation. +func TestStopReasonKeepsACancellationReadableAsOne(t *testing.T) { + t.Run("a worker's error is its own, even once the caller cancels", func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + workers, stop := context.WithCancelCause(ctx) + failed := errors.New("copy chunk: connection reset") + stop(failed) + assert.Same(t, failed, stopReason(ctx, workers)) + cancel(errors.New("owner is ready to cut over")) + assert.Same(t, failed, stopReason(ctx, workers), "the first failure stopped the copy, not the cancel after it") + }) + + t.Run("a plain cancel is ctx.Err() as it stands", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + workers, stop := context.WithCancelCause(ctx) + defer stop(nil) + cancel() + assert.Equal(t, context.Canceled, stopReason(ctx, workers)) + }) + + t.Run("a cancel with a cause is the cause and a cancellation", func(t *testing.T) { + ctx, cancel := context.WithCancelCause(t.Context()) + workers, stop := context.WithCancelCause(ctx) + defer stop(nil) + cause := errors.New("owner is ready to cut over") + cancel(cause) + err := stopReason(ctx, workers) + assert.ErrorIs(t, err, cause) + assert.ErrorIs(t, err, context.Canceled) + assert.EqualError(t, err, "owner is ready to cut over (context canceled)") + }) +} diff --git a/pkg/migrate/copy_swap_integration_test.go b/pkg/migrate/copy_swap_integration_test.go index 8dce823c..a7f1b0d8 100644 --- a/pkg/migrate/copy_swap_integration_test.go +++ b/pkg/migrate/copy_swap_integration_test.go @@ -476,3 +476,48 @@ func TestCopyAndSwapStopsAtAPeriodicSaveOfTheCopy(t *testing.T) { assert.False(t, f.hasColumn(t, "orders", "note")) assert.EqualValues(t, 2500, f.rowCount(t, "orders")) } + +// A caller that cancels with a cause of its own (context.WithCancelCause) +// has still cancelled: the verdict is coded cancelled-by-caller, exactly +// as for a plain cancel, and the error carries both the caller's cause and +// the cancellation. The cancel lands at a periodic save mid-copy, where +// the copy is the one reporting why it stopped; the row stands in the +// copying phase and the next run carries the change to the executed +// verdict. +func TestCopyAndSwapCancelledWithACauseIsCancelledByTheCaller(t *testing.T) { + f := newCopySwapFixture(t) + f.createOrders(t) + const alter = `ALTER TABLE {schema}.orders DROP COLUMN note` + ownerStop := errors.New("owner is ready to cut over") + runCtx, cancel := context.WithCancelCause(t.Context()) + defer cancel(nil) + opts := f.options() + opts.CopySwap.CheckpointInterval = time.Millisecond + opts.CopySwap.Copier = copier.Options{ + Workers: 1, + Chunker: copier.ChunkerOptions{InitialRows: 10, MinRows: 10, MaxRows: 10}, + } + opts.CopySwap.OnCheckpoint = func(_ context.Context, cp checkpoint.Checkpoint) error { + if midCopy(cp) { + cancel(ownerStop) + } + return nil + } + st, err := statement.ParseOne(f.inSchema(alter)) + require.NoError(t, err) + + v, err := copyAndSwap(runCtx, f.pool, st, opts, slog.New(slog.DiscardHandler)) + require.Error(t, err) + assert.ErrorIs(t, err, context.Canceled, "a cancel with a cause still reads as a cancel") + assert.ErrorIs(t, err, ownerStop, "and still names the caller's cause") + assert.Equal(t, verdict.OutcomeFailed, v.Outcome) + assert.Equal(t, string(executor.CodeCancelledByCaller), v.Code) + assert.Equal(t, checkpoint.PhaseCopying, f.inspectRow(t).Phase) + assert.True(t, f.hasColumn(t, "orders", "note"), "the source is untouched while the run is stopped") + + opts.CopySwap.OnCheckpoint = nil + v, err = f.run(t, alter, opts) + require.NoError(t, err) + assert.Equal(t, verdict.OutcomeExecutedViaCopyAndSwap, v.Outcome) + assert.EqualValues(t, 2500, f.rowCount(t, "orders")) +} diff --git a/pkg/schemachange/runner_crash_integration_test.go b/pkg/schemachange/runner_crash_integration_test.go index 4a1e8d89..f8d4d385 100644 --- a/pkg/schemachange/runner_crash_integration_test.go +++ b/pkg/schemachange/runner_crash_integration_test.go @@ -447,8 +447,11 @@ func TestRunnerRecordsTheWorkACancelledCopyLanded(t *testing.T) { st := f.alter(t, alter) shadow := schemachange.ShadowName(f.schema, "orders") - runCtx, cancel := context.WithCancel(t.Context()) - defer cancel() + // The owner cancels with a cause of its own, as an embedder does; the + // run must still read as the cancellation it is. + runCtx, cancel := context.WithCancelCause(t.Context()) + defer cancel(nil) + ownerStop := errors.New("owner is ready to cut over") results := make(chan error, 1) var wg sync.WaitGroup wg.Go(func() { @@ -461,11 +464,12 @@ func TestRunnerRecordsTheWorkACancelledCopyLanded(t *testing.T) { require.Eventually(t, func() bool { return f.relationExists(t, shadow) && f.rowCount(t, shadow) == landedBelowPin }, landDeadline, 20*time.Millisecond, "every chunk below the pinned one should land") - cancel() + cancel(ownerStop) const cancelPropagation = 5 * time.Second select { case err := <-results: - require.ErrorIs(t, err, context.Canceled) + require.ErrorIs(t, err, context.Canceled, "a cancel with the owner's own cause is still a cancellation") + require.ErrorIs(t, err, ownerStop, "and carries the owner's cause") case <-time.After(cancelPropagation): t.Fatalf("run did not return within %s of being cancelled", cancelPropagation) } diff --git a/pkg/schemachange/runner_live_integration_test.go b/pkg/schemachange/runner_live_integration_test.go index 33cd2529..2376b5c7 100644 --- a/pkg/schemachange/runner_live_integration_test.go +++ b/pkg/schemachange/runner_live_integration_test.go @@ -2,6 +2,7 @@ package schemachange_test import ( "context" + "errors" "fmt" "testing" "time" @@ -245,6 +246,52 @@ func TestRunnerResumesALiveRunWhoseSlotWasDropped(t *testing.T) { assert.Equal(t, checkpoint.PhaseCatchingUp, second.seen()[0], "the reconcile hands the run a row catching up on the new slot") } +// A live run cancelled while it catches up — the stream open, the +// catch-up applying — ends as the cancellation it is, whatever the owner +// cancelled with: an owner's own cause (context.WithCancelCause) is +// carried and the error still reads as context.Canceled, so an embedder +// that stops the run with a cause of its own is not told the run failed. +// The stop is clean: the row stays catching up, the slot is let go, and +// a resume carries the run to done under the same writer. +func TestRunnerCancelledWithACauseWhileCatchingUpReadsAsCancelled(t *testing.T) { + const seeded = 2500 + f := newDecodingFixture(t, seeded) + w := f.startWriter(t, seeded, liveWriterInterval) + + runCtx, cancel := context.WithCancelCause(t.Context()) + defer cancel(nil) + ownerStop := errors.New("owner is ready to cut over") + cancelWhileCatchingUp := func(_ context.Context, cp checkpoint.Checkpoint) error { + if cp.Phase == checkpoint.PhaseCatchingUp { + cancel(ownerStop) + } + return nil + } + r, err := schemachange.NewRunner(f.cfg, f.store, liveRunOptions(schemachange.RunOptions{AfterCheckpoint: cancelWhileCatchingUp})) + require.NoError(t, err) + _, err = r.Run(runCtx, f.pool, f.target, f.liveStatement(t)) + require.ErrorIs(t, err, context.Canceled, "a cancel with the owner's own cause is still a cancellation") + require.ErrorIs(t, err, ownerStop, "and carries the owner's cause") + first := w.stop(t) + require.Positive(t, first, "the source was written while the first run was in progress") + + row, err := f.store.Inspect(t.Context(), f.table.Schema, f.table.Table) + require.NoError(t, err) + assert.Equal(t, checkpoint.PhaseCatchingUp, row.Phase, "the cancelled run left its row where it was") + status, found, err := decode.InspectSlot(t.Context(), f.pool, row.SlotName) + require.NoError(t, err) + require.True(t, found, "the cancelled run leaves its slot for the resume") + assert.False(t, status.Active, "the cancelled run's stream let the slot go") + + w = f.startWriterFrom(t, seeded, liveWriterInterval, liveUniqBase+first+1) + var writes int + result, err := f.runLive(t, schemachange.RunOptions{AfterCheckpoint: untilVerifying(t, w, &writes, nil)}) + require.NoError(t, err) + f.assertLiveDone(t, result) + assert.Positive(t, writes) + assert.Equal(t, row.ShadowTable, result.Shadow.ShadowTable, "the resumed run continued on the shadow the cancelled run built") +} + // A row a quiesced run wrote is refused by a live run, and a row a live // run wrote is refused by a quiesced run, both as resume-row-decoding: // the two never continue each other's rows (ST-3). Nothing of the stopped From 88657e0abc1b017a40acbc292d9b9fb82a71bf2d Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Mon, 2 Nov 2026 17:06:13 +1100 Subject: [PATCH 2/2] copier, applier: pin every endedWith branch and document the stop Tests now cover a cause-cancelled Pause, a copier deadline, and a cause that already wraps the cancellation in both copies; the exported Run and Pause comments and the integration guide say what a stop returns. --- docs/schemabot-integration.md | 2 +- internal/cli/signal.go | 10 ++++++---- internal/cli/signal_integration_test.go | 4 ++-- pkg/applier/catchup.go | 20 +++++++++++++++----- pkg/applier/catchup_test.go | 5 +++++ pkg/applier/pause.go | 10 ++++++---- pkg/applier/pause_test.go | 20 ++++++++++++++++++++ pkg/copier/copier.go | 14 ++++++++++++-- pkg/copier/copier_test.go | 18 ++++++++++++++++++ 9 files changed, 85 insertions(+), 18 deletions(-) diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 23a2858f..94c3e722 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -444,7 +444,7 @@ backend ends in exactly one of three verdicts: | --- | --- | --- | | `outcome: executed-via-copy-and-swap` | The shadow was built, filled, verified, and swapped in under the source's name; `detail` says how many rows and chunks the verification pass covered, whether the retained source was dropped or kept under its `_old` name, whether the run first reconciled a lost slot, and — for a run that blocked writes (`--block-writes`) — that the source's writes were fenced for the run's length, since exit 0 otherwise reads as online-safe; `writes_blocked: true` says the same for a machine, on this outcome and on a failed run whose fence went on: one that left its fence standing, or that was abandoned after the fence went on. A blocked-writes run that stopped before its fence went on, for example at a table lock another session holds, reports `false`, since the table kept taking writes; its detail says so and that running the statement again starts from the beginning. A run whose done row is saved is this outcome even when what came after failed — `OnCheckpoint` handed the done row, or the release of the lock session; the detail then names that error. The error is nil | Done. Exit 0 at the CLI; the outcome token is distinct from `executed-natively` so an adapter can tell the two paths apart in its record | | `outcome: refused`, `reason: copy-and-swap-refused` | The route's own gates declined: the preflight found a table shape or an environment it will not run against, or a shadow operation refused mid-run to keep an invariant. `cause` is the route's token unchanged — a `copy-and-swap-…` cause from the preflight, a `shadow-…`, `cutover-…`, `resume-…`, or `copy-…` cause from the shadow operations — and `class` is its routing class. The error is nil | Route on `cause` with the two tables above and [refusal-classes.md](refusal-classes.md#copy-and-swap-refused-keyed-on-copyswaprefusalcause). A preflight refusal touched nothing. A refusal raised after the row was saved — a `cutover-…` or `resume-…` cause — leaves the row, the shadow, and the run's hold on the source in place: the write fence of a run that blocked writes, or the replication slot of a live run, which pins WAL until it is dropped; the cause says whether the next run clears it or the run is abandoned first. A missing privilege is not this reason: it is the same `insufficient-privileges` refusal the native route reports, with the exact `GRANT` in `detail` | -| `outcome: failed`, with `code` | The run stopped on an error that is not a refusal; the error is returned alongside the verdict. `code` is `cancelled-by-caller` when the caller's context ended — whatever cause the caller cancelled it with, `invariant-violation` when the engine caught a breach of its own contract, `data-fault` when the server rejected the rows as the statement would shape them — a `*pgconn.PgError` in SQLSTATE class `22` or `23`: a value the new type will not hold, a constraint the copied rows break — `pool-too-small` when the pool holds fewer than `migrate.CopySwapMinConns(Copier)` connections — refused before the preflight with nothing created, the error matching `errors.Is(err, executor.ErrPoolTooSmall)`; resize the pool rather than retry — and `execution-failed` for everything else — a divergence, an unsettled chunk, a rolled-back swap, a lost lock session, exhausted lock retries, a slot past its lag ceiling | `detail` says what the stop left and what to do with it. A data fault would stop the same way on every resume, so the Runner abandons the run itself before returning (`*schemachange.AbandonedError`): the shadow is dropped, the fence is taken off the source or the slot is dropped, and the checkpoint row is deleted — nothing is held; correct the data or the statement, then run it again without any acknowledgement, and the run starts from the beginning. Every other stop leaves the checkpoint row at its last save with the run's hold on the source — the fence or the slot: a divergence (`*checksum.DivergenceError`) is wedged in `verifying`, and a data fault whose abandon itself failed (the detail says the run "could not be abandoned") is held the same way; for both the detail says to correct the data or the statement and then start the run over with the restart acknowledgement (`--restart SCHEMA.TABLE`; `CopySwapOptions.Restart`), which drops the shadow, takes the fence off the source or drops the slot, deletes the row, and runs the change from the start. Any other stop says that running the same statement again resumes from the row. The `Run` outcome table above is the finer map. Match the wrapped error with `errors.Is`/`errors.As` for the finer routing — the verdict's `code` is the coarse class | +| `outcome: failed`, with `code` | The run stopped on an error that is not a refusal; the error is returned alongside the verdict. `code` is `cancelled-by-caller` when the caller's context ended — whatever cause the caller cancelled it with: the code says only that the context ended, so an adapter that cancels with reasons of its own (a lost lease, a failed health gate) tells them apart with `errors.Is(err, )` on the returned error, `invariant-violation` when the engine caught a breach of its own contract, `data-fault` when the server rejected the rows as the statement would shape them — a `*pgconn.PgError` in SQLSTATE class `22` or `23`: a value the new type will not hold, a constraint the copied rows break — `pool-too-small` when the pool holds fewer than `migrate.CopySwapMinConns(Copier)` connections — refused before the preflight with nothing created, the error matching `errors.Is(err, executor.ErrPoolTooSmall)`; resize the pool rather than retry — and `execution-failed` for everything else — a divergence, an unsettled chunk, a rolled-back swap, a lost lock session, exhausted lock retries, a slot past its lag ceiling | `detail` says what the stop left and what to do with it. A data fault would stop the same way on every resume, so the Runner abandons the run itself before returning (`*schemachange.AbandonedError`): the shadow is dropped, the fence is taken off the source or the slot is dropped, and the checkpoint row is deleted — nothing is held; correct the data or the statement, then run it again without any acknowledgement, and the run starts from the beginning. Every other stop leaves the checkpoint row at its last save with the run's hold on the source — the fence or the slot: a divergence (`*checksum.DivergenceError`) is wedged in `verifying`, and a data fault whose abandon itself failed (the detail says the run "could not be abandoned") is held the same way; for both the detail says to correct the data or the statement and then start the run over with the restart acknowledgement (`--restart SCHEMA.TABLE`; `CopySwapOptions.Restart`), which drops the shadow, takes the fence off the source or drops the slot, deletes the row, and runs the change from the start. Any other stop says that running the same statement again resumes from the row. The `Run` outcome table above is the finer map. Match the wrapped error with `errors.Is`/`errors.As` for the finer routing — the verdict's `code` is the coarse class | The adapter supplies the route's policy on `migrate.Options.CopySwap`: the connection the pool was dialed with (the route opens dedicated sessions — the table lock, the copy workers, diff --git a/internal/cli/signal.go b/internal/cli/signal.go index e352e70a..069aa96d 100644 --- a/internal/cli/signal.go +++ b/internal/cli/signal.go @@ -14,10 +14,12 @@ import ( const exitCodeInterrupted = 128 + int(syscall.SIGINT) // errInterrupted is the cancellation cause recorded when an operator's -// signal stops a run. It wraps context.Canceled, so a caller that reads -// the cause (context.Cause) sees the same ordinary cancellation as one -// that reads ctx.Err(), and the run's verdict is cancelled-by-caller -// whichever phase the signal lands in. +// signal stops a run. The copier and the catch-up report any cause as a +// cancellation, so the run's verdict is cancelled-by-caller whichever +// phase the signal lands in. Wrapping context.Canceled as well keeps the +// cause readable as a cancellation to code that reads it directly +// (context.Cause), and lets the engine report it as it stands rather than +// appending the cancellation a second time. var errInterrupted = fmt.Errorf("interrupted by signal: %w", context.Canceled) // Interruptible derives a context that an operator's SIGINT or SIGTERM diff --git a/internal/cli/signal_integration_test.go b/internal/cli/signal_integration_test.go index 563f16c9..c2c1d62c 100644 --- a/internal/cli/signal_integration_test.go +++ b/internal/cli/signal_integration_test.go @@ -25,8 +25,8 @@ const copyStartedDeadline = 30 * time.Second // An operator's first signal while the copy is running stops the run // through its own stop path and reports the cancellation as the caller's: -// the copier returns the context's cause, and the cause must still read -// as a cancellation, or the verdict says the run broke. +// the copier's error carries the signal's cause and reads as a +// cancellation, or the verdict says the run broke. func TestMigrateSignalDuringCopyReportsCancelledByCaller(t *testing.T) { url := testutil.StartPostgres(t) pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) diff --git a/pkg/applier/catchup.go b/pkg/applier/catchup.go index bcf13724..b6b5130c 100644 --- a/pkg/applier/catchup.go +++ b/pkg/applier/catchup.go @@ -249,10 +249,13 @@ func (c *Catchup) Confirmed() decode.LSN { // // Run never returns nil: a catch-up has no end of its own, so its owner // stops it by ending ctx once the shadow is close enough to cut over, and -// Run returns the cause ctx ended with, wrapped, once the stream reports -// it. A stopped catch-up leaves the stream failed and the slot at the last -// confirmed position, which is where the next stream resumes. A cycle -// error that is not the stop itself travels with the stop, so a caller +// Run returns the stop, wrapped, once the stream reports it. The stop +// answers errors.Is for ctx.Err() (context.Canceled or +// context.DeadlineExceeded), and also for the cause when ctx was cancelled +// with one (context.WithCancelCause). A stopped catch-up leaves the stream +// failed and the slot at the last confirmed position, which is where the +// next stream resumes. A cycle error that is not the stop itself travels +// with the stop, so a caller // checks the failure sentinels — ErrInvariantViolation, ErrTableLockLost, // ErrBatchDeferred — before it reads the stop as clean. The whole run is // bound to the table lock session, so losing the lock ends the cycle in @@ -319,7 +322,14 @@ func stopped(ctx context.Context, schema, table string, err error) error { // of its own (context.WithCancelCause) gets that cause wrapped together // with ctx.Err(), so the error reads as the cause and as a cancellation // (errors.Is(err, context.Canceled)) alike; a caller that cancelled -// plainly, or whose deadline passed, gets ctx.Err() as it stands. +// plainly, or whose deadline passed, gets ctx.Err() as it stands, and so +// does a cause that already wraps it. +// +// It reads only that ctx ended, whoever ended it: a cancellation with a +// cause always reads as a stop. An owner that cancels a context it derived +// for this package because something failed must report that failure +// itself, as the table lock session's loss and the runner's capture and +// save failures are; otherwise the failure reads as the caller's stop. func endedWith(ctx context.Context) error { cause := context.Cause(ctx) if errors.Is(cause, ctx.Err()) { diff --git a/pkg/applier/catchup_test.go b/pkg/applier/catchup_test.go index e15262be..47afb683 100644 --- a/pkg/applier/catchup_test.go +++ b/pkg/applier/catchup_test.go @@ -123,6 +123,11 @@ func TestEndedWithKeepsACancellationReadableAsOne(t *testing.T) { assert.ErrorIs(t, err, cause) assert.ErrorIs(t, err, context.Canceled) assert.EqualError(t, err, "owner is ready to cut over (context canceled)") + + interrupted := fmt.Errorf("interrupted by signal: %w", context.Canceled) + wrapped, cancelWrapped := context.WithCancelCause(t.Context()) + cancelWrapped(interrupted) + assert.Same(t, interrupted, endedWith(wrapped), "a cause that is already a cancellation is reported as it stands") } // The copy has landed when its cut frontier is complete and no claimed diff --git a/pkg/applier/pause.go b/pkg/applier/pause.go index 02e5c34e..1d765875 100644 --- a/pkg/applier/pause.go +++ b/pkg/applier/pause.go @@ -20,10 +20,12 @@ import ( // that waits for the catch-up to confirm past its own work resumes first, // since the confirmation needs the flushes. Pause waits for a flush in // flight to commit or fail, or for ctx to end; a second Pause waits for the -// first's resume; resume is safe to call more than once. Pause before Run -// takes the hold at once, and a Run started under it flushes nothing until -// resume. Status reports the hold as FlushesHeld from Pause's return until -// the resume. +// first's resume; resume is safe to call more than once. A Pause whose ctx +// ends first takes no hold and returns an error that answers errors.Is for +// ctx.Err(), and also for the cause when ctx was cancelled with one. Pause +// before Run takes the hold at once, and a Run started under it flushes +// nothing until resume. Status reports the hold as FlushesHeld from Pause's +// return until the resume. func (c *Catchup) Pause(ctx context.Context) (resume func(), err error) { if err := c.holdFlushes(ctx); err != nil { return nil, err diff --git a/pkg/applier/pause_test.go b/pkg/applier/pause_test.go index efca2d53..df6f87ee 100644 --- a/pkg/applier/pause_test.go +++ b/pkg/applier/pause_test.go @@ -2,6 +2,7 @@ package applier import ( "context" + "errors" "testing" "github.com/stretchr/testify/assert" @@ -31,3 +32,22 @@ func TestCatchupPauseTakesAFreeGateOnAnEndedContext(t *testing.T) { c.releaseFlushes() } } + +// A Pause that waits behind a flush in flight and whose context is cancelled +// with a cause takes no hold and returns an error that carries the cause +// and still reads as a cancellation, so a live repair stopped this way is +// reported as the caller's stop. +func TestCatchupPauseCancelledWithACauseReadsAsCancelled(t *testing.T) { + c := &Catchup{flushes: newFlushGate(), flusher: &Flusher{}} + require.True(t, c.tryFlushes(), "a flush in flight holds the gate") + ctx, cancel := context.WithCancelCause(t.Context()) + ownerStop := errors.New("owner is ready to cut over") + cancel(ownerStop) + + _, err := c.Pause(ctx) + + require.Error(t, err) + assert.ErrorIs(t, err, ownerStop, "the owner's cause is carried") + assert.ErrorIs(t, err, context.Canceled, "a cancel with a cause is still a cancellation") + assert.False(t, c.Status().FlushesHeld) +} diff --git a/pkg/copier/copier.go b/pkg/copier/copier.go index 5b6ae6fc..8f74191f 100644 --- a/pkg/copier/copier.go +++ b/pkg/copier/copier.go @@ -256,7 +256,10 @@ func (c *Copier) Position() Position { // that was running keeps running until it finishes or hits the transaction's // statement_timeout, and only then is its transaction rolled back; the // server-side end of a cancelled chunk is bounded by that timeout, not by -// Run's return. +// Run's return. A Run stopped by ctx returns an error that answers +// errors.Is for ctx.Err() (context.Canceled or context.DeadlineExceeded), +// and also for the cause when ctx was cancelled with one +// (context.WithCancelCause). func (c *Copier) Run(ctx context.Context, pool *pgxpool.Pool) error { if err := c.start(); err != nil { return err @@ -305,7 +308,14 @@ func stopReason(ctx, workers context.Context) error { // of its own (context.WithCancelCause) gets that cause wrapped together // with ctx.Err(), so the error reads as the cause and as a cancellation // (errors.Is(err, context.Canceled)) alike; a caller that cancelled -// plainly, or whose deadline passed, gets ctx.Err() as it stands. +// plainly, or whose deadline passed, gets ctx.Err() as it stands, and so +// does a cause that already wraps it. +// +// It reads only that ctx ended, whoever ended it: a cancellation with a +// cause always reads as a stop. An owner that cancels a context it derived +// for this package because something failed must report that failure +// itself, as the table lock session's loss and the runner's capture and +// save failures are; otherwise the failure reads as the caller's stop. func endedWith(ctx context.Context) error { cause := context.Cause(ctx) if errors.Is(cause, ctx.Err()) { diff --git a/pkg/copier/copier_test.go b/pkg/copier/copier_test.go index 1ebcfb5d..29be651a 100644 --- a/pkg/copier/copier_test.go +++ b/pkg/copier/copier_test.go @@ -3,6 +3,7 @@ package copier import ( "context" "errors" + "fmt" "math" "testing" "time" @@ -131,3 +132,20 @@ func TestStopReasonKeepsACancellationReadableAsOne(t *testing.T) { assert.EqualError(t, err, "owner is ready to cut over (context canceled)") }) } + +// A passed deadline is reported as context.DeadlineExceeded as it stands, +// and a cause that already wraps the cancellation — the CLI's signal cause +// is one — is returned as it stands rather than wrapped a second time. +func TestEndedWithReportsADeadlineAndAWrappedCancellationAsTheyStand(t *testing.T) { + expired, cancelExpired := context.WithDeadline(t.Context(), time.Unix(0, 0)) + defer cancelExpired() + <-expired.Done() + assert.Equal(t, context.DeadlineExceeded, endedWith(expired)) + + interrupted := fmt.Errorf("interrupted by signal: %w", context.Canceled) + caused, cancel := context.WithCancelCause(t.Context()) + cancel(interrupted) + err := endedWith(caused) + assert.Same(t, interrupted, err, "a cause that is already a cancellation is reported as it stands") + assert.False(t, errors.Is(err, context.DeadlineExceeded)) +}