From 4d8c1d6e8c0c120e34bf43c81ad5998aaddddfdf Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Thu, 29 Oct 2026 12:53:48 +1100 Subject: [PATCH 1/2] reconcile: clear the shadow above a partial watermark before the recopy pass The pass compares only through the watermark, so a shadow row above it holding a unique value the source has since moved into a differing chunk made every recopy of every reconcile collide; only --restart escaped. --- docs/copy-and-swap-design.md | 12 +-- docs/invariants.md | 20 +++-- pkg/checksum/recopy_retry.go | 4 +- pkg/copier/resume.go | 25 ++++++ pkg/schemachange/reconcile.go | 70 ++++++++++++---- .../reconcile_clear_integration_test.go | 83 +++++++++++++++++++ 6 files changed, 181 insertions(+), 33 deletions(-) create mode 100644 pkg/schemachange/reconcile_clear_integration_test.go diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index 1713edf9..c861e5c7 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -553,11 +553,11 @@ between attempts, no backoff), counts every collision in `Outcome.Collisions`, r comparison the last recopy was cut from as `Outcome.Report`, and ends in the last collision's error, as the server raised it, when every attempt collides — the row then stays in `reconciling` and the next attempt restarts the reconcile. Comparing again clears a collision -only when the colliding shadow row is inside the comparison. A reconcile on resume compares -through the row's watermark and runs before the copier clears the shadow above it, so a value -the source moves out of a shadow row above a partial watermark collides on every attempt of -every reconcile until the source moves it back; the run stays in `reconciling`, and `--restart` -starts it over (ST-4 *Known limit*). Under `repair` the collision is the pass's error: a repair +only when the colliding shadow row is inside the comparison, and a reconcile on resume compares +only through the row's watermark, so before its pass it clears the shadow above a partial +watermark through `copier.ClearAbove` — the same clear the copier that resumes from the +watermark runs first (CO-4) — leaving every shadow row inside the comparison. Under `repair` the +collision is the pass's error: a repair assumes the source rows it recopies hold still. ### D15 — Capture changes with pgoutput @@ -671,7 +671,7 @@ next pass's to catch. | `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/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/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, and when anything landed clears the shadow above the watermark through `copier.ClearAbove` (CO-4), so no shadow row lies outside the pass, then runs one `checksum` pass at or below the watermark under `recopy` — 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. | — | Each producing package owns its types. `pkg/schemachange` imports every producer; no producer diff --git a/docs/invariants.md b/docs/invariants.md index 63218c22..7771ee3d 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -1098,7 +1098,9 @@ dies restarts as a reconcile, and the slot it is about to make is owned by the r moment it exists (ST-3) — replaces whatever slot bears the derived name with a fresh one through `decode.ReplaceSlot`, keeping the publication in place, since the source's writers look it up as they run and one dropped under them fails the write, and -— when the watermark says anything landed — runs one `checksum` pass at or below the watermark +— when the watermark says anything landed — clears the shadow above the watermark through +`copier.ClearAbove`, as the copier that resumes from it would first (CO-4), so no shadow row +lies outside the pass that follows, then runs one `checksum` pass at or below the watermark under the `recopy` policy (D14), recopying only the chunks the gap touched without a reread, since the source is live and the new slot carries every write that lands after the recopy, then writes the row back to `copying` or `catching_up` by the watermark with the new slot's @@ -1112,8 +1114,9 @@ refuses-replaced-relations-before-its-marker, leaves-the-slot-it-created-owned-by-the-row, resumes-in-copying-below-a-partial-watermark, refuses-a-row-of-another-source-shape, stops-when-the-lock-is-lost-while-the-drop-waits, refuses-a-foreign-slot-of-the-name, refuses-the-slot-while-another-backend-holds-it, -reads-a-slot-the-server-reports-lost-as-reconcile, and -survives-a-unique-value-moved-during-its-recopy tests). The recopy pass compares a chunk and +reads-a-slot-the-server-reports-lost-as-reconcile, +survives-a-unique-value-moved-during-its-recopy, and +clears-the-shadow-above-a-partial-watermark-before-its-recopy tests). The recopy pass compares a chunk and recopies it in separate statements against a live source, so a source that moves a unique value from a row in a chunk the compare read equal to a row in a differing chunk, between the two, makes the recopy collide with a shadow row the compare left alone; the one recopy @@ -1121,12 +1124,11 @@ transaction rolls back, the pass compares again, and the recopy cut from that co covers both chunks and lands (CO-3). A source that moves a value into the differing chunks before every recopy exhausts the pass's bounded attempts; the pass then ends in the last collision's error with the row still in `reconciling` and the shadow unchanged, and the next -attempt restarts the reconcile — closed, never a shadow with a row the source lacks. *Known -limit:* on resume the reconcile runs before the copier clears the shadow rows above a partial -watermark, and the pass compares only through that watermark, so a unique value the source -moves out of a shadow row above it into a differing chunk collides on every attempt of every -reconcile until the source moves it back; the run stays in `reconciling`, closed, and -`--restart` (`CopySwapOptions.Restart`) starts it over. *Source:* +attempt restarts the reconcile — closed, never a shadow with a row the source lacks. The +pass compares only through the watermark, which is why the reconcile clears the shadow above +a partial watermark before it: a unique value the source moved out of a shadow row above the +watermark into a differing chunk below it would otherwise collide with a row no comparison +can reach, on every attempt of every reconcile, until the source moved it back. *Source:* [low-level-design § failover](low-level-design.md#failover-during-migration-what-survives-and-what-doesnt). ### ST-5 — The swap is gated on a fidelity checklist, not just the checksum diff --git a/pkg/checksum/recopy_retry.go b/pkg/checksum/recopy_retry.go index 71ae4d53..4aef3b23 100644 --- a/pkg/checksum/recopy_retry.go +++ b/pkg/checksum/recopy_retry.go @@ -52,7 +52,9 @@ func collided(err error) bool { // no other. Comparing again clears a collision only when the colliding row // lies inside the comparison: a shadow row above the watermark the pass // reads through is never compared, so a collision with one repeats on -// every attempt and the pass ends in its error. The Outcome returned +// every attempt and the pass ends in its error — which is why a caller +// that compares through a partial watermark clears the shadow above it +// first (copier.ClearAbove). The Outcome returned // carries the comparison the committed recopy acted on, the repairs it // committed, and the attempts that collided, with the error when the last // one did; a recopy that fails for diff --git a/pkg/copier/resume.go b/pkg/copier/resume.go index 64bb14b4..f24049c0 100644 --- a/pkg/copier/resume.go +++ b/pkg/copier/resume.go @@ -7,9 +7,34 @@ import ( "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" + "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/preflight" ) +// ClearAbove removes every shadow row whose key lies above from, as a +// resumed Run does before it cuts its first chunk, and nothing else. It is +// for a caller that compares the shadow against the source through from +// before any copier resumes — reconcile mode (ST-4) — since a shadow row +// above the watermark is outside such a comparison yet still holds its +// values: a unique value the source has since moved out of it into a chunk +// the comparison finds differing would make every recopy of that chunk +// collide with it, and no comparison through from can clear the collision. +// With the shadow clear above from, the rows the comparison reads are the +// only rows the shadow has, and the copier that resumes from the same +// watermark lands the rest. ClearAbove refuses what NewCopier refuses and +// bounds each batch as a chunk transaction is bounded; it runs under the +// lock session's Bind context, so a lost lock cancels the batch in flight +// (LK-1). A complete watermark has nothing above it and clears nothing. +func ClearAbove(ctx context.Context, pool *pgxpool.Pool, target preflight.CopySwapTarget, shadow Shadow, lock *dbconn.TableLockSession, from Watermark, opts Options) error { + c, err := NewCopier(target, shadow, lock, from, opts) + if err != nil { + return err + } + ctx, unbind := c.lock.Bind(ctx) + defer unbind() + return c.clearAbove(ctx, pool, from) +} + // clearAbove removes every shadow row whose key lies above the resume // watermark before the resumed copy cuts its first chunk. The earlier run's // unlanded chunks may have committed rows there, and the applier treats diff --git a/pkg/schemachange/reconcile.go b/pkg/schemachange/reconcile.go index b8243844..64b53f12 100644 --- a/pkg/schemachange/reconcile.go +++ b/pkg/schemachange/reconcile.go @@ -10,6 +10,7 @@ import ( "github.com/block/pg-sprite/pkg/checkpoint" "github.com/block/pg-sprite/pkg/checksum" + "github.com/block/pg-sprite/pkg/copier" "github.com/block/pg-sprite/pkg/dbconn" "github.com/block/pg-sprite/pkg/decode" "github.com/block/pg-sprite/pkg/preflight" @@ -18,7 +19,8 @@ import ( // Reconciler is reconcile mode (ST-4): the recovery for a run whose // replication slot is gone or lost, keeping the shadow and the copier's // watermark and replacing only what the slot carried. One Reconciler -// serves one built shadow; its verifier is built up front so a reconcile +// serves one built shadow; its verifier is built up front, and the options +// its clear of the shadow runs under are validated then, so a reconcile // never starts with options that cannot run a pass. type Reconciler struct { cfg dbconn.Config @@ -27,6 +29,10 @@ type Reconciler struct { target preflight.CopySwapTarget shadow BuiltShadow verifier *checksum.Verifier + // clear bounds the clear of the shadow above a partial watermark as + // the pass's options bound the pass: the same timeouts, the pass's + // chunk size as the batch size. + clear copier.Options } // Reconciled is what a completed reconcile hands back: the slot change @@ -66,7 +72,23 @@ func NewReconciler(cfg dbconn.Config, lock *dbconn.TableLockSession, store *chec if err != nil { return nil, err } - return &Reconciler{cfg: cfg, lock: lock, store: store, target: target, shadow: shadow, verifier: verifier}, nil + clear := clearOptions(opts) + if err := clear.Validate(); err != nil { + return nil, err + } + return &Reconciler{cfg: cfg, lock: lock, store: store, target: target, shadow: shadow, verifier: verifier, clear: clear}, nil +} + +// clearOptions bounds the reconcile's clear of the shadow as its pass is +// bounded. The copier's worker count has no bearing on a clear, which +// runs one batch at a time, and is left to its default. +func clearOptions(opts checksum.Options) copier.Options { + return copier.Options{ + LockTimeout: opts.LockTimeout, + StatementTimeout: opts.StatementTimeout, + Chunker: opts.Chunker, + Clock: opts.Clock, + } } // Reconcile recovers from slot loss for the run cp records, in an order a @@ -83,18 +105,26 @@ func NewReconciler(cfg dbconn.Config, lock *dbconn.TableLockSession, store *chec // success, a half-made one from an earlier attempt goes too. // 4. A new slot is created under the same name. Every change committed // after its consistent point replays when the stream is reopened. -// 5. When anything has landed, one recopy pass runs through the row's -// watermark under checksum.DivergenceRecopy: every chunk that differs -// is recopied from the source, which is where the changes the lost -// slot never delivered are, and is not read again. The source keeps -// taking writes throughout; a write that lands in a chunk after its -// recopy is behind the new slot's consistent point and replays when -// the stream reopens, so the pass has nothing to prove about it. A -// unique value the source moves from a chunk the pass read equal into -// a differing chunk, between the comparison and the recopy, makes the -// recopy collide with the shadow row still holding it; the pass then -// compares again and recopies both chunks, a bounded number of times. -// 6. The row is saved out of PhaseReconciling: PhaseCatchingUp when the +// 5. When anything has landed, the shadow is cleared above the row's +// watermark, as the copier that resumes from it would clear it first +// (CO-4): the rows the earlier run's unlanded chunks committed there +// lie outside the pass that follows, and one still holding a unique +// value the source has since moved into a chunk the pass recopies +// would collide with every recopy of it, with no comparison through +// the watermark able to clear the collision. A complete watermark has +// nothing above it. +// 6. Then one recopy pass runs through the row's watermark under +// checksum.DivergenceRecopy: every chunk that differs is recopied +// from the source, which is where the changes the lost slot never +// delivered are, and is not read again. The source keeps taking +// writes throughout; a write that lands in a chunk after its recopy is +// behind the new slot's consistent point and replays when the stream +// reopens, so the pass has nothing to prove about it. A unique value +// the source moves from a chunk the pass read equal into a differing +// chunk, between the comparison and the recopy, makes the recopy +// collide with the shadow row still holding it; the pass then compares +// again and recopies both chunks, a bounded number of times. +// 7. The row is saved out of PhaseReconciling: PhaseCatchingUp when the // watermark is complete, PhaseCopying when the copier still has keys // to land, with the new slot's consistent point as the last applied // position. @@ -216,13 +246,19 @@ func (r *Reconciler) confirmShadowRelations(ctx context.Context, pool *pgxpool.P return nil } -// recopy runs the recopy pass through the row's watermark, or no pass -// when nothing has landed: a shadow with no rows has nothing the lost -// slot could have left stale. +// recopy clears the shadow above the row's watermark and runs the recopy +// pass through it, or does neither when nothing has landed: with no +// watermark there is no pass, so no row above one to collide with, and +// whatever the earlier run committed is the resumed copier's to clear. func (r *Reconciler) recopy(ctx context.Context, pool *pgxpool.Pool, cp checkpoint.Checkpoint) (checksum.Outcome, error) { if !cp.Watermark.Valid() { return checksum.Outcome{}, nil } + // INV: CO-4 — every shadow row the pass does not compare is removed + // before the pass, so none can hold a value a recopy puts back. + if err := copier.ClearAbove(ctx, pool, r.target, r.shadow, r.lock, cp.Watermark, r.clear); err != nil { + return checksum.Outcome{}, err + } // INV: CO-3 — the policy is stated: reconcile mode recopies. return r.verifier.Check(ctx, pool, cp.Watermark, checksum.DivergenceRecopy) } diff --git a/pkg/schemachange/reconcile_clear_integration_test.go b/pkg/schemachange/reconcile_clear_integration_test.go new file mode 100644 index 00000000..5d2b1f91 --- /dev/null +++ b/pkg/schemachange/reconcile_clear_integration_test.go @@ -0,0 +1,83 @@ +package schemachange_test + +import ( + "context" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/checkpoint" + "github.com/block/pg-sprite/pkg/checksum" + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/decode" + "github.com/block/pg-sprite/pkg/schemachange" +) + +// A row whose watermark covers part of the key space can sit over a shadow +// that holds rows above it: chunks the earlier run committed but never +// checkpointed. The pass compares only through the watermark, so such a +// row is outside every comparison and yet holds its values. When the +// source has since moved a unique value out of one of those rows into a +// chunk the pass finds differing, the recopy would collide with the row +// above on every attempt, with no comparison able to clear it. The +// reconcile clears the shadow above the watermark before its pass, as the +// copier does before it resumes, so the recopy lands first time; the row +// resumes in copying at the same watermark, and the copier that resumes +// from it lands the rest. +func TestReconcileClearsTheShadowAboveAPartialWatermarkBeforeItsRecopy(t *testing.T) { + f := newReconcileFixture(t, 500) + shadow := f.buildShadow(t) + copyAll(t, f, newCopier(t, f, shadow)) + // The run checkpointed the copier at key 300 when its slot was lost; + // the shadow holds keys 301 through 500 from chunks the row never + // recorded. + cp := f.row(shadow, checkpoint.PhaseCopying, copier.NewWatermark(300), "", 0) + require.NoError(t, f.store.Save(t.Context(), f.lock, cp)) + + // Nothing captures the source while the slot is gone: a chunk below + // the watermark changes, and the unique value of key 400 — a shadow + // row above the watermark — moves to key 160, in that differing + // chunk. + f.exec(t, `UPDATE %s SET amount = 0 WHERE id = 150`) + f.exec(t, `UPDATE %s SET uniq = -400 WHERE id = 400`) + f.exec(t, `UPDATE %s SET uniq = 400 WHERE id = 160`) + + reconciler, err := schemachange.NewReconciler(f.cfg, f.lock, f.store, f.target, shadow, checksum.Options{ + Chunker: copier.ChunkerOptions{InitialRows: 100, MaxRows: 100}, + }) + require.NoError(t, err) + reconciled, err := reconciler.Reconcile(t.Context(), f.pool, cp) + require.NoError(t, err) + t.Cleanup(func() { + ctx := context.WithoutCancel(t.Context()) + assert.NoError(t, reconciled.Slot.Close(ctx)) + assert.NoError(t, decode.DropSlot(ctx, f.cfg, f.pool, reconciled.Slot.Name())) + }) + + assert.Equal(t, 0, reconciled.Outcome.Collisions, "the shadow row holding the moved value was cleared before the recopy") + assertRepairsCoverExactly(t, reconciled.Outcome.Repairs, []int64{150, 160}, []int64{50, 250}) + assert.Equal(t, checkpoint.PhaseCopying, reconciled.Checkpoint.Phase, "keys above the watermark are the copier's to land") + assert.Equal(t, copier.NewWatermark(300), reconciled.Checkpoint.Watermark, "the watermark survived the reconcile") + assert.Equal(t, 0, f.shadowRowsAbove(t, shadow, 300), "the reconcile cleared every shadow row above the watermark") + + // The copier resumes from the reconciled row's watermark and the + // shadow converges. + resumed, err := copier.NewCopier(f.target, shadow, f.lock, reconciled.Checkpoint.Watermark, copier.Options{ + Workers: 2, + Chunker: copier.ChunkerOptions{InitialRows: 100, MaxRows: 100}, + }) + require.NoError(t, err) + copyAll(t, f, resumed) + f.assertConverged(t, shadow) +} + +// shadowRowsAbove counts the shadow rows whose key lies above key. +func (f reconcileFixture) shadowRowsAbove(t *testing.T, shadow schemachange.BuiltShadow, key int64) int { + t.Helper() + table := pgx.Identifier{f.table.Schema, shadow.ShadowTable()}.Sanitize() + var n int + require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT count(*) FROM `+table+` WHERE id > $1`, key).Scan(&n)) + return n +} From fd8a9d5e2e6bb92db165e039bcdd961eb520cc29 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Tue, 3 Nov 2026 18:03:23 +1100 Subject: [PATCH 2/2] reconcile: clear above the watermark before touching the slot A failed clear now ends the attempt with no slot made; the failed-clear boundary and the nothing-landed skip are pinned by tests. --- docs/copy-and-swap-design.md | 7 +- docs/invariants.md | 20 ++-- pkg/copier/resume.go | 5 +- pkg/schemachange/reconcile.go | 68 +++++++---- .../reconcile_clear_integration_test.go | 108 ++++++++++++++++++ 5 files changed, 172 insertions(+), 36 deletions(-) diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index c861e5c7..bb2bad82 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -556,9 +556,10 @@ error, as the server raised it, when every attempt collides — the row then sta only when the colliding shadow row is inside the comparison, and a reconcile on resume compares only through the row's watermark, so before its pass it clears the shadow above a partial watermark through `copier.ClearAbove` — the same clear the copier that resumes from the -watermark runs first (CO-4) — leaving every shadow row inside the comparison. Under `repair` the -collision is the pass's error: a repair -assumes the source rows it recopies hold still. +watermark runs first (CO-4) — leaving every shadow row inside the comparison. The clear runs +before the reconcile touches the slot, so a new slot retains WAL only while the pass runs. +Under `repair` the collision is the pass's error: a repair assumes the source rows it recopies +hold still. ### D15 — Capture changes with pgoutput diff --git a/docs/invariants.md b/docs/invariants.md index 7771ee3d..f9539008 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -202,7 +202,8 @@ transaction; chunks are cut and registered under one lock, and the ledger refuse does not start just above the frontier (`copy-frontier-broken`, a typed `copier.RefusalCause`), so the frontier never runs ahead of an unregistered chunk; a chunk is registered in flight before its transaction begins and leaves the in-flight set only when it commits — a chunk whose transaction did not commit stays in flight, so its keys -never read as landed; a resumed copy first deletes every shadow row above W in bounded batches, +never read as landed; a resumed copy, and a reconcile before its recopy pass over a partial +watermark (`copier.ClearAbove`, ST-4), first deletes every shadow row above W in bounded batches, each in its own guarded transaction, before its first chunk is cut — the whole shadow when nothing landed, since a zero watermark says nothing about what an earlier run left above it — so every key above W is genuinely uncut when the applier starts discarding for it; the first clear @@ -1123,12 +1124,17 @@ two, makes the recopy collide with a shadow row the compare left alone; the one transaction rolls back, the pass compares again, and the recopy cut from that comparison covers both chunks and lands (CO-3). A source that moves a value into the differing chunks before every recopy exhausts the pass's bounded attempts; the pass then ends in the last -collision's error with the row still in `reconciling` and the shadow unchanged, and the next -attempt restarts the reconcile — closed, never a shadow with a row the source lacks. The -pass compares only through the watermark, which is why the reconcile clears the shadow above -a partial watermark before it: a unique value the source moved out of a shadow row above the -watermark into a differing chunk below it would otherwise collide with a row no comparison -can reach, on every attempt of every reconcile, until the source moved it back. *Source:* +collision's error with the row still in `reconciling`, no recopy committed, and the watermark +unmoved, and the next attempt restarts the reconcile — closed, never a shadow with a row the +source lacks. The pass compares only through the watermark, which is why the reconcile clears +the shadow above a partial watermark before it, and before any slot command so a new slot +retains WAL only while the pass runs: a unique value the source moved out of a shadow row +above the watermark into a differing chunk below it would otherwise collide with a row no +comparison can reach, on every attempt of every reconcile, until the source moved it back. A +clear that fails, a lock timeout behind a writer still holding the shadow included, ends the +attempt before the marker with the row and its watermark as they were +(retries-after-the-clear-fence-times-out test); with nothing landed there is no pass and no +clear (with-nothing-landed-leaves-the-shadow-alone test). *Source:* [low-level-design § failover](low-level-design.md#failover-during-migration-what-survives-and-what-doesnt). ### ST-5 — The swap is gated on a fidelity checklist, not just the checksum diff --git a/pkg/copier/resume.go b/pkg/copier/resume.go index f24049c0..6029b4f2 100644 --- a/pkg/copier/resume.go +++ b/pkg/copier/resume.go @@ -22,7 +22,10 @@ import ( // With the shadow clear above from, the rows the comparison reads are the // only rows the shadow has, and the copier that resumes from the same // watermark lands the rest. ClearAbove refuses what NewCopier refuses and -// bounds each batch as a chunk transaction is bounded; it runs under the +// bounds each batch as a chunk transaction is bounded. Of opts it honors +// LockTimeout, StatementTimeout, Chunker (the batch size) and Clock; the +// clear runs one batch at a time and reports no progress, so Workers and +// Tracker have no effect. It runs under the // lock session's Bind context, so a lost lock cancels the batch in flight // (LK-1). A complete watermark has nothing above it and clears nothing. func ClearAbove(ctx context.Context, pool *pgxpool.Pool, target preflight.CopySwapTarget, shadow Shadow, lock *dbconn.TableLockSession, from Watermark, opts Options) error { diff --git a/pkg/schemachange/reconcile.go b/pkg/schemachange/reconcile.go index 64b53f12..2bb5fdd6 100644 --- a/pkg/schemachange/reconcile.go +++ b/pkg/schemachange/reconcile.go @@ -29,10 +29,10 @@ type Reconciler struct { target preflight.CopySwapTarget shadow BuiltShadow verifier *checksum.Verifier - // clear bounds the clear of the shadow above a partial watermark as - // the pass's options bound the pass: the same timeouts, the pass's + // clearOpts bounds the clear of the shadow above a partial watermark + // as the pass's options bound the pass: the same timeouts, the pass's // chunk size as the batch size. - clear copier.Options + clearOpts copier.Options } // Reconciled is what a completed reconcile hands back: the slot change @@ -72,11 +72,11 @@ func NewReconciler(cfg dbconn.Config, lock *dbconn.TableLockSession, store *chec if err != nil { return nil, err } - clear := clearOptions(opts) - if err := clear.Validate(); err != nil { + clearOpts := clearOptions(opts) + if err := clearOpts.Validate(); err != nil { return nil, err } - return &Reconciler{cfg: cfg, lock: lock, store: store, target: target, shadow: shadow, verifier: verifier, clear: clear}, nil + return &Reconciler{cfg: cfg, lock: lock, store: store, target: target, shadow: shadow, verifier: verifier, clearOpts: clearOpts}, nil } // clearOptions bounds the reconcile's clear of the shadow as its pass is @@ -97,22 +97,24 @@ func clearOptions(opts checksum.Options) copier.Options { // 1. The source and shadow are confirmed to be the relations the built // shadow proved, so a row whose tables have been swapped or replaced // under their names is refused before anything is written. -// 2. The row is saved in PhaseReconciling, naming the slot the reconcile -// will own under the target's derived name, so a resume that finds -// it reconciles again rather than trusting whatever slot it finds, -// and the reaper sees the slot as the row's. -// 3. The slot under the derived name is dropped — a slot already gone is -// success, a half-made one from an earlier attempt goes too. -// 4. A new slot is created under the same name. Every change committed -// after its consistent point replays when the stream is reopened. -// 5. When anything has landed, the shadow is cleared above the row's +// 2. When anything has landed, the shadow is cleared above the row's // watermark, as the copier that resumes from it would clear it first // (CO-4): the rows the earlier run's unlanded chunks committed there // lie outside the pass that follows, and one still holding a unique // value the source has since moved into a chunk the pass recopies // would collide with every recopy of it, with no comparison through // the watermark able to clear the collision. A complete watermark has -// nothing above it. +// nothing above it. The clear runs before any slot command, so a new +// slot retains WAL only while the pass runs; a crash after it leaves +// the row in its earlier phase, and clearing again is harmless. +// 3. The row is saved in PhaseReconciling, naming the slot the reconcile +// will own under the target's derived name, so a resume that finds +// it reconciles again rather than trusting whatever slot it finds, +// and the reaper sees the slot as the row's. +// 4. The slot under the derived name is dropped — a slot already gone is +// success, a half-made one from an earlier attempt goes too. +// 5. A new slot is created under the same name. Every change committed +// after its consistent point replays when the stream is reopened. // 6. Then one recopy pass runs through the row's watermark under // checksum.DivergenceRecopy: every chunk that differs is recopied // from the source, which is where the changes the lost slot never @@ -135,7 +137,11 @@ func clearOptions(opts checksum.Options) copier.Options { // this target and shadow, is refused. On an error the slot the attempt // created, if any, stays under the row's name, so the reaper leaves it for // the next attempt to drop, and the row stays in PhaseReconciling for that -// attempt. +// attempt. The clear can end an attempt before any slot command, with the +// copier's own refusals (a lost lock, a replaced relation, conversion +// drift) or with an untyped lock timeout when a writer still holds the +// shadow past lock_timeout; either leaves the row and the watermark as +// they were, and a later attempt clears again. Retry the timeout. // // The whole sequence runs under the lock session's context: a lock lost // at any point cancels the statement in flight — a drop still waiting out @@ -167,6 +173,12 @@ func (r *Reconciler) reconcile(ctx context.Context, pool *pgxpool.Pool, cp check if err := r.confirmShadowRelations(ctx, pool); err != nil { return Reconciled{}, err } + // The clear needs only the lock and the shadow, so it runs before any + // slot command: a new slot then retains WAL only while the pass runs, + // and a clear that fails ends the attempt with no slot made. + if err := r.clearAbove(ctx, pool, cp); err != nil { + return Reconciled{}, err + } // INV: ST-4 — the durable marker precedes every change to the slot. name := r.target.DecodingName() marked := cp @@ -246,18 +258,24 @@ func (r *Reconciler) confirmShadowRelations(ctx context.Context, pool *pgxpool.P return nil } -// recopy clears the shadow above the row's watermark and runs the recopy -// pass through it, or does neither when nothing has landed: with no -// watermark there is no pass, so no row above one to collide with, and -// whatever the earlier run committed is the resumed copier's to clear. -func (r *Reconciler) recopy(ctx context.Context, pool *pgxpool.Pool, cp checkpoint.Checkpoint) (checksum.Outcome, error) { +// clearAbove clears the shadow above the row's watermark, or does nothing +// when nothing has landed: with no watermark there is no pass, so no row +// above one to collide with, and whatever the earlier run committed is the +// resumed copier's to clear. +func (r *Reconciler) clearAbove(ctx context.Context, pool *pgxpool.Pool, cp checkpoint.Checkpoint) error { if !cp.Watermark.Valid() { - return checksum.Outcome{}, nil + return nil } // INV: CO-4 — every shadow row the pass does not compare is removed // before the pass, so none can hold a value a recopy puts back. - if err := copier.ClearAbove(ctx, pool, r.target, r.shadow, r.lock, cp.Watermark, r.clear); err != nil { - return checksum.Outcome{}, err + return copier.ClearAbove(ctx, pool, r.target, r.shadow, r.lock, cp.Watermark, r.clearOpts) +} + +// recopy runs the recopy pass through the row's watermark, or nothing +// when nothing has landed. +func (r *Reconciler) recopy(ctx context.Context, pool *pgxpool.Pool, cp checkpoint.Checkpoint) (checksum.Outcome, error) { + if !cp.Watermark.Valid() { + return checksum.Outcome{}, nil } // INV: CO-3 — the policy is stated: reconcile mode recopies. return r.verifier.Check(ctx, pool, cp.Watermark, checksum.DivergenceRecopy) diff --git a/pkg/schemachange/reconcile_clear_integration_test.go b/pkg/schemachange/reconcile_clear_integration_test.go index 5d2b1f91..c56948ad 100644 --- a/pkg/schemachange/reconcile_clear_integration_test.go +++ b/pkg/schemachange/reconcile_clear_integration_test.go @@ -2,9 +2,12 @@ package schemachange_test import ( "context" + "math" "testing" + "time" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -73,6 +76,111 @@ func TestReconcileClearsTheShadowAboveAPartialWatermarkBeforeItsRecopy(t *testin f.assertConverged(t, shadow) } +// A writer still holding the shadow keeps the clear's first batch waiting +// past lock_timeout. The clear runs before any slot command, so the +// attempt ends with the server's lock timeout, no slot made, and the row +// and its watermark as they were. Once the writer ends, the next attempt +// clears, recopies without a collision, and lands. +func TestReconcileRetriesAfterTheClearFenceTimesOut(t *testing.T) { + f := newReconcileFixture(t, 500) + shadow := f.buildShadow(t) + copyAll(t, f, newCopier(t, f, shadow)) + cp := f.row(shadow, checkpoint.PhaseCopying, copier.NewWatermark(300), "", 0) + require.NoError(t, f.store.Save(t.Context(), f.lock, cp)) + f.exec(t, `UPDATE %s SET amount = 0 WHERE id = 150`) + f.exec(t, `UPDATE %s SET uniq = -400 WHERE id = 400`) + f.exec(t, `UPDATE %s SET uniq = 400 WHERE id = 160`) + + // A transaction that has deleted nothing still holds ROW EXCLUSIVE on + // the shadow, which the clear's SHARE ROW EXCLUSIVE fence waits on. + table := pgx.Identifier{f.table.Schema, shadow.ShadowTable()}.Sanitize() + writer, err := f.pool.Begin(t.Context()) + require.NoError(t, err) + defer func() { + // Redundant safety closer: the writer is rolled back below. + _ = writer.Rollback(context.WithoutCancel(t.Context())) + }() + _, err = writer.Exec(t.Context(), ` + DELETE FROM `+table+` + WHERE id = -1`) + require.NoError(t, err) + + reconciler, err := schemachange.NewReconciler(f.cfg, f.lock, f.store, f.target, shadow, checksum.Options{ + LockTimeout: time.Second, + Chunker: copier.ChunkerOptions{InitialRows: 100, MaxRows: 100}, + }) + require.NoError(t, err) + _, err = reconciler.Reconcile(t.Context(), f.pool, cp) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr) + assert.Equal(t, "55P03", pgErr.Code, "the clear's fence ran out of lock_timeout") + row := f.load(t, shadow) + assert.Equal(t, checkpoint.PhaseCopying, row.Phase, "the clear failed before the row was marked") + assert.Equal(t, copier.NewWatermark(300), row.Watermark) + assert.False(t, f.slotExists(t, f.target.DecodingName()), "the clear failed before any slot was made") + require.NoError(t, writer.Rollback(t.Context())) + + reconciled, err := reconciler.Reconcile(t.Context(), f.pool, row) + require.NoError(t, err) + t.Cleanup(func() { + ctx := context.WithoutCancel(t.Context()) + assert.NoError(t, reconciled.Slot.Close(ctx)) + assert.NoError(t, decode.DropSlot(ctx, f.cfg, f.pool, reconciled.Slot.Name())) + }) + assert.Equal(t, 0, reconciled.Outcome.Collisions) + assert.Equal(t, 0, f.shadowRowsAbove(t, shadow, 300)) +} + +// With nothing landed there is no pass, so the reconcile leaves the shadow +// as it is: clearing it is the resumed copier's job, which clears the +// whole key space from a zero watermark. +func TestReconcileWithNothingLandedLeavesTheShadowAlone(t *testing.T) { + f := newReconcileFixture(t, 500) + shadow := f.buildShadow(t) + copyAll(t, f, newCopier(t, f, shadow)) + cp := f.row(shadow, checkpoint.PhaseCopying, copier.Watermark{}, "", 0) + require.NoError(t, f.store.Save(t.Context(), f.lock, cp)) + + reconciler, err := schemachange.NewReconciler(f.cfg, f.lock, f.store, f.target, shadow, checksum.Options{}) + require.NoError(t, err) + reconciled, err := reconciler.Reconcile(t.Context(), f.pool, cp) + require.NoError(t, err) + t.Cleanup(func() { + ctx := context.WithoutCancel(t.Context()) + assert.NoError(t, reconciled.Slot.Close(ctx)) + assert.NoError(t, decode.DropSlot(ctx, f.cfg, f.pool, reconciled.Slot.Name())) + }) + + assert.Equal(t, 500, f.shadowRowsAbove(t, shadow, 0), "nothing landed, so the shadow is the resumed copier's to clear") +} + +// ClearAbove clears nothing above a complete watermark and the whole +// shadow above a zero one. +func TestClearAboveHonorsCompleteAndZeroWatermarks(t *testing.T) { + f := newReconcileFixture(t, 500) + shadow := f.buildShadow(t) + copyAll(t, f, newCopier(t, f, shadow)) + + require.NoError(t, copier.ClearAbove(t.Context(), f.pool, f.target, shadow, f.lock, copier.NewWatermark(math.MaxInt64), copier.Options{})) + assert.Equal(t, 500, f.shadowRowsAbove(t, shadow, 0), "a complete watermark has nothing above it") + + require.NoError(t, copier.ClearAbove(t.Context(), f.pool, f.target, shadow, f.lock, copier.Watermark{}, copier.Options{})) + assert.Equal(t, 0, f.shadowRowsAbove(t, shadow, 0), "a zero watermark leaves the whole key space above it") +} + +// slotExists reports whether the server has a replication slot named name. +func (f reconcileFixture) slotExists(t *testing.T, name string) bool { + t.Helper() + var exists bool + require.NoError(t, f.pool.QueryRow(t.Context(), ` + SELECT EXISTS ( + SELECT 1 + FROM pg_catalog.pg_replication_slots + WHERE slot_name = $1 + )`, name).Scan(&exists)) + return exists +} + // shadowRowsAbove counts the shadow rows whose key lies above key. func (f reconcileFixture) shadowRowsAbove(t *testing.T, shadow schemachange.BuiltShadow, key int64) int { t.Helper()