From 37050c168e0e6784f2caddbde0aa0af6481ff264 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Fri, 23 Oct 2026 08:14:14 -0700 Subject: [PATCH] applier: wait for a delivered commit until the source shows it visible The walsender sends a commit once its WAL is flushed, before the backend has made it visible to snapshots; under synchronous commit the gap lasts as long as the standby wait. A chunk or completion read taken in that gap does not see the change, so the drain now judges each entry's xid against pg_current_snapshot() and keeps an invisible one buffered as in flight. --- docs/copy-and-swap-design.md | 4 +- docs/invariants.md | 41 +++-- docs/schemabot-integration.md | 2 +- pkg/applier/absent_integration_test.go | 10 +- pkg/applier/buffer.go | 11 +- pkg/applier/buffer_test.go | 32 +++- pkg/applier/catchup.go | 40 ++++- pkg/applier/catchup_integration_test.go | 9 +- .../catchup_sync_commit_integration_test.go | 78 +++++++++ pkg/applier/drain.go | 69 +++++--- pkg/applier/drain_test.go | 165 +++++++++++++++--- pkg/applier/entry.go | 10 ++ pkg/applier/flusher.go | 2 +- pkg/applier/guard_integration_test.go | 8 +- pkg/applier/hold_integration_test.go | 2 +- pkg/applier/hold_test.go | 18 +- pkg/applier/requeue_test.go | 6 +- pkg/applier/snapshot.go | 139 +++++++++++++++ pkg/applier/snapshot_test.go | 72 ++++++++ pkg/decode/stream.go | 8 +- pkg/decode/stream_test.go | 24 ++- pkg/decode/types.go | 14 +- pkg/schemachange/runner_drain.go | 8 +- 23 files changed, 664 insertions(+), 108 deletions(-) create mode 100644 pkg/applier/catchup_sync_commit_integration_test.go create mode 100644 pkg/applier/snapshot.go create mode 100644 pkg/applier/snapshot_test.go diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index 9b4c6d38..43af3a19 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -613,8 +613,8 @@ next pass's to catch. | `pkg/preflight` | Produces `CopySwapShape` (the table facts `PreflightedTable` carries plus the v1 shape, replica identity, and dependent-object checks above, minted by `CheckCopySwapShape`) and from it `CopySwapTarget`, the copy-and-swap route's proof, minted only by `CheckCopySwapEnvironment` once the decoding and headroom facts are proven against that shape; `CheckCopySwap` runs the privilege, shape, and environment checks as one call; owns Tier-3 refusals. | ST-6, RF-1..RF-3 | | `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 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 the shadow itself rejects is the server's `*pgconn.PgError` and carries no 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), `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. `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, `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. A relation other than the target, 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 and `Drain` judges every key against the copier's `Position`, returning 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), `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/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, 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), `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, 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 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, drops whatever bears the derived slot name, creates a fresh slot, 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), 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/invariants.md b/docs/invariants.md index 4585f215..29022ebc 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -237,20 +237,27 @@ after the buffer's last `Add`, so a key the copier cuts in between is in flight chunk, a deferred batch stays buffered until the next pass, and the confirmed position never passes a change still buffered, held, or deferred (convergence tests under mixed load, key moves, hot-row contention, unchanged-TOAST updates, and unique-key swaps; a refused batch -requeued while its chunk is in flight and flushed once the chunk lands). *Open:* a source -transaction whose commit the walsender has already sent while its backend is still in a -synchronous-commit wait is not yet visible to a snapshot taken after the stream delivered it, so -an uncut key's change can be discarded ahead of a chunk read that does not see it; the drain -has no rule for that window yet. The same window reaches a moved image whose unchanged-TOAST -markers the flush completes from the source: an image `CompleteFirst` names whose old key's -shadow row is absent or was reused, so its completion is read from the source row under `Key`. -Inside the window that read does not see the delivered move. When it finds no row, the flush -writes a delete at `Key` (`Result.Skipped`) with no stream gate, and the shadow ends up without -a row the source holds; when the same invisible commit also removed another row from `Key`, the -read finds that row and completes the image with its stale marker values. Nothing completes the -image again. The checksum (CO-1) finds the differing row when the loss lands before the pass -settles its chunk, and under `abort` refuses the cutover while `repair` recopies the chunk; a -loss between the pass and the swap has no later comparison. Also open: a primary-key move or a +requeued while its chunk is in flight and flushed once the chunk lands). The server writes a +transaction's commit to the WAL — and waits out any synchronous standby — before it marks the +transaction visible to new snapshots, and the walsender decodes the commit as soon as it is +flushed, so the stream can deliver a change whose transaction no snapshot shows yet; a chunk +read or a completion read from the source taken in that window does not see it, and a drain +that discarded an uncut key's change, or flushed a moved image completed from a source row that +does not yet show the move, would lose it. The drain judges that window as it judges an +in-flight chunk: every `ChangeEvent` carries its transaction's xid, each buffered `Entry` +carries the xid of the latest event merged into it (a key move stamps both halves, a delete its +marker, the flush's replacement marker the image it replaced), and `Drain` takes the source's +`pg_current_snapshot()` — read on the pool before the `Position`, and only when the buffer holds +anything — beside the position: an entry whose xid the snapshot does not show as committed is +treated as in flight, kept buffered and counted in `Batch.Invisible` (its paired entries +deferred with it under the three-way rule), and judged again next cycle, when visibility, which +is monotone, has caught up; the zero `Snapshot` shows nothing visible, so a drain with no +snapshot keeps everything. The swap's drain (`flushAll`) drains against `AllVisible`, since a +writer holds its locks until its commit is visible and the swap's `ACCESS EXCLUSIVE` lock waited +them all out (sync-commit test: a writer under `synchronous_commit = on` against a standby that +never answers has its UPDATE delivered and left buffered with `Invisible` one and nothing +applied, the confirmed position below the delivered one, until the backend is cancelled, after +which the change lands and the catch-up converges). Also open: a primary-key move or a unique-value swap that straddles a chunk the copier is reading collides, on the copier's own insert of the moved row, with the stale shadow row the flush has not yet deleted (SQLSTATE `23505`); the catch-up's half of the exchange is correct and the resolution is the copier's, so @@ -263,8 +270,10 @@ with the DELETE not yet delivered, is held until the stream passes the flush's c catch-up converges, while the same image for a row the source still holds and the shadow lost fails closed once the stream has passed it (CO-8); a key-moving UPDATE that straddles the watermark (`UPDATE t SET id = 5000 WHERE id = 5`, watermark -1000) asserts key 5 is deleted from the shadow. *Source:* this doc set (Spirit's model -translated). +1000) asserts key 5 is deleted from the shadow; an UPDATE committed under `synchronous_commit = +on` against a standby that never acknowledges asserts the delivered change stays buffered, with +nothing applied and the slot unconfirmed past it, until the commit is visible, and lands then. +*Source:* this doc set (Spirit's model translated). ### CO-5 — The change buffer is disjoint and current at flush time diff --git a/docs/schemabot-integration.md b/docs/schemabot-integration.md index 0d958fd8..4f00c73a 100644 --- a/docs/schemabot-integration.md +++ b/docs/schemabot-integration.md @@ -465,7 +465,7 @@ each asks for a different action. Whether the pass wrote the shadow is part of t | --- | --- | --- | --- | | `Outcome` with `VerifiedShadow` minted (`Live()` true) | Every chunk was equal or settled against the catch-up's confirmed position; the shadow equals the source as of those positions and every change since is in the stream | Hand the proof to `GateCutover`; `Cutover` requires a `DrainFunc` that applies the stream through the final position under the swap lock, and refuses the proof without one (`cutover-undrained`) | No | | `*UnresolvedError`, no repairs | A chunk could not be settled: the catch-up did not confirm past the read within the wait (`lagging`), or the chunk changed on every reread (`hot`). Not a finding about the data | Run the pass again — after the catch-up has confirmed further, or when the source is quieter; `Outcome.Report.Unsettled` counts the chunks that did not settle | No | -| `*DivergenceError` (policy `abort`) | At least one chunk held still on both sides across a confirmed position and still differed; the stream delivered everything and the shadow does not match. `Report.Unsettled` may also be non-zero | Stop and report; a divergence under the stream is a defect, or the known synchronous-commit window recorded in [CO-4's *Open* note](invariants.md#co-4--the-copyapply-ordering-invariants), and either way not a retry: a second pass reads the same shadow and stops the same way, so start the run over (the `*checksum.DivergenceError` row above). A chunk that settled is not compared again, so a loss that lands in that window after the pass has settled its chunk is not caught | No | +| `*DivergenceError` (policy `abort`) | At least one chunk held still on both sides across a confirmed position and still differed; the stream delivered everything and the shadow does not match. `Report.Unsettled` may also be non-zero | Stop and report; a divergence under the stream is a defect in the copy/apply ordering [CO-4](invariants.md#co-4--the-copyapply-ordering-invariants) enforces, not a retry: a second pass reads the same shadow and stops the same way, so start the run over (the `*checksum.DivergenceError` row above) | No | | `*UnresolvedError` with `Outcome.Repairs` listed (policy `repair`) | Every confirmed divergence was recopied and read back clean, but another chunk did not settle | The repairs are committed; run the pass again as above. No proof was minted (CO-2) | Yes — the listed repairs | | `*RepairError` (policy `repair`) | A repaired chunk was a confirmed divergence again after its recopy | Stop and report; the recopy committed and is listed in `Outcome.Repairs`, and something other than the stream is writing the shadow | Yes — the listed repairs | diff --git a/pkg/applier/absent_integration_test.go b/pkg/applier/absent_integration_test.go index bb5edd6a..dffc62e7 100644 --- a/pkg/applier/absent_integration_test.go +++ b/pkg/applier/absent_integration_test.go @@ -109,7 +109,7 @@ func TestHeldAbsentImageWaitsForTheStreamToPassTheCheck(t *testing.T) { assert.GreaterOrEqual(t, result.Held[0].ReadLSN, deleteDone, "the check reads a position at or past the DELETE's commit") assert.Equal(t, 0, buffer.Release(deleteDone-1), "the stream has not delivered the DELETE") assert.Equal(t, 1, buffer.Held()) - next := buffer.Drain(everyKeyLanded()) + next := buffer.Drain(everyKeyLanded(), applier.AllVisible()) assert.Empty(t, next.Entries, "the held image is not flushed before the stream passes the check") assert.Equal(t, 1, next.Held) } @@ -159,7 +159,7 @@ func TestFlushRefusesAnAbsentRowOnceTheStreamHasPassedTheCheck(t *testing.T) { require.Len(t, result.Held, 1) assert.Equal(t, 1, buffer.Release(passed(result.Held)), "the stream passed the check and nothing replaced the image") - _, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded())) + _, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.ErrorIs(t, err, applier.ErrInvariantViolation) assert.ErrorContains(t, err, "(CO-8): flush key 2") @@ -187,7 +187,7 @@ func TestFlushRefusesAHeldImageAnUpdateFilledInWhenTheRowIsStillAbsent(t *testin require.Len(t, result.Held, 1) require.NoError(t, buffer.Add(updateEvent(2, 2, col("slot", "Q"), col("doc", "a new document"), col("note", "seat 2")))) assert.Equal(t, 1, buffer.Release(passed(result.Held)), "an UPDATE keeps the held image pending") - _, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded())) + _, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.ErrorIs(t, err, applier.ErrInvariantViolation) assert.ErrorContains(t, err, "(CO-8): flush key 2") @@ -212,12 +212,12 @@ func TestFlushInRefusesALostRowOnTheSecondPassOfTheSwapDrain(t *testing.T) { require.NoError(t, err) t.Cleanup(func() { _ = tx.Rollback(context.WithoutCancel(t.Context())) }) - first, err := p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded())) + first, err := p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.NoError(t, err) require.Len(t, first.Held, 1) require.NoError(t, buffer.Hold(first.Held)) assert.Equal(t, 1, buffer.Release(math.MaxUint64)) - _, err = p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded())) + _, err = p.flush.FlushIn(t.Context(), tx, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.ErrorIs(t, err, applier.ErrInvariantViolation) assert.ErrorContains(t, err, "(CO-8): flush key 2") diff --git a/pkg/applier/buffer.go b/pkg/applier/buffer.go index a0d15f7a..b625fe02 100644 --- a/pkg/applier/buffer.go +++ b/pkg/applier/buffer.go @@ -103,7 +103,7 @@ func (b *Buffer) addInsert(ev decode.ChangeEvent) error { } } b.markOldKeyReused(ev.Key, nil) - b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: firstLSN(ev.Delivered, cur)}) + b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: firstLSN(ev.Delivered, cur), Xid: ev.Xid}) return nil } @@ -117,13 +117,14 @@ func (b *Buffer) addUpdate(ev decode.ChangeEvent) error { cur, ok := b.entries[ev.Key] if !ok { b.markOldKeyReused(ev.Key, nil) - b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: ev.Delivered}) + b.put(&Entry{Key: ev.Key, Kind: Image, Columns: cloneColumns(ev.Columns), FirstLSN: ev.Delivered, Xid: ev.Xid}) return nil } if cur.Kind == DeleteMarker { return fmt.Errorf("%w (CO-5): buffer key %d: update over a buffered delete", ErrInvariantViolation, ev.Key) } cur.overlay(ev.Columns) + cur.Xid = ev.Xid return nil } @@ -152,7 +153,7 @@ func (b *Buffer) addKeyMove(ev decode.ChangeEvent) error { return fmt.Errorf("%w (CO-5): buffer key %d: key move from %d, which is buffered deleted", ErrInvariantViolation, ev.Key, oldKey) } b.markOldKeyReused(ev.Key, from) - moved := &Entry{Key: ev.Key, Kind: Image, OldKey: &oldKey, FirstLSN: firstLSN(ev.Delivered, cur, from)} + moved := &Entry{Key: ev.Key, Kind: Image, OldKey: &oldKey, FirstLSN: firstLSN(ev.Delivered, cur, from), Xid: ev.Xid} if from != nil { moved.Columns = cloneColumns(from.Columns) moved.overlay(ev.Columns) @@ -165,14 +166,14 @@ func (b *Buffer) addKeyMove(ev decode.ChangeEvent) error { moved.Columns = cloneColumns(ev.Columns) } b.put(moved) - b.put(&Entry{Key: oldKey, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, from)}) + b.put(&Entry{Key: oldKey, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, from), Xid: ev.Xid}) return nil } // addDelete replaces whatever the key held with a delete marker: the newest // fact about the key is that its row is gone (CO-5). func (b *Buffer) addDelete(ev decode.ChangeEvent) { - b.put(&Entry{Key: ev.Key, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, b.entries[ev.Key])}) + b.put(&Entry{Key: ev.Key, Kind: DeleteMarker, FirstLSN: firstLSN(ev.Delivered, b.entries[ev.Key]), Xid: ev.Xid}) } // markOldKeyReused records that key now holds a row other than the one each diff --git a/pkg/applier/buffer_test.go b/pkg/applier/buffer_test.go index 02dbc4c3..717b9d14 100644 --- a/pkg/applier/buffer_test.go +++ b/pkg/applier/buffer_test.go @@ -167,7 +167,7 @@ func TestBufferReusedOldKeyIsNotACompletionSource(t *testing.T) { t.Run("chain, after the old key's marker was discarded", func(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 50, 15, col("label", "r"), marker("doc")))) - _ = b.Drain(position(t, 30, [2]int64{11, 20})) + _ = b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) require.NoError(t, b.Add(insert(20, 50, col("label", "s"), col("doc", "S-doc")))) require.NoError(t, b.Add(keyMove(30, 15, 25, col("label", "r"), marker("doc")))) got := entry(t, b, 25) @@ -178,7 +178,7 @@ func TestBufferReusedOldKeyIsNotACompletionSource(t *testing.T) { t.Run("reuse seen as an update of a key the buffer does not hold", func(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 50, 25, col("label", "r"), marker("doc")))) - _ = b.Drain(position(t, 30, [2]int64{21, 30})) + _ = b.Drain(position(t, 30, [2]int64{21, 30}), AllVisible()) require.NoError(t, b.Add(update(20, 50, col("label", "s2")))) assert.True(t, entry(t, b, 25).OldKeyReused, "a live row at 50 is not R") }) @@ -212,7 +212,7 @@ func TestBufferReuseIndexFollowsTheEntries(t *testing.T) { t.Run("image drained, then the old key reused", func(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 50, 25, col("label", "r"), marker("doc")))) - batch := b.Drain(position(t, 6000)) + batch := b.Drain(position(t, 6000), AllVisible()) require.Equal(t, []int64{25, 50}, keys(batch.Entries)) require.NoError(t, b.Add(insert(20, 50, col("label", "s"), col("doc", "S-doc")))) assert.Equal(t, 1, b.Len()) @@ -389,6 +389,32 @@ func TestBufferKeysEveryEventKindOnItsDeliveredPosition(t *testing.T) { } } +// An entry carries the transaction of the latest event merged into it: an +// overlaying UPDATE replaces it, a key move stamps both halves with the +// move's, and a DELETE stamps the marker with its own. +func TestBufferEntryCarriesItsLatestTransaction(t *testing.T) { + t.Run("overlay", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 1, col("label", "a")), 600))) + assert.Equal(t, uint32(600), entry(t, b, 1).Xid) + require.NoError(t, b.Add(inTx(update(20, 1, col("label", "b")), 700))) + assert.Equal(t, uint32(700), entry(t, b, 1).Xid) + }) + t.Run("key move", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 1, col("label", "a")), 600))) + require.NoError(t, b.Add(inTx(keyMove(20, 1, 2, col("label", "b")), 700))) + assert.Equal(t, uint32(700), entry(t, b, 1).Xid, "the marker at the old key") + assert.Equal(t, uint32(700), entry(t, b, 2).Xid, "the moved image") + }) + t.Run("delete", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 1, col("label", "a")), 600))) + require.NoError(t, b.Add(inTx(del(20, 1), 700))) + assert.Equal(t, uint32(700), entry(t, b, 1).Xid) + }) +} + func TestEntryKindStrings(t *testing.T) { assert.Equal(t, "image", Image.String()) assert.Equal(t, "delete-marker", DeleteMarker.String()) diff --git a/pkg/applier/catchup.go b/pkg/applier/catchup.go index 61133e0b..3f0f71f2 100644 --- a/pkg/applier/catchup.go +++ b/pkg/applier/catchup.go @@ -126,6 +126,12 @@ type Status struct { // check that found an unmoved image's shadow row absent. A nonzero value // is normal for the cycle or two a hold lasts. Held int + // Invisible is the number of entries the last drain left waiting for + // the source to show their transaction visible: a commit the stream + // delivered that the server was still flushing or waiting on a + // synchronous standby for. A nonzero value is normal for a cycle; + // one that persists means the source's commits are not completing. + Invisible int // Applied is the number of images and delete markers committed flushes // have written to the shadow. Applied uint64 @@ -224,9 +230,11 @@ func (c *Catchup) Confirmed() decode.LSN { // Run consumes the stream in cycles until ctx ends or the stream, the buffer, // or a flush fails. Each cycle gathers changes until MaxChanges are buffered -// or Interval has passed, reads the copier's Position after the last Add — so -// a discarded uncut key's change is one the copier's read will see (CO-4) — -// drains, flushes the batch, puts a refused batch back (ErrBatchDeferred) or +// or Interval has passed, takes the source's snapshot and then reads the +// copier's Position after the last Add — so a discarded uncut key's change +// is one the copier's read will see, and a change whose commit the source +// has not yet made visible waits for a later cycle (CO-4) — drains, flushes +// the batch, puts a refused batch back (ErrBatchDeferred) or // holds the images the flush did not write — those it completed and those // whose shadow row it found absent — and then confirms the lowest of the // stream's delivered position and the buffer's oldest pending position: @@ -343,9 +351,15 @@ func (c *Catchup) apply(ctx context.Context, pool *pgxpool.Pool) (cycleOutcome, return cycleOutcome{paused: true}, nil } defer c.releaseFlushes() - // INV: CO-4 — the position is read after the last Add of this cycle. + // INV: CO-4 — the snapshot is taken, and then the position read, after + // the last Add of this cycle: a transaction the snapshot shows visible + // is visible to every chunk read the copier makes after the position. + snap, err := c.snapshot(ctx, pool) + if err != nil { + return cycleOutcome{}, err + } pos := c.positions.Position() - batch := c.buffer.Drain(pos) + batch := c.buffer.Drain(pos, snap) result, err := c.flusher.Flush(ctx, pool, batch) outcome := cycleOutcome{batch: batch, result: result} switch { @@ -369,6 +383,20 @@ func (c *Catchup) apply(ctx context.Context, pool *pgxpool.Pool) (cycleOutcome, return outcome, nil } +// snapshot takes the source's snapshot on pool for this cycle's drain, and +// skips the round trip when the buffer holds nothing to judge: the zero +// Snapshot then judges nothing. +func (c *Catchup) snapshot(ctx context.Context, pool *pgxpool.Pool) (Snapshot, error) { + if c.buffer.Len() == 0 { + return Snapshot{}, nil + } + snap, err := ReadSnapshot(ctx, pool) + if err != nil { + return Snapshot{}, fmt.Errorf("catch-up on %s.%s: %w", c.flusher.target.Schema(), c.flusher.target.Table(), err) + } + return snap, nil +} + // checkSlot reads the stream's slot from the catalog after the cycle has // confirmed and judges it against the ceiling: what the slot retains after // this cycle's confirmation is what the next cycle will have to live with. @@ -495,10 +523,12 @@ func (c *Catchup) record(outcome cycleOutcome, walEnd decode.LSN) { case outcome.deferred: c.status.Discarded += uint64(outcome.batch.Discarded) c.status.Deferred = outcome.batch.Deferred + len(outcome.batch.Entries) + c.status.Invisible = outcome.batch.Invisible c.status.BatchesDeferred++ default: c.status.Discarded += uint64(outcome.batch.Discarded) c.status.Deferred = outcome.batch.Deferred + c.status.Invisible = outcome.batch.Invisible c.status.Applied += uint64(outcome.result.Images + outcome.result.Deletes) if outcome.result.Fallback { c.status.Fallbacks++ diff --git a/pkg/applier/catchup_integration_test.go b/pkg/applier/catchup_integration_test.go index fa3fdbcc..2114f9f8 100644 --- a/pkg/applier/catchup_integration_test.go +++ b/pkg/applier/catchup_integration_test.go @@ -49,7 +49,14 @@ const walSenderTimeout = "wal_sender_timeout=2s" func newCatchupFixture(t *testing.T, rows int) catchupFixture { t.Helper() - serverURL := testutil.StartPostgresWithSettings(t, "wal_level=logical", walSenderTimeout) + return newCatchupFixtureWithSettings(t, rows) +} + +// newCatchupFixtureWithSettings starts the fixture's server with extra +// settings beyond the ones every catch-up test needs. +func newCatchupFixtureWithSettings(t *testing.T, rows int, settings ...string) catchupFixture { + t.Helper() + serverURL := testutil.StartPostgresWithSettings(t, append([]string{"wal_level=logical", walSenderTimeout}, settings...)...) cfg := dbconn.Config{URL: testutil.NewDatabase(t, serverURL), MaxConns: 16} pool, err := dbconn.NewPool(t.Context(), cfg) require.NoError(t, err) diff --git a/pkg/applier/catchup_sync_commit_integration_test.go b/pkg/applier/catchup_sync_commit_integration_test.go new file mode 100644 index 00000000..3d522095 --- /dev/null +++ b/pkg/applier/catchup_sync_commit_integration_test.go @@ -0,0 +1,78 @@ +package applier_test + +import ( + "context" + "sync" + "testing" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/applier" +) + +// A commit the server has flushed to the WAL but not yet made visible — one +// waiting for a synchronous standby — is delivered by the stream all the +// same. The catch-up buffers the change and does not apply it while the +// source's snapshot still hides its transaction, since a read of the source +// taken then would miss it (CO-4); once the wait ends and the transaction is +// visible, the change is applied and the shadow converges. +// +// The server names a standby that never connects, so a session that asks +// for synchronous commit waits in its commit until the test cancels the +// wait; every other session commits locally and never waits. +func TestCatchupWaitsForACommitTheServerHasNotMadeVisible(t *testing.T) { + f := newCatchupFixtureWithSettings(t, 20, + "synchronous_commit=local", + "synchronous_standby_names=blocked") + run := startCatchupWith(t, f, catchupSetup{positions: &switchPosition{pos: allLanded()}}) + + // The writer is its own connection, outside the pool's statement + // timeout: its commit is meant to wait. + writer, err := pgx.Connect(t.Context(), f.cfg.URL) + require.NoError(t, err) + t.Cleanup(func() { + assert.NoError(t, writer.Close(context.WithoutCancel(t.Context()))) + }) + var writerPID int + require.NoError(t, writer.QueryRow(t.Context(), `SELECT pg_backend_pid()`).Scan(&writerPID)) + _, err = writer.Exec(t.Context(), `SET synchronous_commit = on`) + require.NoError(t, err) + + var wg sync.WaitGroup + var commitErr error + wg.Go(func() { + _, commitErr = writer.Exec(t.Context(), `UPDATE `+f.table.Qualified()+` SET amount = 99 WHERE id = 2`) + }) + t.Cleanup(func() { + // A writer still waiting when the test fails is released so the + // connection can close. + _, err := f.pool.Exec(context.WithoutCancel(t.Context()), `SELECT pg_cancel_backend($1)`, writerPID) + assert.NoError(t, err) + wg.Wait() + }) + + run.awaitStatus(t, "buffer the delivered change without applying it", func(s applier.Status) bool { + return s.Invisible == 1 && s.Buffered == 1 + }) + waiting := run.status(t) + assert.Zero(t, waiting.Applied, "nothing reaches the shadow while the source hides the transaction") + assert.Less(t, waiting.Confirmed, waiting.Delivered, "the slot is not confirmed past the waiting change") + + // Cancelling a backend that is waiting for a synchronous standby ends + // the wait with a warning; the transaction has committed locally and + // becomes visible. + var cancelled bool + require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT pg_cancel_backend($1)`, writerPID).Scan(&cancelled)) + require.True(t, cancelled) + wg.Wait() + require.NoError(t, commitErr, "the cancelled wait is a warning, not a failed commit") + + end := f.currentWALLSN(t) + run.awaitStatus(t, "apply the change once its transaction is visible", func(s applier.Status) bool { + return s.Applied >= 1 && s.Buffered == 0 && s.Confirmed >= end + }) + run.stop(t) + f.assertConverged(t, run.shadow) +} diff --git a/pkg/applier/drain.go b/pkg/applier/drain.go index 95c74ada..b14c03ee 100644 --- a/pkg/applier/drain.go +++ b/pkg/applier/drain.go @@ -22,14 +22,21 @@ type Batch struct { // the copier's own read of those chunks will see the change. Discarded int // Deferred counts the entries still buffered because their keys are - // inside an in-flight chunk, or travel with one that is. This is the - // number a stalled catch-up is waiting on the copier for. + // inside an in-flight chunk, or travel with one that is or with one + // that is invisible. This is the number a stalled catch-up is waiting + // on the copier for. Deferred int // Held counts the entries still buffered because a flush held them (see // Buffer.Hold) — with a completion, or because their shadow row was // absent: they wait for the stream to pass the position of the flush's // read, not for the copier. Held int + // Invisible counts the entries still buffered because the snapshot the + // drain judged them against did not show their transaction visible yet + // (see Snapshot). They wait for a later drain's snapshot, not for the + // copier; an entry deferred only because it travels with one of them + // counts as Deferred. + Invisible int } // OldestFirstLSN returns the earliest FirstLSN among the batch's entries, and @@ -78,32 +85,43 @@ func (b Batch) CompleteFirst() []Entry { return first } -// Drain decides every buffered entry against one snapshot of the copier's -// position. Each key is judged on its own (CO-4): an uncut key's entry is -// discarded, because the copier's own read of that chunk will see the change; -// a landed key's entry is flushed; an in-flight key's entry stays buffered -// until a later Drain finds its chunk landed. Discarding is sound only when -// the copier reads an uncut key's chunk after the buffered change is visible, -// so pos must be read after the last Add: the caller adds every event it -// holds, then reads the copier's Position, then drains. Discarded and flushed -// entries leave the buffer; a flush that then fails is not re-buffered — the -// stream replays from its last confirmed position, which stays below -// OldestPending and the batch's OldestFirstLSN until the flush commits, and -// rebuilds them. +// Drain decides every buffered entry against one reading of the copier's +// position and one snapshot of the source. Each key is judged on its own +// (CO-4): an uncut key's entry is discarded, because the copier's own read +// of that chunk will see the change; a landed key's entry is flushed; an +// in-flight key's entry stays buffered until a later Drain finds its chunk +// landed. Discarding is sound only when the copier reads an uncut key's +// chunk after the buffered change is visible, so the caller adds every +// event it holds, then takes snap, then reads the copier's Position, then +// drains: a transaction snap shows visible is visible to every read the +// copier or a flush makes after it. An entry whose transaction snap does +// not show visible yet stays buffered whatever its key's state — discarded, +// the copier's read could miss it; flushed, a completion read could — and +// is judged again by the next Drain. Discarded and flushed entries leave the +// buffer; a flush that then fails is not re-buffered — the stream replays +// from its last confirmed position, which stays below OldestPending and the +// batch's OldestFirstLSN until the flush commits, and rebuilds them. // // An image that moved from an old key, and the entry still buffered at that -// old key, travel together once neither is uncut: if either is in flight both -// wait, otherwise both flush. The flush completes the image's unchanged-TOAST -// markers before it deletes anything (D13), and that reading must not race the -// deletion or copy of the old key's row; Batch.CompleteFirst names the images -// it applies to. An image whose completion is pending (Buffer.Hold) stays -// buffered whatever its key's state, until Release lets the completion in -// or an event drops it. -func (b *Buffer) Drain(pos copier.Position) Batch { +// old key, travel together once neither is uncut: if either is in flight or +// invisible both wait, otherwise both flush. The flush completes the image's +// unchanged-TOAST markers before it deletes anything (D13), and that reading +// must not race the deletion or copy of the old key's row; Batch.CompleteFirst +// names the images it applies to. An image whose completion is pending +// (Buffer.Hold) stays buffered whatever its key's state, until Release lets +// the completion in or an event drops it. +func (b *Buffer) Drain(pos copier.Position, snap Snapshot) Batch { // INV: CO-4 state := make(map[int64]copier.KeyState, len(b.entries)) - for key := range b.entries { + invisible := make(map[int64]struct{}) + for key, e := range b.entries { state[key] = pos.Classify(key) + if !snap.Visible(e.Xid) { + // An invisible entry waits as an in-flight one does, and so + // does anything that travels with it. + state[key] = copier.KeyInFlight + invisible[key] = struct{}{} + } } b.deferPairs(state) var batch Batch @@ -114,6 +132,11 @@ func (b *Buffer) Drain(pos copier.Position) Batch { b.trackOldest(e.FirstLSN) continue } + if _, waits := invisible[key]; waits { + batch.Invisible++ + b.trackOldest(e.FirstLSN) + continue + } switch state[key] { case copier.KeyUncut: b.remove(key) diff --git a/pkg/applier/drain_test.go b/pkg/applier/drain_test.go index f1bebba2..b26b5335 100644 --- a/pkg/applier/drain_test.go +++ b/pkg/applier/drain_test.go @@ -41,7 +41,7 @@ func TestDrainJudgesEachKeyAgainstThePosition(t *testing.T) { require.NoError(t, b.Add(update(30, 5, col("label", "landed-low")))) require.NoError(t, b.Add(del(40, 40))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Equal(t, []int64{5, 30}, keys(batch.Entries)) assert.Equal(t, 1, batch.Discarded, "key 40 is above the cut frontier") @@ -54,7 +54,7 @@ func TestDrainJudgesEachKeyAgainstThePosition(t *testing.T) { require.True(t, ok) assert.Equal(t, decode.LSN(10), oldest, "nor the batch's first event until its flush commits") - batch = b.Drain(position(t, 30)) + batch = b.Drain(position(t, 30), AllVisible()) assert.Equal(t, []int64{15}, keys(batch.Entries)) assert.Equal(t, 0, batch.Discarded) assert.Equal(t, 0, batch.Deferred) @@ -69,7 +69,7 @@ func TestDrainDiscardsEverythingBeforeTheFirstClaim(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(update(10, 1, col("label", "a")))) require.NoError(t, b.Add(del(20, 2))) - batch := b.Drain(copier.Position{}) + batch := b.Drain(copier.Position{}, AllVisible()) assert.Empty(t, batch.Entries) assert.Equal(t, 2, batch.Discarded) assert.Equal(t, 0, b.Len()) @@ -82,10 +82,10 @@ func TestDrainDiscardsEverythingBeforeTheFirstClaim(t *testing.T) { func TestDrainDeferredEntryKeepsMerging(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(update(10, 15, col("label", "first"), col("doc", "d")))) - _ = b.Drain(position(t, 30, [2]int64{11, 20})) + _ = b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) require.NoError(t, b.Add(update(20, 15, col("label", "second"), marker("doc")))) - flush := b.Drain(position(t, 30)).Entries + flush := b.Drain(position(t, 30), AllVisible()).Entries require.Len(t, flush, 1) assert.Equal(t, []decode.Column{col("label", "second"), col("doc", "d")}, flush[0].Columns) } @@ -95,7 +95,7 @@ func TestDrainDeferredEntryKeepsMerging(t *testing.T) { func TestDrainReturnsCopies(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(update(10, 1, col("label", "a")))) - flush := b.Drain(position(t, 30)).Entries + flush := b.Drain(position(t, 30), AllVisible()).Entries require.NoError(t, b.Add(update(20, 1, col("label", "b")))) assert.Equal(t, []decode.Column{col("label", "a")}, flush[0].Columns) } @@ -106,7 +106,7 @@ func TestDrainKeyMoveStraddlingTheFrontier(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 5, 5000, col("label", "a")))) - batch := b.Drain(position(t, 1000)) + batch := b.Drain(position(t, 1000), AllVisible()) require.Len(t, batch.Entries, 1) assert.Equal(t, int64(5), batch.Entries[0].Key) @@ -130,13 +130,13 @@ func TestDrainKeyMovePairWaitsTogether(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, tc.from, tc.to, col("label", "a"), marker("doc")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Empty(t, batch.Entries) assert.Equal(t, 0, batch.Discarded) assert.Equal(t, 2, batch.Deferred) assert.Equal(t, 2, b.Len()) - batch = b.Drain(position(t, 30)) + batch = b.Drain(position(t, 30), AllVisible()) assert.Equal(t, sortedPair(tc.from, tc.to), keys(batch.Entries), "both halves flush in one batch once the chunk lands") }) } @@ -150,7 +150,7 @@ func TestDrainKeyMoveUncutHalfIsJudgedAlone(t *testing.T) { t.Run("old key uncut, image landed", func(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 5000, 25, col("label", "a"), marker("doc")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Equal(t, []int64{25}, keys(batch.Entries)) assert.Equal(t, 1, batch.Discarded) assert.Equal(t, 0, b.Len()) @@ -158,7 +158,7 @@ func TestDrainKeyMoveUncutHalfIsJudgedAlone(t *testing.T) { t.Run("old key in flight, image uncut", func(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 15, 5000, col("label", "a")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Empty(t, batch.Entries) assert.Equal(t, 1, batch.Discarded) assert.Equal(t, 1, batch.Deferred, "the marker at 15 waits for its own chunk") @@ -167,7 +167,7 @@ func TestDrainKeyMoveUncutHalfIsJudgedAlone(t *testing.T) { t.Run("old key uncut, image in flight", func(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 5000, 15, col("label", "a"), marker("doc")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Empty(t, batch.Entries) assert.Equal(t, 1, batch.Discarded, "the marker at 5000 is uncut; nothing of that row will be copied") assert.Equal(t, 1, batch.Deferred, "the image at 15 waits for its own chunk") @@ -184,7 +184,7 @@ func TestDrainDeferralSpreadsThroughSharedOldKey(t *testing.T) { require.NoError(t, b.Add(insert(20, 25, col("label", "reinserted")))) require.NoError(t, b.Add(keyMove(30, 25, 28, col("label", "second")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Empty(t, batch.Entries, "15 is in flight, so 25 waits, so 28 waits") assert.Equal(t, 0, batch.Discarded) assert.Equal(t, 3, batch.Deferred) @@ -205,13 +205,13 @@ func TestDrainDeferralSpreadsAlongAChainOfPairs(t *testing.T) { require.NoError(t, b.Add(keyMove(30, 45, 35, col("label", "c")))) require.NoError(t, b.Add(keyMove(40, 55, 45, col("label", "d")))) - batch := b.Drain(position(t, 60, [2]int64{51, 60})) + batch := b.Drain(position(t, 60, [2]int64{51, 60}), AllVisible()) assert.Empty(t, batch.Entries, "55 is in flight, so 45 waits, so 35, 25 and 15 wait") assert.Equal(t, 0, batch.Discarded) assert.Equal(t, 5, batch.Deferred) assert.Equal(t, 5, b.Len()) - batch = b.Drain(position(t, 60)) + batch = b.Drain(position(t, 60), AllVisible()) assert.Equal(t, []int64{15, 25, 35, 45, 55}, keys(batch.Entries), "the whole chain flushes together once the chunk lands") } @@ -222,7 +222,7 @@ func TestDrainKeyMovePairsWithReinsertedOldKey(t *testing.T) { require.NoError(t, b.Add(keyMove(10, 15, 25, col("label", "moved")))) require.NoError(t, b.Add(insert(20, 15, col("label", "reinserted")))) - batch := b.Drain(position(t, 30, [2]int64{11, 20})) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) assert.Empty(t, batch.Entries, "the landed image at 25 waits for the in-flight re-inserted row at 15") assert.Equal(t, 2, b.Len()) } @@ -237,7 +237,7 @@ func TestDrainBatchNamesMovedImagesToCompleteFirst(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 5, 5000, col("label", "a"), marker("doc")))) - batch := b.Drain(position(t, 6000)) + batch := b.Drain(position(t, 6000), AllVisible()) require.Equal(t, []int64{5, 5000}, keys(batch.Entries), "the delete at 5 sorts before the image that must read 5's row") first := batch.CompleteFirst() @@ -251,7 +251,7 @@ func TestDrainBatchNamesMovedImagesToCompleteFirst(t *testing.T) { require.NoError(t, b.Add(keyMove(10, 5, 5000, col("label", "a"), marker("doc")))) require.NoError(t, b.Add(update(20, 5000, col("doc", "filled")))) - batch := b.Drain(position(t, 6000)) + batch := b.Drain(position(t, 6000), AllVisible()) require.Equal(t, []int64{5, 5000}, keys(batch.Entries)) assert.Empty(t, batch.CompleteFirst(), "the image is whole; nothing in the batch has to be read first") @@ -261,7 +261,7 @@ func TestDrainBatchNamesMovedImagesToCompleteFirst(t *testing.T) { require.NoError(t, b.Add(update(10, 7, col("label", "a"), marker("doc")))) require.NoError(t, b.Add(del(20, 5))) - batch := b.Drain(position(t, 6000)) + batch := b.Drain(position(t, 6000), AllVisible()) require.Equal(t, []int64{5, 7}, keys(batch.Entries)) assert.True(t, batch.Entries[1].HasMarker()) @@ -272,7 +272,7 @@ func TestDrainBatchNamesMovedImagesToCompleteFirst(t *testing.T) { require.NoError(t, b.Add(keyMove(10, 50, 25, col("label", "r"), marker("doc")))) require.NoError(t, b.Add(insert(20, 50, col("label", "s"), col("doc", "S-doc")))) - batch := b.Drain(position(t, 6000)) + batch := b.Drain(position(t, 6000), AllVisible()) require.Equal(t, []int64{25, 50}, keys(batch.Entries)) first := batch.CompleteFirst() @@ -287,7 +287,7 @@ func TestDrainBatchNamesMovedImagesToCompleteFirst(t *testing.T) { func TestDrainCompleteFirstCopiesTheEntries(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 5, 5000, col("label", "a"), marker("doc")))) - batch := b.Drain(position(t, 6000)) + batch := b.Drain(position(t, 6000), AllVisible()) first := batch.CompleteFirst() require.Len(t, first, 1) @@ -297,6 +297,129 @@ func TestDrainCompleteFirstCopiesTheEntries(t *testing.T) { assert.True(t, batch.Entries[1].HasMarker(), "the batch's entry still carries the marker") } +// inTx stamps an event with the transaction that wrote it. +func inTx(ev decode.ChangeEvent, xid uint32) decode.ChangeEvent { + ev.Xid = xid + return ev +} + +// completedBelow is a source snapshot in which every transaction below xid +// has completed and nothing at or above it has. +func completedBelow(xid uint32) Snapshot { + return Snapshot{read: true, xmin: uint64(xid), xmax: uint64(xid), inProgress: map[uint64]struct{}{}} +} + +// An entry whose transaction the snapshot does not show visible stays +// buffered whatever its key's state (CO-4): an uncut key is not discarded, +// because the copier's read of that chunk could still miss the change, and +// a landed key is not flushed, because a completion read could. Both still +// hold the stream's confirm, and the next drain whose snapshot shows the +// transaction judges them as usual. +func TestDrainKeepsAnEntryUntilItsTransactionIsVisible(t *testing.T) { + t.Run("uncut key is not discarded", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 5000, col("label", "late")), 700))) + + batch := b.Drain(position(t, 30), completedBelow(700)) + assert.Empty(t, batch.Entries) + assert.Equal(t, 0, batch.Discarded, "discarding now would let the copier's chunk read miss the change") + assert.Equal(t, 1, batch.Invisible) + assert.Equal(t, 0, batch.Deferred) + assert.Equal(t, 1, b.Len()) + oldest, ok := b.OldestPending() + require.True(t, ok) + assert.Equal(t, decode.LSN(10), oldest, "the kept entry still bounds the confirm") + + batch = b.Drain(position(t, 30), completedBelow(701)) + assert.Equal(t, 1, batch.Discarded, "visible now, and uncut: the copier will read it") + assert.Equal(t, 0, batch.Invisible) + assert.Equal(t, 0, b.Len()) + }) + t.Run("landed key is not flushed", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 25, col("label", "late")), 700))) + + batch := b.Drain(position(t, 30), completedBelow(700)) + assert.Empty(t, batch.Entries) + assert.Equal(t, 1, batch.Invisible) + assert.Equal(t, 1, b.Len()) + + batch = b.Drain(position(t, 30), completedBelow(701)) + assert.Equal(t, []int64{25}, keys(batch.Entries)) + assert.Equal(t, 0, batch.Invisible) + assert.Equal(t, 0, b.Len()) + }) +} + +// The entry is judged on the transaction of its latest event: an earlier, +// visible event overlaid by a later, invisible one waits with the later +// one, since the flush would write the merged image. +func TestDrainJudgesAnEntryOnItsLatestTransaction(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 25, col("label", "early")), 600))) + require.NoError(t, b.Add(inTx(update(20, 25, col("label", "late")), 700))) + + batch := b.Drain(position(t, 30), completedBelow(700)) + assert.Empty(t, batch.Entries) + assert.Equal(t, 1, batch.Invisible) + assert.Equal(t, 1, b.Len()) + oldest, ok := b.OldestPending() + require.True(t, ok) + assert.Equal(t, decode.LSN(10), oldest, "the entry still owes the stream its first event") +} + +// An invisible half holds its travelling partner: a landed move whose image +// is not visible yet keeps the old key's marker buffered too, counted as +// deferred rather than invisible, and both flush once the transaction is +// visible. +func TestDrainKeyMovePairWaitsForVisibilityTogether(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(keyMove(10, 15, 25, col("label", "moved")), 700))) + + batch := b.Drain(position(t, 30), completedBelow(700)) + assert.Empty(t, batch.Entries) + assert.Equal(t, 2, batch.Invisible, "both halves carry the move's transaction") + assert.Equal(t, 0, batch.Deferred) + assert.Equal(t, 2, b.Len()) + + batch = b.Drain(position(t, 30), completedBelow(701)) + assert.Equal(t, []int64{15, 25}, keys(batch.Entries), "both halves flush in one batch once the move is visible") + assert.Equal(t, 0, b.Len()) +} + +// A partner that is visible on its own still waits for the invisible half: +// a visible moved image stays buffered when the entry at its old key was +// last written by a transaction not visible yet, and is counted as deferred. +func TestDrainVisiblePartnerWaitsForInvisibleHalf(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(keyMove(10, 15, 25, col("label", "moved")), 600))) + require.NoError(t, b.Add(inTx(insert(20, 15, col("label", "reinserted")), 650))) + require.NoError(t, b.Add(inTx(del(30, 15), 700))) + + batch := b.Drain(position(t, 30), completedBelow(700)) + assert.Empty(t, batch.Entries) + assert.Equal(t, 1, batch.Invisible, "the marker at 15 was last written at 700") + assert.Equal(t, 1, batch.Deferred, "the image at 25 is visible but travels with the marker") + assert.Equal(t, 2, b.Len()) + + batch = b.Drain(position(t, 30), completedBelow(701)) + assert.Equal(t, []int64{15, 25}, keys(batch.Entries)) +} + +// The zero Snapshot shows nothing visible: a drain against one keeps every +// entry, whatever the position says. +func TestDrainAgainstTheZeroSnapshotKeepsEverything(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(update(10, 25, col("label", "landed")), 700))) + require.NoError(t, b.Add(inTx(update(20, 5000, col("label", "uncut")), 700))) + + batch := b.Drain(position(t, 30), Snapshot{}) + assert.Empty(t, batch.Entries) + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 2, batch.Invisible) + assert.Equal(t, 2, b.Len()) +} + func sortedPair(a, b int64) []int64 { if a < b { return []int64{a, b} diff --git a/pkg/applier/entry.go b/pkg/applier/entry.go index 30db6251..02f7a2f5 100644 --- a/pkg/applier/entry.go +++ b/pkg/applier/entry.go @@ -69,6 +69,16 @@ type Entry struct { // would be lost on replay. It is not the event's own LSN, which can lie // below a position already confirmed. FirstLSN decode.LSN + // Xid is the transaction of the latest event merged into this entry + // (decode.ChangeEvent.Xid). Drain judges the entry only once a snapshot + // of the source shows that transaction visible: a commit the stream + // has delivered can still be invisible to readers while the server + // waits on WAL flush or synchronous replication, and a copier read or + // flush read taken in that window would not see it. The latest + // transaction suffices because transactions that write the same key + // are serialized by the row's lock, which is released only after the + // earlier one became visible. + Xid uint32 } // HasMarker reports whether the image still carries a column no buffered diff --git a/pkg/applier/flusher.go b/pkg/applier/flusher.go index 69540026..28cfe41f 100644 --- a/pkg/applier/flusher.go +++ b/pkg/applier/flusher.go @@ -295,7 +295,7 @@ func (f *Flusher) completeMoved(ctx context.Context, tx pgx.Tx, batch Batch, res continue } result.Skipped = append(result.Skipped, e.Key) - replace[e.Key] = &Entry{Key: e.Key, Kind: DeleteMarker, FirstLSN: e.FirstLSN} + replace[e.Key] = &Entry{Key: e.Key, Kind: DeleteMarker, FirstLSN: e.FirstLSN, Xid: e.Xid} } entries := make([]Entry, 0, len(batch.Entries)) for _, e := range batch.Entries { diff --git a/pkg/applier/guard_integration_test.go b/pkg/applier/guard_integration_test.go index e3a03edf..68859ccb 100644 --- a/pkg/applier/guard_integration_test.go +++ b/pkg/applier/guard_integration_test.go @@ -115,14 +115,14 @@ func TestFlushIgnoresTheSessionSearchPath(t *testing.T) { require.NoError(t, buffer.Add(deleteEvent(2, 3))) require.NoError(t, buffer.Add(moveEvent(3, 2, 10, col("slot", "B"), marker("doc"), col("note", "seat 2")))) - result, err := p.flush.Flush(t.Context(), shadowing, buffer.Drain(everyKeyLanded())) + result, err := p.flush.Flush(t.Context(), shadowing, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.NoError(t, err) require.NoError(t, buffer.Hold(result.Held)) assert.Equal(t, applier.Result{Images: 1, Deletes: 2, Held: result.Held}, result) require.Len(t, result.Held, 1) assert.False(t, result.Held[0].FromSource, "the completion read found the old key's row through the catalog's =") require.Equal(t, 1, buffer.Release(passed(result.Held))) - result, err = p.flush.Flush(t.Context(), shadowing, buffer.Drain(everyKeyLanded())) + result, err = p.flush.Flush(t.Context(), shadowing, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.NoError(t, err) assert.Equal(t, applier.Result{Images: 1}, result) @@ -161,7 +161,7 @@ func TestFlushCompletesFloatsExactlyWhateverTheConnectionRenders(t *testing.T) { buffer := applier.NewBuffer() require.NoError(t, buffer.Add(moveEvent(1, 1, 10, col("value", "0.30000000000000004"), marker("samples"), col("note", "n")))) - result, err := p.flush.Flush(t.Context(), pool, buffer.Drain(everyKeyLanded())) + result, err := p.flush.Flush(t.Context(), pool, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.NoError(t, err) require.NoError(t, buffer.Hold(result.Held)) require.Len(t, result.Held, 1) @@ -171,7 +171,7 @@ func TestFlushCompletesFloatsExactlyWhateverTheConnectionRenders(t *testing.T) { assert.Contains(t, samples, "0.30000000000000004", "the completion carries every digit") assert.NotContains(t, samples, "0.3,", "and never the rounded form") require.Equal(t, 1, buffer.Release(passed(result.Held))) - _, err = p.flush.Flush(t.Context(), pool, buffer.Drain(everyKeyLanded())) + _, err = p.flush.Flush(t.Context(), pool, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.NoError(t, err) f.assertConverged(t, p.shadow) diff --git a/pkg/applier/hold_integration_test.go b/pkg/applier/hold_integration_test.go index 702ed172..7b15f61f 100644 --- a/pkg/applier/hold_integration_test.go +++ b/pkg/applier/hold_integration_test.go @@ -46,7 +46,7 @@ func deleteEvent(lsn decode.LSN, key int64) decode.ChangeEvent { // owner does after every flush. func (f flushFixture) flushAll(t *testing.T, p prepared, buffer *applier.Buffer) applier.Result { t.Helper() - result, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded())) + result, err := p.flush.Flush(t.Context(), f.pool, buffer.Drain(everyKeyLanded(), applier.AllVisible())) require.NoError(t, err) require.NoError(t, buffer.Hold(result.Held)) return result diff --git a/pkg/applier/hold_test.go b/pkg/applier/hold_test.go index e35f6748..99b823a4 100644 --- a/pkg/applier/hold_test.go +++ b/pkg/applier/hold_test.go @@ -13,7 +13,7 @@ import ( // back as a flush would: completed with value, read at readLSN. func held(t *testing.T, b *Buffer, readLSN decode.LSN, value string) HeldImage { t.Helper() - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) require.Len(t, batch.CompleteFirst(), 1, "exactly one image to complete") return HeldImage{ Entry: batch.CompleteFirst()[0], @@ -40,7 +40,7 @@ func TestHoldKeepsTheImageUntilTheStreamPassesTheRead(t *testing.T) { oldest, ok := b.OldestPending() require.True(t, ok) assert.Equal(t, decode.LSN(10), oldest, "the confirmed position may not pass the held image") - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) assert.Empty(t, batch.Entries, "a held image is not flushed") assert.Equal(t, 1, batch.Held) assert.Equal(t, 0, batch.Deferred) @@ -50,7 +50,7 @@ func TestHoldKeepsTheImageUntilTheStreamPassesTheRead(t *testing.T) { assert.True(t, entry(t, b, 8).HasMarker()) assert.Equal(t, 1, b.Release(500)) assert.Equal(t, 0, b.Held()) - batch = b.Drain(position(t, 100)) + batch = b.Drain(position(t, 100), AllVisible()) require.Equal(t, []int64{8}, keys(batch.Entries)) assert.Equal(t, 0, batch.Held) assert.Equal(t, []decode.Column{ @@ -67,7 +67,7 @@ func TestReleaseTakesOnlyTheCompletionsTheStreamHasPassed(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 1, 2, marker("blob")))) require.NoError(t, b.Add(keyMove(20, 3, 4, marker("blob")))) - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) first := batch.CompleteFirst() require.Len(t, first, 2) require.NoError(t, b.Hold([]HeldImage{ @@ -80,7 +80,7 @@ func TestReleaseTakesOnlyTheCompletionsTheStreamHasPassed(t *testing.T) { assert.Equal(t, 1, b.Held()) assert.False(t, entry(t, b, 2).HasMarker()) assert.True(t, entry(t, b, 4).HasMarker()) - batch = b.Drain(position(t, 100)) + batch = b.Drain(position(t, 100), AllVisible()) assert.Equal(t, []int64{2}, keys(batch.Entries)) assert.Equal(t, 1, batch.Held) } @@ -109,7 +109,7 @@ func TestHoldRefusesAKeyTheBufferHolds(t *testing.T) { func TestReleaseFillsOnlyTheMarkersAnUpdateLeft(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 7, 8, col("label", "a"), marker("blob"), marker("doc")))) - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) require.NoError(t, b.Hold([]HeldImage{{ Entry: batch.CompleteFirst()[0], Completed: []decode.Column{col("blob", "read blob"), col("doc", "read doc")}, @@ -154,7 +154,7 @@ func TestAnEventThatReplacesTheHeldKeyDropsTheCompletion(t *testing.T) { assert.True(t, moved.HasMarker(), "the dropped completion is not carried on") require.NotNil(t, moved.OldKey) assert.Equal(t, int64(7), *moved.OldKey, "the row's shadow row is still under the key it started at") - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) assert.Equal(t, []int64{8, 9}, keys(batch.Entries)) assert.Equal(t, []int64{9}, keys(batch.CompleteFirst())) }) @@ -176,7 +176,7 @@ func TestReuseOfTheOldKeyDropsAShadowCompletion(t *testing.T) { moved := entry(t, b, 8) assert.True(t, moved.OldKeyReused) assert.True(t, moved.HasMarker()) - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) assert.Equal(t, []int64{7, 8}, keys(batch.Entries)) assert.Equal(t, []int64{8}, keys(batch.CompleteFirst())) } @@ -214,7 +214,7 @@ func TestHoldKeepsACompletionInTheFormOfTheRowItWasReadFrom(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(keyMove(10, 1, 2, marker("blob")))) require.NoError(t, b.Add(keyMove(20, 3, 4, marker("blob")))) - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) first := batch.CompleteFirst() require.Len(t, first, 2) require.NoError(t, b.Hold([]HeldImage{ diff --git a/pkg/applier/requeue_test.go b/pkg/applier/requeue_test.go index 22702de9..49e0c70e 100644 --- a/pkg/applier/requeue_test.go +++ b/pkg/applier/requeue_test.go @@ -18,7 +18,7 @@ func TestRequeuePutsTheBatchBack(t *testing.T) { require.NoError(t, b.Add(update(10, 5, col("label", "a")))) require.NoError(t, b.Add(keyMove(20, 7, 8, marker("blob")))) require.NoError(t, b.Add(del(30, 9))) - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) require.Equal(t, []int64{5, 7, 8, 9}, keys(batch.Entries)) require.Equal(t, 0, b.Len()) @@ -31,7 +31,7 @@ func TestRequeuePutsTheBatchBack(t *testing.T) { require.NoError(t, b.Add(insert(40, 7, col("label", "other"), col("blob", "x")))) assert.True(t, entry(t, b, 8).OldKeyReused, "the requeued image is indexed under its old key again") - again := b.Drain(position(t, 100)) + again := b.Drain(position(t, 100), AllVisible()) assert.Equal(t, []int64{5, 7, 8, 9}, keys(again.Entries)) assert.Equal(t, []decode.Column{decode.PresentColumn("label", "a")}, again.Entries[0].Columns) } @@ -43,7 +43,7 @@ func TestRequeueRefusesAHeldKey(t *testing.T) { b := NewBuffer() require.NoError(t, b.Add(update(10, 5, col("label", "a")))) require.NoError(t, b.Add(update(11, 6, col("label", "b")))) - batch := b.Drain(position(t, 100)) + batch := b.Drain(position(t, 100), AllVisible()) require.NoError(t, b.Add(update(12, 6, col("label", "c")))) err := b.Requeue(batch) diff --git a/pkg/applier/snapshot.go b/pkg/applier/snapshot.go new file mode 100644 index 00000000..f16912c4 --- /dev/null +++ b/pkg/applier/snapshot.go @@ -0,0 +1,139 @@ +package applier + +import ( + "context" + "fmt" + "strconv" + "strings" + + "github.com/block/pg-sprite/pkg/dbconn" +) + +// Snapshot is what the source showed as committed at one instant: which +// transactions a reader taking a snapshot then would see. Drain judges a +// buffered entry only once its transaction is visible in one (CO-4). +// +// The server writes a transaction's commit to the WAL, and waits for any +// synchronous standby to acknowledge it, before it marks the transaction +// visible to new snapshots. The walsender decodes the commit from the WAL +// as soon as it is flushed, so the stream can deliver a change whose +// transaction no reader can see yet. A copier chunk read or a flush's +// completion read taken in that window misses the change; a drain that +// discarded or flushed the entry then would lose it (CO-4). Visibility is +// monotone, so a transaction visible in this snapshot is visible to every +// read taken after it. +// +// The zero Snapshot shows nothing visible, so a drain against one keeps +// every entry: the safe reading when no snapshot was taken. +type Snapshot struct { + // read is set once the snapshot carries what the server reported. + read bool + // all is set by AllVisible: every transaction counts as visible. + all bool + // xmin is the lowest transaction still in progress; every transaction + // below it has completed. xmax is one above the highest completed + // transaction; every transaction at or above it was still in progress + // or had not begun. Both are full 64-bit transaction ids, the epoch in + // the high half. + xmin, xmax uint64 + // inProgress holds the transactions between xmin and xmax that were + // still in progress. + inProgress map[uint64]struct{} +} + +// AllVisible returns a snapshot in which every transaction is visible. It +// is for a drain under a lock that excludes every writer of the source: +// each writer's transaction finished, and became visible, before the lock +// was granted. +func AllVisible() Snapshot { + return Snapshot{read: true, all: true} +} + +// ReadSnapshot takes the server's current snapshot on db. The snapshot is +// read as text and parsed here rather than decoded as the server's xid8 +// type: the text form is stable across the supported majors. +func ReadSnapshot(ctx context.Context, db dbconn.RowQuerier) (Snapshot, error) { + var text string + if err := db.QueryRow(ctx, `SELECT pg_catalog.pg_current_snapshot()::text`).Scan(&text); err != nil { + return Snapshot{}, fmt.Errorf("read the source's snapshot: %w", err) + } + snap, err := parseSnapshot(text) + if err != nil { + return Snapshot{}, fmt.Errorf("read the source's snapshot: %w", err) + } + return snap, nil +} + +// parseSnapshot parses the server's text form of a snapshot: +// xmin:xmax:xip1,xip2,… with the in-progress list possibly empty. +func parseSnapshot(text string) (Snapshot, error) { + parts := strings.Split(text, ":") + if len(parts) != 3 { + return Snapshot{}, fmt.Errorf("%w (CO-4): snapshot %q is not xmin:xmax:xip", ErrInvariantViolation, text) + } + xmin, err := strconv.ParseUint(parts[0], 10, 64) + if err != nil { + return Snapshot{}, fmt.Errorf("%w (CO-4): snapshot %q: xmin: %w", ErrInvariantViolation, text, err) + } + xmax, err := strconv.ParseUint(parts[1], 10, 64) + if err != nil { + return Snapshot{}, fmt.Errorf("%w (CO-4): snapshot %q: xmax: %w", ErrInvariantViolation, text, err) + } + if xmin > xmax { + return Snapshot{}, fmt.Errorf("%w (CO-4): snapshot %q: xmin above xmax", ErrInvariantViolation, text) + } + snap := Snapshot{read: true, xmin: xmin, xmax: xmax, inProgress: make(map[uint64]struct{})} + if parts[2] == "" { + return snap, nil + } + for field := range strings.SplitSeq(parts[2], ",") { + xid, err := strconv.ParseUint(field, 10, 64) + if err != nil { + return Snapshot{}, fmt.Errorf("%w (CO-4): snapshot %q: in-progress transaction: %w", ErrInvariantViolation, text, err) + } + snap.inProgress[xid] = struct{}{} + } + return snap, nil +} + +// Visible reports whether a reader taking this snapshot would see the +// transaction's writes. The stream numbers transactions in 32 bits, as the +// WAL does; the transaction is widened to the 64-bit id nearest the +// snapshot's xmax with those low bits, the way the server itself orders +// 32-bit ids. A transaction at or above xmax is one the server had not +// marked complete when the snapshot was taken, however far the stream has +// delivered it. The zero Snapshot reports nothing visible. +func (s Snapshot) Visible(xid uint32) bool { + if s.all { + return true + } + if !s.read { + return false + } + full := s.lift(xid) + if full < s.xmin { + return true + } + if full >= s.xmax { + return false + } + _, inProgress := s.inProgress[full] + return !inProgress +} + +// lift widens a 32-bit transaction id to the 64-bit id nearest xmax whose +// low 32 bits are xid: the signed 32-bit distance from xmax's low bits +// places it within half an epoch either side of xmax, in xmax's epoch, the +// one before, or the one after. A distance that would reach below the +// first id lifts to zero. +func (s Snapshot) lift(xid uint32) uint64 { + distance := int64(int32(xid - uint32(s.xmax))) + if distance >= 0 { + return s.xmax + uint64(distance) + } + back := uint64(-distance) + if back > s.xmax { + return 0 + } + return s.xmax - back +} diff --git a/pkg/applier/snapshot_test.go b/pkg/applier/snapshot_test.go new file mode 100644 index 00000000..5c0cef2d --- /dev/null +++ b/pkg/applier/snapshot_test.go @@ -0,0 +1,72 @@ +package applier + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// A transaction is visible when it completed before the snapshot: below +// xmin, or between xmin and xmax and not in the in-progress list. The +// in-progress list and xmax are judged exactly, on both sides of each +// boundary. +func TestSnapshotVisibleJudgesEachTransactionAgainstTheHorizon(t *testing.T) { + snap, err := parseSnapshot("100:110:100,103,107") + require.NoError(t, err) + + assert.True(t, snap.Visible(99), "below xmin: completed") + assert.False(t, snap.Visible(100), "xmin itself is the oldest in progress") + assert.True(t, snap.Visible(101), "between the horizons and not in progress") + assert.False(t, snap.Visible(103), "in progress") + assert.True(t, snap.Visible(109), "just below xmax, not in progress") + assert.False(t, snap.Visible(110), "xmax: not yet marked complete") + assert.False(t, snap.Visible(140), "above xmax: delivered by the stream before the server marked it complete") +} + +// A snapshot with nothing in progress still hides transactions at or +// above xmax. +func TestSnapshotWithNothingInProgress(t *testing.T) { + snap, err := parseSnapshot("110:110:") + require.NoError(t, err) + + assert.True(t, snap.Visible(109)) + assert.False(t, snap.Visible(110)) +} + +// The stream's 32-bit transaction id is placed in the epoch nearest the +// snapshot's xmax: one just below xmax's low bits that wrapped is the +// previous epoch's and completed; one just above that wrapped is the next +// epoch's and not yet complete. +func TestSnapshotVisibleLiftsTransactionsAcrossAnEpochBoundary(t *testing.T) { + const epoch = uint64(1) << 32 + + intoNewEpoch, err := parseSnapshot("4294967290:4294967301:") // xmin 0xFFFFFFFA, xmax epoch 1 | 5 + require.NoError(t, err) + assert.True(t, intoNewEpoch.Visible(0xFFFFFFF0), "previous epoch, below xmin") + assert.True(t, intoNewEpoch.Visible(3), "this epoch, below xmax") + assert.False(t, intoNewEpoch.Visible(5), "this epoch, xmax itself") + + beforeNewEpoch := Snapshot{read: true, xmin: epoch - 10, xmax: epoch - 5, inProgress: map[uint64]struct{}{}} + assert.True(t, beforeNewEpoch.Visible(0xFFFFFFF0), "this epoch, below xmin") + assert.False(t, beforeNewEpoch.Visible(7), "next epoch: a commit the server has not marked complete") +} + +// The zero snapshot shows nothing visible, so a drain with no snapshot +// keeps every entry; AllVisible shows everything, for a drain under a lock +// that excludes every writer. +func TestSnapshotZeroHidesEverythingAndAllVisibleShowsEverything(t *testing.T) { + assert.False(t, Snapshot{}.Visible(1)) + assert.False(t, Snapshot{}.Visible(0xFFFFFFFF)) + assert.True(t, AllVisible().Visible(1)) + assert.True(t, AllVisible().Visible(0xFFFFFFFF)) +} + +// A snapshot the server rendered in an unexpected shape is an invariant +// violation, not a guess at what it meant. +func TestParseSnapshotRefusesAMalformedSnapshot(t *testing.T) { + for _, text := range []string{"", "100", "100:110", "a:110:", "100:b:", "100:110:c", "120:110:"} { + _, err := parseSnapshot(text) + assert.ErrorIs(t, err, ErrInvariantViolation, text) + } +} diff --git a/pkg/decode/stream.go b/pkg/decode/stream.go index 741dc3f8..fb28b1eb 100644 --- a/pkg/decode/stream.go +++ b/pkg/decode/stream.go @@ -83,6 +83,9 @@ type Stream struct { // until the first change arrives. relation *relation inTransaction bool + // xid is the transaction being yielded, from its BEGIN; every change + // inside it is stamped with it. + xid uint32 // failed is the error that ended the stream, returned from then on. failed error } @@ -345,6 +348,7 @@ func (s *Stream) handleWALData(xld pglogrepl.XLogData) (Delivery, bool, error) { ErrInvariantViolation, s.slotName, LSN(xld.WALStart)) } s.inTransaction = true + s.xid = msg.Xid return Delivery{}, false, nil case *pglogrepl.CommitMessage: if !s.inTransaction { @@ -377,12 +381,14 @@ func (s *Stream) handleWALData(xld pglogrepl.XLogData) (Delivery, bool, error) { } // deliverChange stamps a decoded change with the delivered position it -// arrives with and wraps it as a delivery; a decoding error yields nothing. +// arrives with and the transaction it belongs to, and wraps it as a +// delivery; a decoding error yields nothing. func (s *Stream) deliverChange(ev *ChangeEvent, err error) (Delivery, bool, error) { if err != nil { return Delivery{}, false, err } ev.Delivered = s.delivered + ev.Xid = s.xid return Delivery{Change: ev, Delivered: s.delivered}, true, nil } diff --git a/pkg/decode/stream_test.go b/pkg/decode/stream_test.go index 80cfb432..f3339c47 100644 --- a/pkg/decode/stream_test.go +++ b/pkg/decode/stream_test.go @@ -16,12 +16,18 @@ func walData(walStart LSN, message []byte) pglogrepl.XLogData { return pglogrepl.XLogData{WALStart: pglogrepl.LSN(walStart), ServerWALEnd: pglogrepl.LSN(walStart), WALData: message} } -// beginMessage encodes a pgoutput BEGIN: final LSN, commit time, xid. +// beginMessage encodes a pgoutput BEGIN of transaction 1: final LSN, commit +// time, xid. func beginMessage(finalLSN LSN) []byte { + return beginMessageOf(finalLSN, 1) +} + +// beginMessageOf encodes a pgoutput BEGIN of the given transaction. +func beginMessageOf(finalLSN LSN, xid uint32) []byte { b := []byte{byte(pglogrepl.MessageTypeBegin)} b = binary.BigEndian.AppendUint64(b, uint64(finalLSN)) b = binary.BigEndian.AppendUint64(b, 0) - return binary.BigEndian.AppendUint32(b, 1) + return binary.BigEndian.AppendUint32(b, xid) } // commitMessage encodes a pgoutput COMMIT: flags, commit LSN, end LSN, @@ -128,6 +134,20 @@ func TestStreamStampsAChangeWithTheDeliveredPositionItArrivedWith(t *testing.T) assert.EqualValues(t, 101, d.Change.Key) } +// A change is stamped with the transaction its BEGIN announced, so a +// consumer can ask a later snapshot whether that transaction is visible yet. +func TestStreamStampsAChangeWithItsTransaction(t *testing.T) { + s := &Stream{delivered: 180, relation: keyOnlyRelation(7)} + _, _, err := s.handleWALData(walData(160, beginMessageOf(190, 4242))) + require.NoError(t, err) + + d, yielded, err := s.handleWALData(walData(170, insertMessage(7, "101"))) + + require.NoError(t, err) + require.True(t, yielded) + assert.EqualValues(t, 4242, d.Change.Xid) +} + // A row change the server sent outside BEGIN … COMMIT is refused rather // than yielded: no commit would ever deliver a position covering it. func TestStreamRefusesAChangeOutsideATransaction(t *testing.T) { diff --git a/pkg/decode/types.go b/pkg/decode/types.go index f49d29c5..9943f752 100644 --- a/pkg/decode/types.go +++ b/pkg/decode/types.go @@ -71,9 +71,17 @@ type ChangeEvent struct { // the change is unapplied: a stream reopened from it replays the // change's transaction. Delivered LSN - Key int64 - OldKey *int64 - Columns []Column + // Xid is the transaction the change belongs to, as the server numbers + // transactions. The server sends a transaction once its commit is in + // the WAL, which is before the committing backend has made it visible + // to snapshots — a backend waiting for synchronous replication sits + // between the two for as long as the wait lasts — so a reader that + // takes a snapshot after the change arrived checks Xid against that + // snapshot before it trusts a read to see the change. + Xid uint32 + Key int64 + OldKey *int64 + Columns []Column } // PresentColumns returns only values present in the decoded row image. diff --git a/pkg/schemachange/runner_drain.go b/pkg/schemachange/runner_drain.go index 3db1462b..dcf3ad69 100644 --- a/pkg/schemachange/runner_drain.go +++ b/pkg/schemachange/runner_drain.go @@ -176,7 +176,11 @@ func gather(ctx context.Context, stream *decode.Stream, marker decode.LSN, buffe // change it gathered. func flushAll(ctx context.Context, tx pgx.Tx, flusher *applier.Flusher, buffer *applier.Buffer, watermark copier.Watermark) error { landed := copier.Position{Watermark: watermark, Cut: watermark} - result, err := flushBatch(ctx, tx, flusher, buffer.Drain(landed)) + // Every writer of the source finished, and its commit became visible, + // before the swap's lock was granted, so no gathered change can be + // invisible here. + visible := applier.AllVisible() + result, err := flushBatch(ctx, tx, flusher, buffer.Drain(landed, visible)) if err != nil { return err } @@ -185,7 +189,7 @@ func flushAll(ctx context.Context, tx pgx.Tx, flusher *applier.Flusher, buffer * return err } buffer.Release(decode.LSN(math.MaxUint64)) - again, err := flushBatch(ctx, tx, flusher, buffer.Drain(landed)) + again, err := flushBatch(ctx, tx, flusher, buffer.Drain(landed, visible)) if err != nil { return err }