From 4862c2bd04bb61706cb97f203238222c133bd3a8 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Thu, 22 Oct 2026 13:09:07 -0700 Subject: [PATCH 1/3] copier: retry a chunk whose insert collides with a unique index A unique value the source moves between a landed key and a key in flight collides on the chunk's insert until the applier has flushed the move; the copy used to stop on the first 23505. Each attempt re-reads the source, so the chunk lands once the flush commits. The catch-up convergence test for unique-key swaps now runs its load throughout the copy. --- docs/copy-and-swap-design.md | 2 +- docs/invariants.md | 20 +++- pkg/applier/catchup_integration_test.go | 39 ++++++- pkg/copier/copier.go | 60 ++++++++-- pkg/copier/doc.go | 4 +- pkg/copier/retry.go | 90 +++++++++++++++ pkg/copier/retry_integration_test.go | 143 ++++++++++++++++++++++++ pkg/copier/retry_test.go | 55 +++++++++ 8 files changed, 390 insertions(+), 23 deletions(-) create mode 100644 pkg/copier/retry.go create mode 100644 pkg/copier/retry_integration_test.go create mode 100644 pkg/copier/retry_test.go diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index 51f70667..4102ec06 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -611,7 +611,7 @@ next pass's to catch. | --- | --- | --- | | `pkg/dbconn` | Produces `TableLock`, carried by `TableLockSession`; `Confirm` is the in-transaction check every writer runs from its own connection before its first write. | LK-1 | | `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/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 whose insert collides with a shadow row on a unique index or exclusion constraint (SQLSTATE `23505` / `23P01`) — a value the catch-up has moved between rows and the flush has yet to rewrite — is run again after a bounded linear wait (`Options.CollisionAttempts`, `Options.CollisionBackoff`; `Collisions()` counts them) and, when every attempt collides, the server's `*pgconn.PgError` is returned wrapped; any other chunk the shadow itself rejects is that error as it stands, and neither carries a cause. | CO-4, CO-9, LK-1, LK-2, LK-3 | | `pkg/checksum` | `Verifier` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) compares the source with its shadow up to the copier's landed watermark and returns a `Report` of the chunks that differ. It cuts its own chunks with a `copier.Chunker`, cutting and digesting each chunk in one read-only `REPEATABLE READ` transaction, so the cut and the chunk's two digests describe one snapshot, no snapshot outlives one chunk, and the cut's lock wait is bounded like the reads (LK-2); each transaction runs the copier's guard — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, so a repair's recopy resolves a USING expression as the copy did), `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 | diff --git a/docs/invariants.md b/docs/invariants.md index bbea6fed..700a25a2 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -250,12 +250,20 @@ a row the source holds; when the same invisible commit also removed another row 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 -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 -the convergence tests for key moves and unique-key swaps start their load after the copy has -landed. *Test obligation:* a +loss between the pass and the swap has no later comparison. *Enforced* for the copier's side +of a unique value that moves between rows while its chunk is in flight: the chunk's insert +collides (SQLSTATE `23505`, or `23P01` for an exclusion constraint) with the shadow row that +still holds the value — a landed row the flush has yet to rewrite, or a row of the same chunk +the copier itself read before the swap — and the copier runs the chunk again after a bounded +wait (`copier.Options.CollisionAttempts` × `CollisionBackoff`, linear), counting each +collision, until the flush has moved the value and the chunk lands or the attempts run out and +the server's error is returned with no cause of its own (a copier integration test holds a +colliding row across several attempts and removes it; a catch-up convergence test runs +unique-key swaps throughout the copy). *Open:* a primary-key move whose row carries a unique +secondary value, from a landed key to a key in flight: the drain's pair rule defers the old +key's delete until the chunk lands, and the chunk cannot land while the old key's row holds the +value, so the two wait on each other until the copier's attempts run out; the convergence test +for key moves starts its load after the copy has landed. *Test obligation:* a marker-bearing UPDATE for a key inside an in-flight chunk asserts the flush waits for the chunk and the row is then completed from the copied shadow row, never an absent-row abort; a marker-bearing UPDATE for a key whose row the source then deleted before its chunk was read, diff --git a/pkg/applier/catchup_integration_test.go b/pkg/applier/catchup_integration_test.go index fa3fdbcc..8fc166a3 100644 --- a/pkg/applier/catchup_integration_test.go +++ b/pkg/applier/catchup_integration_test.go @@ -341,9 +341,10 @@ func TestCatchupConvergesUnderMixedLoad(t *testing.T) { // travels together through the drain, and the flush completes the image's // unchanged blob from the old key's row before it deletes that row (CO-4, // D13). The workload starts after the copy has landed every key: the moved -// row keeps its unique key, so mid copy the copier's own insert of the row -// under its new key collides with the stale row the flush has not deleted -// yet, which is the copier's race to resolve, not the catch-up's. +// row keeps its unique key, so a move from a landed key to a key in flight +// collides on the copier's insert with the row still under the old key, and +// the pair rule defers that key's delete until the chunk lands — the two +// would wait on each other until the copier's collision attempts ran out. func TestCatchupConvergesUnderKeyMoves(t *testing.T) { f := newCatchupFixture(t, 1500) run := startCatchup(t, f, applier.CatchupOptions{}) @@ -442,6 +443,38 @@ func TestCatchupConvergesUnderUniqueKeyMoves(t *testing.T) { f.assertConverged(t, run.shadow) } +// Unique key swaps run from before the copy claims its first chunk until +// after it has landed every key, so a swap can straddle a chunk the copier +// is reading: one row's new unique key is in a chunk in flight while the +// other row, already landed, has had its half of the swap flushed — or the +// chunk read both rows before the swap and the flush has since given a +// landed row the value the chunk carries. Either way the chunk's insert +// collides with a shadow row the applier still has to rewrite, the copier +// runs the chunk again once the flush has committed, and the shadow +// converges (CO-4). +func TestCatchupConvergesUnderUniqueKeyMovesDuringTheCopy(t *testing.T) { + f := newCatchupFixture(t, 1500) + run := startCatchup(t, f, applier.CatchupOptions{}) + load := testutil.StartLoad(t, f.pool, f.table, testutil.LoadSpec{ + Seed: 6, Workers: 2, RatePerSecond: 40, + Mix: testutil.Mix{UniqueMove: 1}, + }) + + run.awaitStatus(t, "discard changes to keys the copier has not claimed", func(s applier.Status) bool { + return s.Discarded > 0 + }) + run.copy(t) + t.Logf("chunk attempts the copier ran again after a collision: %d", run.copier.Collisions()) + run.awaitStatus(t, "apply unique key swaps to landed keys", func(s applier.Status) bool { + return s.Applied > 0 + }) + summary := run.quiesce(t, load) + assert.Positive(t, summary.UniqueMoves) + + run.stop(t) + f.assertConverged(t, run.shadow) +} + // A second Run on the same catch-up is refused: its stream is single-use. func TestCatchupRunsOnce(t *testing.T) { f := newCatchupFixture(t, 10) diff --git a/pkg/copier/copier.go b/pkg/copier/copier.go index 9fc418a6..20da57f4 100644 --- a/pkg/copier/copier.go +++ b/pkg/copier/copier.go @@ -6,6 +6,7 @@ import ( "fmt" "math" "sync" + "sync/atomic" "time" "github.com/jackc/pgx/v5/pgxpool" @@ -21,8 +22,9 @@ const DefaultWorkers = 4 var ( // ErrInvalidOptions reports copier options that cannot bound the copy: a - // negative worker count, or a timeout below PostgreSQL's one-millisecond - // resolution, which the server would read as no timeout at all. + // negative worker count, a timeout below PostgreSQL's one-millisecond + // resolution, which the server would read as no timeout at all, a + // collision attempt count below one, or a negative collision backoff. ErrInvalidOptions = errors.New("invalid copier options") // ErrAlreadyRun reports a second Run on the same Copier; a Copier drives // one copy and is then only a record of where it stopped. @@ -31,7 +33,8 @@ var ( // Options bounds the copy. Zero values take the defaults: DefaultWorkers // workers, the dbconn session timeouts for every chunk transaction, the -// ChunkerOptions defaults, and the wall clock. +// ChunkerOptions defaults, DefaultCollisionAttempts attempts spaced by +// DefaultCollisionBackoff, and the wall clock. type Options struct { // Workers is the number of chunk transactions in flight at once. The // pool's connection limit bounds it from above. @@ -42,6 +45,15 @@ type Options struct { StatementTimeout time.Duration // Chunker sizes the chunks. Chunker ChunkerOptions + // CollisionAttempts is the number of times a chunk whose insert the + // server refused on a unique index or an exclusion constraint is run + // before the copy stops with that refusal. Each attempt reads the source + // afresh, so a value a source change moved between keys lands once the + // applier has flushed the move's other half into the shadow. + CollisionAttempts int + // CollisionBackoff is the wait before a collided chunk's second attempt; + // each later wait is one backoff longer. + CollisionBackoff time.Duration // Clock times each chunk for the chunker's chunk-time throttling // feedback (docs/copy-and-swap-design.md#d12--throttle-by-chunk-time-and-slot-lag). Clock progress.Clock @@ -62,6 +74,12 @@ func (o Options) withDefaults() Options { if o.StatementTimeout == 0 { o.StatementTimeout = dbconn.DefaultStatementTimeout } + if o.CollisionAttempts == 0 { + o.CollisionAttempts = DefaultCollisionAttempts + } + if o.CollisionBackoff == 0 { + o.CollisionBackoff = DefaultCollisionBackoff + } if o.Clock == nil { o.Clock = progress.WallClock{} } @@ -82,6 +100,12 @@ func (o Options) validate() error { return fmt.Errorf("%w: %s %s is below PostgreSQL's one-millisecond resolution; use zero for the default", ErrInvalidOptions, timeout.name, timeout.value) } } + if o.CollisionAttempts < 1 { + return fmt.Errorf("%w: collision attempts %d is below one; use zero for the default", ErrInvalidOptions, o.CollisionAttempts) + } + if o.CollisionBackoff < 0 { + return fmt.Errorf("%w: collision backoff %s is negative; use zero for the default", ErrInvalidOptions, o.CollisionBackoff) + } return nil } @@ -117,6 +141,9 @@ type Copier struct { // rowsTotal is the source's catalog row count read once at the start of // Run. rowsTotal uint64 + // collisions counts chunk attempts the server refused on a unique index + // or an exclusion constraint; Collisions reads it from any goroutine. + collisions atomic.Int64 } // NewCopier prepares a copy of target into shadow that resumes after from @@ -204,6 +231,14 @@ func (c *Copier) Position() Position { // Copier from Position().Watermark clears the shadow above the watermark // and continues, so a caller can back off and resume behind a busy table. // +// A chunk whose insert the server refuses on a unique index or an +// exclusion constraint (SQLSTATE 23505 or 23P01) is run again after a wait, +// up to Options.CollisionAttempts times: the value it collided on was moved +// between keys by a source change the applier has yet to flush, and a later +// attempt's snapshot sees the shadow after that flush. A collision no +// attempt outlasts returns the server's error wrapped, the chunk in flight, +// and Collisions counts every refused attempt. +// // Cancellation reaches the server as a closed connection, so a statement // that was running keeps running until it finishes or hits the transaction's // statement_timeout, and only then is its transaction rolled back; the @@ -272,12 +307,14 @@ func (c *Copier) finish(cause error) error { return nil } -// work is one worker's loop: claim a chunk, copy it in its own transaction, -// land it, and tell the chunker how long it took. It returns nil when the -// chunker has no chunk left and the first error otherwise; a cancelled -// context ends the loop with the context's error, which Run resolves to -// the cancellation's cause. A chunk whose copy failed is left in flight: it -// did not land, so its keys must not read as landed. +// work is one worker's loop: claim a chunk, copy it in its own transaction +// (again, after a wait, when the insert collided with a shadow row the +// applier has yet to clear), land it, and tell the chunker how long the +// committed attempt took. It returns nil when the chunker has no chunk left +// and the first error otherwise; a cancelled context ends the loop with the +// context's error, which Run resolves to the cancellation's cause. A chunk +// whose copy failed is left in flight: it did not land, so its keys must +// not read as landed. func (c *Copier) work(ctx context.Context, pool *pgxpool.Pool) error { for ctx.Err() == nil { chunk, ok, err := c.claim(ctx, pool) @@ -287,15 +324,14 @@ func (c *Copier) work(ctx context.Context, pool *pgxpool.Pool) error { if !ok { return nil } - started := c.opts.Clock.Now() - inserted, err := c.copyChunk(ctx, pool, chunk) + inserted, took, err := c.copyUntilClear(ctx, pool, chunk) if err != nil { return err } if err := c.land(chunk, inserted); err != nil { return err } - if err := c.chunker.Feedback(chunk, c.opts.Clock.Now().Sub(started)); err != nil { + if err := c.chunker.Feedback(chunk, took); err != nil { return err } } diff --git a/pkg/copier/doc.go b/pkg/copier/doc.go index 778682c6..7b4cfb5e 100644 --- a/pkg/copier/doc.go +++ b/pkg/copier/doc.go @@ -3,7 +3,9 @@ // that tile the whole int64 range, so every key belongs to exactly one chunk // (CO-4) and each copy statement stays bounded (LK-3). A Copier copies those // chunks with several workers — one guarded, never-overwriting insert per -// chunk — first clearing the shadow above the watermark when it resumes, and +// chunk, run again after a bounded wait when it collides with a shadow row +// the catch-up has yet to rewrite — first clearing the shadow above the +// watermark when it resumes, and // reports a Position whose Classify is the applier's rule for a captured key: // uncut, in flight, or landed. A column the Shadow proof names in // Conversions is copied through that USING expression; the copier does not diff --git a/pkg/copier/retry.go b/pkg/copier/retry.go new file mode 100644 index 00000000..bd9a89dd --- /dev/null +++ b/pkg/copier/retry.go @@ -0,0 +1,90 @@ +package copier + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +// SQLSTATE codes the server raises when a chunk's insert lands a row whose +// value another shadow row already holds under a unique index or an +// exclusion constraint. +const ( + sqlstateUniqueViolation = "23505" + sqlstateExclusionViolation = "23P01" +) + +const ( + // DefaultCollisionAttempts is the number of times a chunk whose insert + // collided is run before the copy stops, when Options leaves + // CollisionAttempts zero. + DefaultCollisionAttempts = 10 + // DefaultCollisionBackoff is the wait before a collided chunk's second + // attempt when Options leaves CollisionBackoff zero. Each later wait is + // one backoff longer, so the default attempts together outlast many + // catch-up cycles, and a cycle's flush is what moves the colliding + // value out of the chunk's way. + DefaultCollisionBackoff = 250 * time.Millisecond +) + +// collided reports whether err is the server refusing a chunk's insert +// because a shadow row already holds one of its values under a unique index +// or an exclusion constraint. The primary key cannot be the index: the +// insert's ON CONFLICT (pk) DO NOTHING skips a row already there (CO-4). +func collided(err error) bool { + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) { + return false + } + return pgErr.Code == sqlstateUniqueViolation || pgErr.Code == sqlstateExclusionViolation +} + +// copyUntilClear runs chunk's copy until it commits, returning the rows the +// committed attempt inserted and how long that attempt took, or until the +// copy must stop. A collision is a row the chunk read from the source whose +// unique value a shadow row still holds under another key: a source change +// moved the value between the keys while this chunk was in flight, and the +// applier's flush has yet to rewrite the landed row the value came from, or +// the chunk read both rows before the move and the flush has since given a +// landed row the value the chunk carries (CO-4). Each attempt is its own +// transaction with its own snapshot, so once the flush has committed a later +// attempt reads the moved rows as the source now has them and lands. A +// collision no attempt outlasts ends the copy with the +// server's error, wrapped with the chunk's bounds and the attempts made; +// the chunk stays in flight. Nothing else is retried here: a row the shadow +// rejects on its own terms, a lock wait that timed out, a lost connection +// is the first attempt's error, returned as such. +func (c *Copier) copyUntilClear(ctx context.Context, pool *pgxpool.Pool, chunk Chunk) (int64, time.Duration, error) { + for attempt := 1; ; attempt++ { + started := c.opts.Clock.Now() + inserted, err := c.copyChunk(ctx, pool, chunk) + if err == nil { + return inserted, c.opts.Clock.Now().Sub(started), nil + } + if !collided(err) { + return 0, 0, err + } + c.collisions.Add(1) + if attempt == c.opts.CollisionAttempts { + return 0, 0, fmt.Errorf("chunk [%d, %d] collided with a shadow row on each of %d attempts; the row holding its value was not removed between them: %w", chunk.Lower(), chunk.Upper(), attempt, err) + } + select { + case <-ctx.Done(): + return 0, 0, ctx.Err() + case <-time.After(c.opts.CollisionBackoff * time.Duration(attempt)): + } + } +} + +// Collisions is the number of chunk attempts this copier's run has had the +// server refuse on a unique index or an exclusion constraint, each followed +// by a wait and another attempt or by the copy stopping. It is safe to read +// from any goroutine while Run is in progress; a copy whose source saw no +// value move during it reads zero. +func (c *Copier) Collisions() int64 { + return c.collisions.Load() +} diff --git a/pkg/copier/retry_integration_test.go b/pkg/copier/retry_integration_test.go new file mode 100644 index 00000000..42116d8c --- /dev/null +++ b/pkg/copier/retry_integration_test.go @@ -0,0 +1,143 @@ +package copier_test + +import ( + "errors" + "fmt" + "math" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/dbconn" + "github.com/block/pg-sprite/pkg/preflight" + "github.com/block/pg-sprite/pkg/schemachange" +) + +// prepareSeats creates, proves, locks, and builds the shadow of a seats +// table whose slot column is unique, holding rows rows with slot = id. The +// shadow keeps the unique index on slot, so a shadow row holding a slot the +// source assigns to another key makes the chunk carrying that key collide. +func (f copierFixture) prepareSeats(t *testing.T, rows int64) (preflight.CopySwapTarget, *dbconn.TableLockSession, schemachange.BuiltShadow) { + t.Helper() + f.exec(t, ` + CREATE TABLE %s.seats ( + id bigint PRIMARY KEY, + slot integer NOT NULL UNIQUE, + note text + )`) + f.exec(t, fmt.Sprintf(` + INSERT INTO %%s.seats (id, slot, note) + SELECT n, n, 'seat ' || n FROM generate_series(1, %d) AS n`, rows)) + target := f.prove(t, "seats") + lock := f.lock(t, "seats") + return target, lock, f.build(t, lock, target) +} + +// holdSlotAtFirstChunk returns a Clock that, at the copier's first reading +// — after the resume clear, before the first chunk's transaction — commits a +// shadow row under key holding slot, standing in for an applier write that +// moved the slot there from a row the first chunk is about to read. The +// returned check fails the test if the row could not be placed. +func (f copierFixture) holdSlotAtFirstChunk(t *testing.T, shadow schemachange.BuiltShadow, key, slot int64) (*hookedClock, func(t *testing.T)) { + t.Helper() + var placeErr error + clock := &hookedClock{hook: func() { + if _, err := f.pool.Exec(t.Context(), "INSERT INTO "+pgx.Identifier{f.schema, shadow.ShadowTable()}.Sanitize()+" (id, slot) VALUES ($1, $2)", key, slot); err != nil { + placeErr = fmt.Errorf("hold slot %d under key %d: %w", slot, key, err) + } + }} + return clock, func(t *testing.T) { + t.Helper() + require.NoError(t, placeErr) + } +} + +// A chunk whose insert collides with a shadow row on the unique slot index +// is run again after a wait, and lands once the colliding row is gone: here +// a shadow row under another key holds slot 15, as it would after a source +// change moved the slot from the row the copier read to a landed key the +// applier has yet to flush the move's other half for (CO-4). Each attempt is +// its own snapshot, so the attempt after the row's removal copies the chunk +// as the source has it, every row lands, and the committed attempt alone +// counts as inserted. +func TestCopierRetriesAChunkThatCollidesUntilTheRowIsGone(t *testing.T) { + f := newCopierFixture(t) + const rows = 300 + target, lock, shadow := f.prepareSeats(t, rows) + shadowName := pgx.Identifier{f.schema, shadow.ShadowTable()}.Sanitize() + clock, placed := f.holdSlotAtFirstChunk(t, shadow, 2000, 15) + + c, err := copier.NewCopier(target, shadow, lock, copier.Watermark{}, copier.Options{ + Workers: 1, + Chunker: copier.ChunkerOptions{InitialRows: 100, MaxRows: 100}, + CollisionBackoff: 20 * time.Millisecond, + Clock: clock, + }) + require.NoError(t, err) + results := make(chan error, 1) + var wg sync.WaitGroup + wg.Go(func() { results <- c.Run(t.Context(), f.pool) }) + t.Cleanup(wg.Wait) + + const collisionDeadline = 15 * time.Second + require.Eventually(t, func() bool { return c.Collisions() >= 1 }, collisionDeadline, 10*time.Millisecond, + "the first chunk, carrying key 15, should collide on the slot the shadow row holds") + placed(t) + pos := c.Position() + require.Len(t, pos.InFlight, 1, "the colliding chunk stays in flight between attempts") + assert.Equal(t, int64(math.MinInt64), pos.InFlight[0].Lower()) + assert.False(t, pos.Watermark.Valid(), "nothing has landed while the first chunk collides") + f.exec(t, "DELETE FROM "+shadowName+" WHERE id = 2000") + + const stopDeadline = 30 * time.Second + require.NoError(t, awaitRun(t, results, stopDeadline, "copy after the colliding row's removal")) + f.assertConverged(t, shadow) + assert.Equal(t, int64(rows), c.Position().RowsInserted, "only the committed attempt's rows count as inserted") + assert.Equal(t, copier.NewWatermark(math.MaxInt64), c.Position().Watermark) + assert.GreaterOrEqual(t, c.Collisions(), int64(1)) +} + +// A collision no attempt outlasts stops the copy with the server's own +// refusal: the error is the unique violation wrapped, not a refusal class of +// the copier's and not a transient dbconn would retry, every attempt is +// counted, and the chunk stays in flight with nothing landed, so a caller +// that checkpoints Position records no key as copied. +func TestCopierStopsWhenACollisionOutlastsEveryAttempt(t *testing.T) { + f := newCopierFixture(t) + target, lock, shadow := f.prepareSeats(t, 300) + clock, placed := f.holdSlotAtFirstChunk(t, shadow, 2000, 15) + + const attempts = 3 + c, err := copier.NewCopier(target, shadow, lock, copier.Watermark{}, copier.Options{ + Workers: 1, + Chunker: copier.ChunkerOptions{InitialRows: 100, MaxRows: 100}, + CollisionAttempts: attempts, + CollisionBackoff: 10 * time.Millisecond, + Clock: clock, + }) + require.NoError(t, err) + err = c.Run(t.Context(), f.pool) + placed(t) + + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr) + assert.Equal(t, "23505", pgErr.Code, "the server's unique violation is the copy's error") + assert.Equal(t, copier.RefusalCause(""), copier.RefusalCauseOf(err), "a collision is the server's answer, not a copier refusal") + assert.False(t, errors.Is(err, copier.ErrInvariantViolation)) + assert.False(t, dbconn.Retryable(err), "a collision is not a transient the connection layer retries") + assert.Contains(t, err.Error(), fmt.Sprintf("on each of %d attempts", attempts)) + assert.Equal(t, int64(attempts), c.Collisions()) + + pos := c.Position() + require.Len(t, pos.InFlight, 1, "the exhausted chunk never landed") + assert.Equal(t, int64(math.MinInt64), pos.InFlight[0].Lower()) + assert.False(t, pos.Watermark.Valid(), "nothing landed") + assert.Equal(t, int64(0), pos.RowsInserted) + assert.Equal(t, int64(0), f.count(t, shadow.ShadowTable(), "id <= 300"), "no attempt's rows survive its rollback") +} diff --git a/pkg/copier/retry_test.go b/pkg/copier/retry_test.go new file mode 100644 index 00000000..a082f75a --- /dev/null +++ b/pkg/copier/retry_test.go @@ -0,0 +1,55 @@ +package copier + +import ( + "errors" + "fmt" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Only the server refusing a row on a unique index or an exclusion +// constraint is a collision the copier waits out; every other failure of a +// chunk is the first attempt's error. +func TestCollidedClassifiesByServerCode(t *testing.T) { + cases := map[string]struct { + err error + collided bool + }{ + "unique violation": {&pgconn.PgError{Code: "23505"}, true}, + "exclusion violation": {&pgconn.PgError{Code: "23P01"}, true}, + "unique violation when wrapped": {fmt.Errorf("copy chunk: %w", &pgconn.PgError{Code: "23505"}), true}, + "check violation": {&pgconn.PgError{Code: "23514"}, false}, + "not-null violation": {&pgconn.PgError{Code: "23502"}, false}, + "lock timeout": {&pgconn.PgError{Code: "55P03"}, false}, + "no error": {nil, false}, + "not a server error": {errors.New("connection reset"), false}, + } + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + assert.Equal(t, tc.collided, collided(tc.err)) + }) + } +} + +func TestCopierCollisionOptionsDefault(t *testing.T) { + opts := Options{}.withDefaults() + assert.Equal(t, DefaultCollisionAttempts, opts.CollisionAttempts) + assert.Equal(t, DefaultCollisionBackoff, opts.CollisionBackoff) + require.NoError(t, opts.validate()) + + given := Options{CollisionAttempts: 3, CollisionBackoff: 20 * time.Millisecond}.withDefaults() + assert.Equal(t, 3, given.CollisionAttempts) + assert.Equal(t, 20*time.Millisecond, given.CollisionBackoff) + require.NoError(t, given.validate()) +} + +// An attempt count below one would never run the chunk, and a negative +// backoff is not a wait; both are refused rather than defaulted. +func TestCopierCollisionOptionsRefuseUnboundedValues(t *testing.T) { + assert.ErrorIs(t, Options{CollisionAttempts: -1}.withDefaults().validate(), ErrInvalidOptions) + assert.ErrorIs(t, Options{CollisionBackoff: -time.Millisecond}.withDefaults().validate(), ErrInvalidOptions) +} From 45d0c05f3282d8f28b38088b7c847e2429eb89fa Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Wed, 28 Oct 2026 08:13:44 +1100 Subject: [PATCH 2/3] applier: flush a landed old key ahead of its in-flight moved image A primary-key move whose row keeps a unique value, from a landed key to a key whose chunk is in flight, deadlocked the drain against the copier: the pair rule deferred the old key's delete until the chunk landed, and the chunk's insert collided with that row until the delete flushed. The pair now waits only for the old key; the image completes from the source row. --- docs/invariants.md | 32 +++-- pkg/applier/catchup_integration_test.go | 12 +- pkg/applier/catchup_stall_integration_test.go | 42 +++++++ pkg/applier/drain.go | 57 +++++---- pkg/applier/drain_test.go | 116 +++++++++++------- 5 files changed, 178 insertions(+), 81 deletions(-) diff --git a/docs/invariants.md b/docs/invariants.md index 0abb2da1..e4966ed3 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -218,11 +218,17 @@ not one the copier's fence — or the verifier's repair fence — waits on; that `pkg/applier` `Buffer.Drain` — every buffered key is judged against one `Position` snapshot: landed flushes, uncut is discarded, in-flight stays buffered until a later drain finds its chunk landed, and a key-moving UPDATE's -two entries are judged per key, except that once neither is uncut the image and its old key's -entry wait together and flush together, so the flush never deletes or copies the old key's row -while an image still needs it for completion (three-way rule, discard-all before the first claim, -deferred entry keeps merging, straddling move, pair waits in either direction, uncut half judged -alone, deferral spreading through a shared old key and along a chain of pairs); `Drain` +two entries are judged per key, except that once neither is uncut the image waits for its old +key's entry — an old key in flight, or invisible, holds its image back so the flush never +deletes or copies the old key's row while the chunk that will copy it is still landing — while a +landed old key's entry flushes ahead of an image in flight for its chunk (an invisible image +still holds its old key's entry back): the delete must not wait for the image's chunk, because +when the moved row carries a unique value the chunk cannot land while the old key's shadow row +holds it, and the image, once its chunk lands, completes from the source row instead (three-way +rule, discard-all before the first claim, deferred entry keeps merging, +straddling move, pair waits for the old key, landed old key flushes ahead of its in-flight image, +uncut half judged alone, deferral spreading through a shared old key and along a chain of +pairs); `Drain` judges against a `Position` read after the last `Add`, and returns a `Batch` whose `Deferred` count is what a catch-up is waiting on the copier for and whose `CompleteFirst` names, as copies of its `Entries`, the moved, marker-bearing images the flush must complete — from the old @@ -266,11 +272,17 @@ wait (`copier.Options.CollisionAttempts` × `CollisionBackoff`, linear), countin collision, until the flush has moved the value and the chunk lands or the attempts run out and the server's error is returned with no cause of its own (a copier integration test holds a colliding row across several attempts and removes it; a catch-up convergence test runs -unique-key swaps throughout the copy). *Open:* a primary-key move whose row carries a unique -secondary value, from a landed key to a key in flight: the drain's pair rule defers the old -key's delete until the chunk lands, and the chunk cannot land while the old key's row holds the -value, so the two wait on each other until the copier's attempts run out; the convergence test -for key moves starts its load after the copy has landed. *Test obligation:* a +unique-key swaps throughout the copy). *Enforced* for a primary-key move whose row carries a +unique secondary value, from a landed key to a key in flight: the drain flushes the landed old +key's delete ahead of the image (the pair rule above), the retried chunk's insert of the row +under the new key then lands, and the image, flushed once its chunk has landed, completes from +the source row because the old key's shadow row is gone (a drain unit test asserts the landed +old key flushes alone as a delete marker with the image deferred and flushed by the next drain; +a catch-up test holds the new key's chunk in flight by hand, asserts the old key's shadow row +is deleted while it is, and converges once the chunk lands). The chunk-level retry remains +bounded: a chunk whose key range keeps receiving unique values from landed rows faster than one +move per catch-up cycle collides on every attempt with a different row, and the copy stops with +the server's error once the attempts run out. *Test obligation:* a marker-bearing UPDATE for a key inside an in-flight chunk asserts the flush waits for the chunk and the row is then completed from the copied shadow row, never an absent-row abort; a marker-bearing UPDATE for a key whose row the source then deleted before its chunk was read, diff --git a/pkg/applier/catchup_integration_test.go b/pkg/applier/catchup_integration_test.go index fbba3437..514ffcd1 100644 --- a/pkg/applier/catchup_integration_test.go +++ b/pkg/applier/catchup_integration_test.go @@ -344,14 +344,10 @@ func TestCatchupConvergesUnderMixedLoad(t *testing.T) { } // A row's primary key moves to a fresh identity value, so the change is an -// image under a new key paired with a delete under the old one. The pair -// travels together through the drain, and the flush completes the image's -// unchanged blob from the old key's row before it deletes that row (CO-4, -// D13). The workload starts after the copy has landed every key: the moved -// row keeps its unique key, so a move from a landed key to a key in flight -// collides on the copier's insert with the row still under the old key, and -// the pair rule defers that key's delete until the chunk lands — the two -// would wait on each other until the copier's collision attempts ran out. +// image under a new key paired with a delete under the old one. With the +// copy landed, both halves are landed: the pair flushes together, and the +// flush completes the image's unchanged blob from the old key's row before +// it deletes that row (CO-4, D13). func TestCatchupConvergesUnderKeyMoves(t *testing.T) { f := newCatchupFixture(t, 1500) run := startCatchup(t, f, applier.CatchupOptions{}) diff --git a/pkg/applier/catchup_stall_integration_test.go b/pkg/applier/catchup_stall_integration_test.go index 85b2fd64..d0583941 100644 --- a/pkg/applier/catchup_stall_integration_test.go +++ b/pkg/applier/catchup_stall_integration_test.go @@ -97,6 +97,48 @@ func TestCatchupRequeuesADeferredBatchAndConverges(t *testing.T) { f.assertConverged(t, run.shadow) } +// Row 7 moves to key 1000, keeping its unique value, while the chunk that +// holds key 1000 is in flight. The delete under the landed key 7 flushes +// ahead of the image: the copier's insert of the row under key 1000 would +// collide with the shadow row under key 7 for as long as that row stood, +// so holding the delete back for the image's chunk would wait on a chunk +// that is waiting on it. Once the chunk lands the image flushes alone, +// completes its unchanged blob from the source row since the shadow row +// under key 7 is gone, and the shadow converges (CO-4, D13). +func TestCatchupFlushesALandedOldKeyAheadOfItsInFlightImage(t *testing.T) { + f := newCatchupFixture(t, 20) + chunk, err := copier.NewChunk(21, math.MaxInt64) + require.NoError(t, err) + inFlight := allLanded() + inFlight.InFlight = []copier.Chunk{chunk} + positions := &switchPosition{pos: inFlight} + run := startCatchupWith(t, f, catchupSetup{positions: positions}) + + q := f.table.Qualified() + _, err = f.pool.Exec(t.Context(), `UPDATE `+q+` SET id = 1000 WHERE id = 7`) + require.NoError(t, err) + + run.awaitStatus(t, "delete the row under key 7 while key 1000's chunk is in flight", func(s applier.Status) bool { + return s.Applied == 1 && s.Deferred == 1 + }) + shadowTable := pgx.Identifier{f.table.Schema, run.shadow.ShadowTable()}.Sanitize() + var sevens int + require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT count(*) FROM `+shadowTable+` WHERE id = 7`).Scan(&sevens)) + assert.Zero(t, sevens, "the shadow row under the old key is gone before the image's chunk lands") + deferred := run.status(t) + assert.Equal(t, 1, deferred.Buffered, "the image waits for its chunk") + assert.Zero(t, deferred.BatchesDeferred, "the delete flushed, it was not refused") + + positions.set(allLanded()) + end := f.currentWALLSN(t) + run.awaitStatus(t, "flush the image once its chunk has landed", func(s applier.Status) bool { + return s.Confirmed >= end && s.Buffered == 0 + }) + assert.Equal(t, uint64(2), run.status(t).Applied, "the delete and then the image") + run.stop(t) + f.assertConverged(t, run.shadow) +} + // The schema change narrows a uniquely indexed column, and after the copy // has landed every key a write gives one row a value that is distinct on // the source and collides in the shadow's type. Nothing is in flight, so no diff --git a/pkg/applier/drain.go b/pkg/applier/drain.go index b14c03ee..f088420b 100644 --- a/pkg/applier/drain.go +++ b/pkg/applier/drain.go @@ -22,9 +22,9 @@ 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 or with one - // that is invisible. This is the number a stalled catch-up is waiting - // on the copier for. + // inside an in-flight chunk, or are moved images whose old key's entry + // is in flight or invisible, or travel with an invisible image. 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 @@ -103,13 +103,21 @@ func (b Batch) CompleteFirst() []Entry { // 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 or -// invisible both wait, otherwise both flush. The flush completes the image's +// old key, travel together once neither is uncut: the image waits while the +// old key's entry is in flight or invisible, and the old key's entry waits +// while the image is invisible. 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. +// must not race the copy of the old key's row or a transaction no snapshot +// shows yet; Batch.CompleteFirst names the images it applies to. A landed +// old key's entry does not wait for an image whose own chunk is in flight: +// the row's unique values are still under the old key in the shadow, and +// the chunk's insert of the row under the new key collides with them until +// that entry flushes, so holding it back would wait on a chunk that is +// waiting on it. The image, flushed once its chunk lands, finds the old +// key's shadow row gone and completes from the source row under its own key +// instead, held until the stream passes the read (D13). 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)) @@ -123,7 +131,7 @@ func (b *Buffer) Drain(pos copier.Position, snap Snapshot) Batch { invisible[key] = struct{}{} } } - b.deferPairs(state) + b.deferPairs(state, invisible) var batch Batch b.hasOldest = false for key, e := range b.entries { @@ -163,13 +171,15 @@ type movePair struct { image, old int64 } -// deferPairs marks both halves of every travelling pair in flight when either -// half is. Two images may share one old key, so a deferral can spread through -// that key to another pair; the passes repeat until one changes nothing. -// Every pass that continues moves at least one key to in flight, so the work -// is bounded by the number of pairs squared, not by the buffer: the pairs are -// collected once, and a buffer with no moves costs one scan. -func (b *Buffer) deferPairs(state map[int64]copier.KeyState) { +// deferPairs marks both halves of every travelling pair in flight when the +// pair must wait together (see waitsTogether). Two images may share one old +// key, so a deferral can spread through that key to another pair; the passes +// repeat until one changes nothing. Every pass that continues moves at least +// one key to in flight, so the work is bounded by the number of pairs +// squared, not by the buffer: the pairs are collected once, and a buffer +// with no moves costs one scan. invisible names the keys whose entries are +// in flight for their transaction's visibility rather than for a chunk. +func (b *Buffer) deferPairs(state map[int64]copier.KeyState, invisible map[int64]struct{}) { var pairs []movePair for key, e := range b.entries { if e.OldKey == nil { @@ -182,7 +192,8 @@ func (b *Buffer) deferPairs(state map[int64]copier.KeyState) { for changed := true; changed; { changed = false for _, p := range pairs { - if !waitsTogether(state[p.image], state[p.old]) { + _, imageInvisible := invisible[p.image] + if !waitsTogether(state[p.image], state[p.old], imageInvisible) { continue } if state[p.image] != copier.KeyInFlight || state[p.old] != copier.KeyInFlight { @@ -195,10 +206,14 @@ func (b *Buffer) deferPairs(state map[int64]copier.KeyState) { } // waitsTogether reports whether a moved image and its old key's entry must -// both be deferred: neither is uncut, and at least one is in flight. -func waitsTogether(image, old copier.KeyState) bool { +// both be deferred: neither is uncut, and the old key's entry is in flight — +// for its chunk or for its visibility — or the image is invisible. An image +// in flight for its chunk alone does not hold a landed old key's entry back +// (see Drain): the chunk cannot land while that entry's row still holds the +// moved row's unique values in the shadow. +func waitsTogether(image, old copier.KeyState, imageInvisible bool) bool { if image == copier.KeyUncut || old == copier.KeyUncut { return false } - return image == copier.KeyInFlight || old == copier.KeyInFlight + return old == copier.KeyInFlight || imageInvisible } diff --git a/pkg/applier/drain_test.go b/pkg/applier/drain_test.go index b26b5335..3e6a0248 100644 --- a/pkg/applier/drain_test.go +++ b/pkg/applier/drain_test.go @@ -115,31 +115,49 @@ func TestDrainKeyMoveStraddlingTheFrontier(t *testing.T) { assert.Equal(t, 0, b.Len()) } -// The two halves of a move travel together once neither is uncut: a landed -// image waits for the in-flight old key, and a landed old key waits for the -// in-flight image. Neither flushes alone. -func TestDrainKeyMovePairWaitsTogether(t *testing.T) { - cases := map[string]struct { - from, to int64 - }{ - "old key in flight, image landed": {from: 15, to: 25}, - "old key landed, image in flight": {from: 25, to: 15}, - } - for name, tc := range cases { - t.Run(name, func(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}), 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), AllVisible()) - assert.Equal(t, sortedPair(tc.from, tc.to), keys(batch.Entries), "both halves flush in one batch once the chunk lands") - }) - } +// A landed image waits for its in-flight old key: the flush must not copy +// the old key's row while the image still needs it for completion, so +// neither half flushes alone, and both flush once the chunk lands. +func TestDrainKeyMovePairWaitsForTheOldKey(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(keyMove(10, 15, 25, col("label", "a"), marker("doc")))) + + 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), AllVisible()) + assert.Equal(t, []int64{15, 25}, keys(batch.Entries), "both halves flush in one batch once the chunk lands") +} + +// A landed old key does not wait for its in-flight image: the delete at the +// old key flushes now, so the shadow row holding the moved row's unique +// values is gone before the chunk's insert of the row under its new key +// runs again — holding it back would wait on a chunk that is waiting on it +// (CO-4). The image stays buffered, marker intact, until its own chunk +// lands, and is then named for completion, which the flush reads from the +// source row under its key once the old key's shadow row is absent. +func TestDrainLandedOldKeyFlushesAheadOfItsInFlightImage(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "a"), marker("doc")))) + + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) + assert.Equal(t, []int64{25}, keys(batch.Entries), "the delete marker at the landed old key flushes alone") + assert.Equal(t, DeleteMarker, batch.Entries[0].Kind) + assert.Empty(t, batch.CompleteFirst(), "nothing in this batch needs the old key's row") + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 1, batch.Deferred, "the image at 15 waits for its own chunk") + assert.Equal(t, 1, b.Len()) + image := entry(t, b, 15) + assert.Equal(t, Image, image.Kind) + assert.True(t, image.HasMarker(), "the deferred image keeps its marker for a later completion") + + batch = b.Drain(position(t, 30), AllVisible()) + assert.Equal(t, []int64{15}, keys(batch.Entries), "the image flushes once its chunk lands") + assert.Equal(t, []int64{15}, keys(batch.CompleteFirst()), "and is named for completion") + assert.Equal(t, 0, b.Len()) } // An uncut half is discarded on its own and does not hold the other back: a @@ -177,18 +195,39 @@ func TestDrainKeyMoveUncutHalfIsJudgedAlone(t *testing.T) { } // A deferral spreads through a shared old key: two images that both left key -// 25 wait when either of them is in flight, even though 25 itself has landed. +// 25 wait when the entry at 25 does, and an invisible image holds it. An +// image in flight for its chunk alone does not: the landed entry at 25 and +// the other image flush, and only the in-flight image waits. func TestDrainDeferralSpreadsThroughSharedOldKey(t *testing.T) { - b := NewBuffer() - require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "first")))) - require.NoError(t, b.Add(insert(20, 25, col("label", "reinserted")))) - require.NoError(t, b.Add(keyMove(30, 25, 28, col("label", "second")))) + t.Run("one image invisible", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(inTx(keyMove(10, 25, 15, col("label", "first")), 600))) + require.NoError(t, b.Add(inTx(insert(20, 25, col("label", "reinserted")), 650))) + require.NoError(t, b.Add(inTx(keyMove(30, 25, 28, col("label", "second")), 700))) + require.NoError(t, b.Add(inTx(insert(40, 25, col("label", "reinserted again")), 710))) + + // 710 has committed and 700 is still in progress: the entry at 25 is + // visible on its own, and only the image at 28 is not. + snap := Snapshot{read: true, xmin: 700, xmax: 711, inProgress: map[uint64]struct{}{700: {}}} + batch := b.Drain(position(t, 30), snap) + assert.Empty(t, batch.Entries, "28 is invisible, so 25 waits, so 15 waits") + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 1, batch.Invisible) + assert.Equal(t, 2, batch.Deferred) + assert.Equal(t, 3, b.Len()) + }) + t.Run("one image in flight", func(t *testing.T) { + b := NewBuffer() + require.NoError(t, b.Add(keyMove(10, 25, 15, col("label", "first")))) + 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}), 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) - assert.Equal(t, 3, b.Len()) + batch := b.Drain(position(t, 30, [2]int64{11, 20}), AllVisible()) + assert.Equal(t, []int64{25, 28}, keys(batch.Entries), "15 is in flight for its chunk; 25 and 28 are landed and flush") + assert.Equal(t, 0, batch.Discarded) + assert.Equal(t, 1, batch.Deferred) + assert.Equal(t, 1, b.Len()) + }) } // The spread is transitive however long the chain. Four rows each move onto @@ -419,10 +458,3 @@ func TestDrainAgainstTheZeroSnapshotKeepsEverything(t *testing.T) { 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} - } - return []int64{b, a} -} From 03ca221a2e1451114569f4583c773444e1959be9 Mon Sep 17 00:00:00 2001 From: Kiran Muddukrishna Date: Mon, 2 Nov 2026 16:22:21 +1100 Subject: [PATCH 3/3] docs: split CO-4's pair rule and state the retry's rate bound The drain's pair rule now reads as two rules, the chunk retry's bound is stated as a sustained-rate limit, and the key-move test says why its load starts after the copy. --- docs/invariants.md | 33 ++++++++++++++----------- pkg/applier/catchup_integration_test.go | 12 ++++++--- 2 files changed, 26 insertions(+), 19 deletions(-) diff --git a/docs/invariants.md b/docs/invariants.md index 128191c9..e7706424 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -227,20 +227,19 @@ takes it afterwards fails the confirmation (parked-before-insert and parked-befo tests). The applier's flush holds only `ACCESS SHARE` on the shadow until its first write, so a flush transaction of an earlier run parked between confirming the table lock and that write is not one the copier's fence — or the verifier's repair fence — waits on; that gap is open. -`pkg/applier` `Buffer.Drain` — every -buffered key is judged against one `Position` snapshot: landed flushes, uncut is discarded, -in-flight stays buffered until a later drain finds its chunk landed, and a key-moving UPDATE's -two entries are judged per key, except that once neither is uncut the image waits for its old -key's entry — an old key in flight, or invisible, holds its image back so the flush never -deletes or copies the old key's row while the chunk that will copy it is still landing — while a -landed old key's entry flushes ahead of an image in flight for its chunk (an invisible image -still holds its old key's entry back): the delete must not wait for the image's chunk, because -when the moved row carries a unique value the chunk cannot land while the old key's shadow row -holds it, and the image, once its chunk lands, completes from the source row instead (three-way -rule, discard-all before the first claim, deferred entry keeps merging, -straddling move, pair waits for the old key, landed old key flushes ahead of its in-flight image, -uncut half judged alone, deferral spreading through a shared old key and along a chain of -pairs); `Drain` +`pkg/applier` `Buffer.Drain` judges +every buffered key against one `Position` snapshot: landed flushes, uncut is discarded, and +in-flight stays buffered until a later drain finds its chunk landed. A key-moving UPDATE's two +entries are judged per key, and once neither is uncut a pair rule applies. **The image waits +for an old key in flight or invisible**, so the flush never deletes or copies the old key's +row while the chunk that will copy it is still landing. **A landed old key's delete flushes +ahead of an image in flight for its chunk**, because a moved row carrying a unique value cannot +land while the old key's shadow row holds that value; once its chunk lands, the image completes +from the source row. An invisible image still holds its old key's entry back. (Drain unit +tests: three-way rule, discard-all before the first claim, deferred entry keeps merging, +straddling move, pair waits for the old key, landed old key flushes ahead of its in-flight +image, uncut half judged alone, deferral spreading through a shared old key and along a chain +of pairs.) `Drain` judges against a `Position` read after the last `Add`, and returns a `Batch` whose `Deferred` count is what a catch-up is waiting on the copier for and whose `CompleteFirst` names, as copies of its `Entries`, the moved, marker-bearing images the flush must complete — from the old @@ -294,7 +293,11 @@ a catch-up test holds the new key's chunk in flight by hand, asserts the old key is deleted while it is, and converges once the chunk lands). The chunk-level retry remains bounded: a chunk whose key range keeps receiving unique values from landed rows faster than one move per catch-up cycle collides on every attempt with a different row, and the copy stops with -the server's error once the attempts run out. *Test obligation:* a +the server's error once the attempts run out. In practice this is a sustained-rate bound: a +load of key moves or unique-value swaps running through the copy at a few dozen a second can +exhaust the attempts, most often on the last chunk, where every key move lands. Such a load +during the copy is not supported until the copier commits a collided chunk around the rows +that collided and retries only those. *Test obligation:* a marker-bearing UPDATE for a key inside an in-flight chunk asserts the flush waits for the chunk and the row is then completed from the copied shadow row, never an absent-row abort; a marker-bearing UPDATE for a key whose row the source then deleted before its chunk was read, diff --git a/pkg/applier/catchup_integration_test.go b/pkg/applier/catchup_integration_test.go index daccec19..99b38aa0 100644 --- a/pkg/applier/catchup_integration_test.go +++ b/pkg/applier/catchup_integration_test.go @@ -348,10 +348,14 @@ func TestCatchupConvergesUnderMixedLoad(t *testing.T) { } // A row's primary key moves to a fresh identity value, so the change is an -// image under a new key paired with a delete under the old one. With the -// copy landed, both halves are landed: the pair flushes together, and the -// flush completes the image's unchanged blob from the old key's row before -// it deletes that row (CO-4, D13). +// image under a new key paired with a delete under the old one. The load +// starts after the copy, so both halves are landed: this pins the +// both-landed flush, where the pair flushes together and the flush +// completes the image's unchanged blob from the old key's row before it +// deletes that row (CO-4, D13). A move with a half in flight is covered by +// the drain unit tests and the stall test, which holds the new key's chunk +// in flight by hand; a sustained key-move load during the copy is the +// chunk retry's rate bound, which CO-4 states. func TestCatchupConvergesUnderKeyMoves(t *testing.T) { f := newCatchupFixture(t, 1500) run := startCatchup(t, f, applier.CatchupOptions{})