diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index 1713edf9..f122a10a 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -666,7 +666,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 guard recomputes the `Shadow` proof's `ConversionFingerprint` (`chunksql.ResolutionFingerprint`: what those names resolve to across that path, function bodies, strictness, settings, and argument defaults included) inside each chunk transaction, before the insert and again after it before the commit, and refuses as `copy-conversion-drift` when it differs, so under `READ COMMITTED`, which every chunk transaction pins whatever the session default, a chunk that a function or operator created on the path after the build changed rolls back unless the change was undone again before the second check, which the verifier then catches; the proof check refuses conversions without a fingerprint and a fingerprint without conversions — and the rendering settings a text conversion depends on (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), pinned in every chunk transaction so a float or `bytea` converted to `text` lands as the flush and the digest render it; a resumed copy first deletes every shadow row above the watermark in batches of the same guarded shape — the whole shadow after a zero watermark — so the resumed cut frontier and the shadow agree on what is uncut, and its first batch's guard takes the shadow in `SHARE ROW EXCLUSIVE` as its one lock on it (two fences queue rather than deadlock), so a chunk transaction of the earlier run still committing into it, or paused before its insert after confirming the lock, ends before the clear reads; the proof check refuses a shadow that is the source by name or by OID, since the clear deletes from one and the copy reads the other; the copy selects each column the `Shadow` proof names in `Conversions` as that column's USING expression — read from the statement by `statement.ColumnConversions`, since the catalog keeps no record of a USING once `ALTER COLUMN TYPE` has run — and every other column bare, so a retyped column without USING takes the server's assignment cast, and the proof check refuses a conversion of a column the copy does not carry, with no expression, or of the primary key, whose values every chunk range, the watermark, the resume clear, and the conflict target are expressed in; `Position` snapshots the cut frontier and landed watermark as two `Watermark` values plus the in-flight chunks, and `Position.Classify` is the applier's uncut / in-flight / landed rule. While `Run` is in progress the `Copier` is the `progress.Tracker`'s `WorkSource` (`Options.Tracker`): `rows_copied` is the rows this run's committed chunks inserted, `rows_total` the source's catalog row count read once at the start of the run, and `bytes_copied` / `bytes_total` the shadow's and the source's `pg_table_size` measured at each poll — every counter measured, none projected; on a resumed run the row ratio is that run's share, not the copy's completion. Each measurement, and each chunk cut, runs in a read-only transaction of its own under the chunk budgets with the catalog alone on its `search_path`, so a poll or a cut on a caller-built pool is still bounded (LK-2) and still resolves to the real catalog (CO-9); the copier stops being the source before `Run` returns. Every refusal the copier returns is a `*copier.RefusalError` carrying a `copier.RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#copier-refusals-keyed-on-copierrefusalcause), fail-closed under `ErrInvariantViolation`; a chunk 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 guard recomputes the `Shadow` proof's `ConversionFingerprint` (`chunksql.ResolutionFingerprint`: what those names resolve to across that path, function bodies, strictness, settings, and argument defaults included) inside each chunk transaction, before the insert and again after it before the commit, and refuses as `copy-conversion-drift` when it differs, so under `READ COMMITTED`, which every chunk transaction pins whatever the session default, a chunk that a function or operator created on the path after the build changed rolls back unless the change was undone again before the second check, which the verifier then catches; the proof check refuses conversions without a fingerprint and a fingerprint without conversions — and the rendering settings a text conversion depends on (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), pinned in every chunk transaction so a float or `bytea` converted to `text` lands as the flush and the digest render it; a resumed copy first deletes every shadow row above the watermark in batches of the same guarded shape — the whole shadow after a zero watermark — so the resumed cut frontier and the shadow agree on what is uncut, and its first batch's guard takes the shadow in `SHARE ROW EXCLUSIVE` as its one lock on it (two fences queue rather than deadlock), so a chunk transaction of the earlier run still committing into it, or paused before its insert after confirming the lock, ends before the clear reads; the proof check refuses a shadow that is the source by name or by OID, since the clear deletes from one and the copy reads the other; the copy selects each column the `Shadow` proof names in `Conversions` as that column's USING expression — read from the statement by `statement.ColumnConversions`, since the catalog keeps no record of a USING once `ALTER COLUMN TYPE` has run — and every other column bare, so a retyped column without USING takes the server's assignment cast, and the proof check refuses a conversion of a column the copy does not carry, with no expression, or of the primary key, whose values every chunk range, the watermark, the resume clear, and the conflict target are expressed in; `Position` snapshots the cut frontier and landed watermark as two `Watermark` values plus the in-flight chunks, and `Position.Classify` is the applier's uncut / in-flight / landed rule. While `Run` is in progress the `Copier` is the `progress.Tracker`'s `WorkSource` (`Options.Tracker`): `rows_copied` is the rows this run's committed chunks inserted, `rows_total` the source's catalog row count read once at the start of the run, and `bytes_copied` / `bytes_total` the shadow's and the source's `pg_table_size` measured at each poll — every counter measured, none projected; on a resumed run the row ratio is that run's share, not the copy's completion. Each measurement, and each chunk cut, runs in a read-only transaction of its own under the chunk budgets with the catalog alone on its `search_path`, so a poll or a cut on a caller-built pool is still bounded (LK-2) and still resolves to the real catalog (CO-9); the copier stops being the source before `Run` returns. Every refusal the copier returns is a `*copier.RefusalError` carrying a `copier.RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#copier-refusals-keyed-on-copierrefusalcause), fail-closed under `ErrInvariantViolation`; a chunk whose insert collides with a shadow row on a unique index or exclusion constraint (SQLSTATE `23505` / `23P01`) — a value the catch-up has moved between rows and the flush has yet to rewrite — is run again after a bounded linear wait (`Options.CollisionAttempts`, `Options.CollisionBackoff`; `Collisions()` counts them) and, when every attempt collides, the server's `*pgconn.PgError` is returned wrapped; any other chunk the shadow itself rejects is that error as it stands, and neither carries a cause. | CO-4, CO-9, LK-1, LK-2, LK-3 | | `pkg/checksum` | `Verifier` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) compares the source with its shadow up to the copier's landed watermark and returns a `Report` of the chunks that differ. It cuts its own chunks with a `copier.Chunker`, cutting and digesting each chunk in one read-only `REPEATABLE READ` transaction, so the cut and the chunk's two digests describe one snapshot, no snapshot outlives one chunk, and the cut's lock wait is bounded like the reads (LK-2); each transaction runs the copier's guard — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, so a repair's recopy resolves a USING expression as the copy did, and the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, and recomputed once more after a recopy before its commit, so the source side of a digest and a recopy run the resolution the build recorded), `ACCESS SHARE` on both relations taken before the snapshot, lock confirmation, relation-OID check — and pins the rendering settings the copy, the flush, and the digest share (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`), since the digest hashes each row's text rendering and a database or role configured otherwise would render two floats that differ only in their last digits the same, or a `bytea` converted to `text` in the escape form the copy never wrote. Both sides run the same frozen statement — row count plus hex `sha256` of the key-ordered concatenation of each row's `sha256(convert_to(ROW(col::shadow_type, …)::text, 'UTF8'))` over `pk BETWEEN $1 AND $2` (SHA-256 so a FIPS-mode OpenSSL, which refuses `md5()`, does not fail the pass) — so only the data can differ, and the cast on every column is D7: a retyped column hashes as the value the shadow holds, and a column the statement retyped with `USING` hashes on the source side as that expression cast to the shadow's type, the one place the two statements differ. Only the copy columns are compared; generated columns present on both sides are not yet hashed. A `Report` proves nothing. `Check` runs the same pass under a `DivergencePolicy` the caller states every time (the zero value is refused): `abort` returns a `DivergenceError` carrying the report with the shadow untouched; `repair` replaces every differing chunk inside one guarded read-write transaction — delete the shadow's rows over every chunk's key range, then the copy statement itself for every chunk, so a unique value the source moved from one differing chunk to another lands instead of colliding with the stale row — and digests each chunk again in a fresh snapshot, returning a `RepairError` for the first that still differs alongside the `Outcome` listing every repair that committed. A repair pass assumes nothing else writes the shadow and the source rows it recopies hold still until the rereads; a write inside the pass reads as a `RepairError`, never as a second repair; `recopy` is the same recopy without the rereads, for a source that is written while the pass runs, and reports what it recopied the same way; a recopy that collides with a shadow row because the source moved a unique value from a chunk the comparison read equal into a differing chunk (SQLSTATE `23505` or `23P01`) is rolled back, the pass compares again, and the recopy cut from that comparison runs, up to `Options.RecopyAttempts` recopies, with every collision counted in `Outcome.Collisions` and the last one's error ending the pass when all collide. `ParseDivergencePolicy` lets a caller refuse a configured policy before any pass; it accepts `abort` and `repair` and refuses `recopy`, which no setting may name. `CheckLive` is the pass for the catch-up stage, where the source is written and the stream is the shadow's only other writer; it is described under [Live verification](#live-verification). Only a pass that found nothing and repaired nothing mints the proofs, whose constructors are private to the package: a `CleanWatermark` at the watermark it read through, and a `VerifiedShadow` only when that watermark is complete; `Outcome.Clean` is true only when the clean watermark was minted. A pass with repairs returns its `Repair`s and no proof; the next pass mints. While `Verify` or `Check` runs — through the repair phase as well — the verifier is the tracker's `progress.WorkSource` (`Options.Tracker`): `chunks_compared` and `rows_hashed` count the comparison's two-sided digests that have committed and the source rows they covered (every comparison of the pass, so under `recopy` a collision's re-compare counts on top of the first), `chunks_mismatched` the chunks the comparison found differing, `chunks_repaired` the chunks whose one recopy transaction has committed, and `chunks_reread` the chunks digested again after the comparison — the repaired chunks under `Check`, every settling reread under `CheckLive` — every counter read from the pass's memory, none from the database, and all reset when a pass starts; a pass started while another runs on the same verifier is refused. | CO-1, CO-2, CO-3, CO-9, LK-1, LK-2 | | `pkg/decode` | Creates, drops, and inspects the route's slot and publication under the derived name (`CreateSlot` with exported snapshot and consistent point; `DropSlot` waiting out a holder, idempotent, never crossing clusters or databases, dropping only the route's own publication; `InspectSlot` with `wal_status` and retained WAL, the latter unknown rather than zero on a server in recovery; `SlotStatus.WithinLagCeiling` judging a slot against a ceiling in bytes — a `lost` slot is `SlotLostError` before any measure, an unknown measure is `ErrSlotLagUnknown` and never within the ceiling, over it is `SlotLagExceededError` carrying both figures; `DefaultSlotLagCeiling` is one GiB; `IsEngineSlotName` tells the engine's name shape from a name merely wearing its prefix; `InspectSlot` also reads the server's account of a lost slot — `invalidation_reason`, `conflicting`, `inactive_since` — where the server version carries it, one statement across every supported major, and `SlotStatus.LostState` renders it into the `SlotLostError` so an operator reads *why* from the error). `OpenStream` decodes the slot with pgoutput from a position — the consistent point, or the applied position a resume checkpointed — on a dedicated replication connection proven to be a session of the pool's cluster on the target's database — the same table on another cluster derives the same slot name — into one `ChangeEvent` per committed row change: every column carried with its presence, so the unchanged-TOAST marker is an absent `Column` and NULL a present one — built by `PresentColumn` and `AbsentColumn`, read through `Get`, which returns the value together with its presence, so a reader takes the presence with the value — and `OldKey` for an UPDATE that moved the primary key under either admitted replica identity; every event also carries the xid of the transaction that committed it (`Xid`, taken from the transaction's BEGIN), which is how the applier tells a delivered commit the server has not yet made visible from one it has. A relation other than the target — another name, or the target's name under an OID other than the one `CopySwapTarget` was proven for, which is a table dropped and recreated under that name — a tuple that does not line up with it, or a change outside a transaction fails closed; a table whose shape changed under the stream is `ErrSourceShapeChanged`, a TRUNCATE is `ErrUnsupportedChange`, and a warning the server sends in copy-both mode — the walsender saying it will withhold changes, as from a publication it skipped loading — stops the stream fail-closed before a keepalive can deliver a position past them; the replication connection asks for warnings itself (`SET client_min_messages = warning` on the open session, the last word on the setting), so no role, database, or server setting, login event trigger, URL parameter, or connection pooler can keep them from it. The stream keeps `Delivered` — moved by a commit or by a keepalive between transactions, never past a transaction not yet yielded in full — apart from `Confirmed`, the one position it reports to the server: `Confirm` refuses a regression or a position beyond `Delivered`, and a keepalive reply carries only what the caller confirmed, so the slot's resume point moves on the caller's word alone. A wait that runs out in `Next` sends the same report, so the stream is never silent for longer than the caller's wait (`Next`'s doc says why that matters to the walsender). Both positions order transactions by their commit, so a change's own LSN can lie below either; each `ChangeEvent` carries the `Delivered` it arrived with, which is what a caller may confirm while the change is unapplied, and the server replays by commit position, so a resume never skips a replayed change for lying below its checkpoint. `ServerWALEnd` is the walsender's send position from its last keepalive and never below `Delivered`; a change does not move it, since its message carries only the change's own position. `ServerWALEnd − Delivered` is a lower bound on lag — the walsender reports how far it has decoded, not how far the server has written — so lag against the server's WAL end is measured on the pool with `pg_current_wal_lsn()`. The server ending replication with the slot intact is `ErrStreamEnded`, the clean restart ST-4 keeps apart from slot loss. | ST-3, ST-4, CO-4, CO-8 | | `pkg/applier` | `Buffer` merges events into one `Entry` per primary key — each carrying the xid of the latest event merged into it — and `Drain` judges every key against the copier's `Position` and a source `Snapshot` (`ReadSnapshot`, the server's `pg_current_snapshot()`): an entry whose transaction the snapshot does not yet show as committed — the server makes a commit visible only after its WAL flush and any synchronous-standby wait, while the walsender delivers it as soon as the WAL is flushed — is treated as in flight and stays buffered (`Batch.Invisible`), so no chunk read or completion read that cannot see the change decides its fate; the zero `Snapshot` shows nothing visible, `AllVisible` is for a drain under a lock every writer has already released. `Drain` returns a `Batch` whose `CompleteFirst` names the moved, marker-bearing images the flush must complete before it writes. `Flusher` (built from a `CopySwapTarget`, a `copier.Shadow`, and the table's `TableLockSession`) applies one batch in one guarded read-write transaction — owner role, the copy statement's `search_path` (the catalog alone, or the shadow build's path when the proof carries conversions, with the proof's conversion fingerprint recomputed and refused as CO-9 when it differs, before the writes and again after them before the commit, so a USING expression in a flush resolves as the copy's did), `ACCESS SHARE` on both relations, lock confirmation, relation-OID check, the copier's guard — bound to the lock session's context so a lost lock cancels the flush in flight. A column the statement retyped with `USING` is written through that expression (D7): the flush reads the column's source type from the catalog inside its transaction and binds the decoded value as `(SELECT FROM (SELECT $n:: AS "col") AS "source")`, so the value lands as the copy would have landed it, while a completion read from the shadow is already in the shadow's form and is bound bare; the flusher refuses, as ST-6, a proof whose conversions name the primary key, a column the copy does not carry, or an empty expression, and a converted column the source no longer carries. It first reads a completion for every `CompleteFirst` image: from the old key's shadow row, read `FOR UPDATE` so the batch's own delete of that key cannot race the read; from the source row under `Key` when `OldKeyReused` or when the old key's shadow row is absent (a move captured while the old key was uncut left no shadow row). Neither row is known to be the stream's, so the image is not written: it returns in `Result.Held` as a `HeldImage` with its completion and the `pg_current_wal_insert_lsn()` the read took in the same statement, and `Buffer.Hold` takes it back with the completion pending — an event that replaces the key's entry or reuses its old key drops the completion, an UPDATE keeps it pending behind its own values, and `Buffer.Release(passed)` fills the markers still left once the stream has delivered every change committed through the read's position, so the next `Drain` flushes the whole image (D13). An image neither row can complete is written as a delete marker at `Key`, since the source holds no row there, and its key named in `Result.Skipped`. A marker-bearing image of a row that did not move whose shadow row is absent — the source deleted the row before its chunk was read and the DELETE is still on its way — is held the same way, with no completion and `Entry.AbsenceAwaited` set, and flushed once the stream has passed the position of the check that found the row absent; an image that comes back flagged for a row still absent is a CO-8 `ErrInvariantViolation` (D13). The primary path then applies the batch column-wise inside a savepoint, every delete marker before any image: a delete marker is a `DELETE`, a whole image an `INSERT … ON CONFLICT (pk) DO UPDATE`, and a marker-bearing image of a row that did not move an `UPDATE` of only its present columns — zero rows updated is a CO-8 `ErrInvariantViolation`, never an insert that invents the omitted value. On SQLSTATE `23505` or `23P01` (a unique index or an exclusion constraint) the flush rolls back to the savepoint and reapplies the batch whole-row (D13): every remaining marker is completed from the shadow row under its own key `FOR UPDATE` (absent is a CO-8 refusal), every key in the batch is deleted in one statement, and every surviving image is inserted plain; a second collision is `ErrBatchDeferred`, and `Buffer.Requeue` puts the batch back — refusing with a CO-5 `ErrInvariantViolation` if any of its keys is held, since nothing may `Add` between `Drain` and the flush's outcome — so the caller drains again once the copier has landed the chunk the rows wait on. `Result` counts the images and deletes written and the completions, carries the held images and the skipped keys, and says whether the fallback ran. The flush pins the same rendering settings as the copy and the digest (`chunksql.Rendering`: `extra_float_digits = 3`, `bytea_output = 'hex'`) so a float or a `bytea` the decoder rendered lands, through a `USING` conversion to `text`, as the text the copy wrote and the verifier will hash, and the completion reads carry the same pin so a held value lands as the shadow renders it. `Catchup` owns the loop: it gathers the stream into the buffer, `Release`s held images at the delivered position, drains against the copier's `Position` read *after* the last `Add` (so a key the copier cuts between the read and the drain is judged in flight, never discarded), flushes, and on `ErrBatchDeferred` requeues the batch for the next pass while a chunk is in flight — once the copy has landed every key, a refusal is one no chunk can resolve, and `Run` ends with `ErrBatchDeferred` rather than requeue it forever. It confirms the lesser of the delivered position and `Buffer.OldestPending` — each keyed on the `Delivered` the event arrived with, never the event's own LSN — so the slot never advances past a change that is still buffered, held, or deferred, and never falls below what the stream had already confirmed (ST-4); the confirmation follows the requeue or hold, so it never reads a buffer emptier than the one the next drain sees. The whole loop runs under the lock session's `Bind` context, so a lost lock ends a quiet wait on the stream as it ends a flush, reported as `ErrInvariantViolation` wrapping `dbconn.ErrTableLockLost`. `Run` never returns nil: it runs once and returns the context's cause when its owner stops it, carrying any cycle error that is not the stop itself alongside. It is the `progress.WorkSource` of a catch-up step: `changes_applied`, `changes_buffered`, and `lag_bytes` measured against `pg_current_wal_lsn()` read on the pool as each cycle ends, never against the walsender's send position. Each cycle then inspects the stream's slot on the pool and judges it against `CatchupOptions.SlotLagCeiling` (zero takes `decode.DefaultSlotLagCeiling`; below one byte is refused, so the ceiling is always in force): a slot that vanished or that the server reports `lost` ends `Run` with `decode.SlotLostError`, and one retaining more WAL than the ceiling ends it with `decode.SlotLagExceededError` — the table lock stays held and the slot stays in place, so the operator reads what the slot retained and the resume path decides what follows (D11, ST-3). | CO-4, CO-5, CO-6, CO-8, LK-1, LK-2, LK-3, ST-3 | diff --git a/docs/invariants.md b/docs/invariants.md index 762d3ed7..1bfc9c51 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -269,12 +269,20 @@ writer holds its locks until its commit is visible and the swap's `ACCESS EXCLUS 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 -the convergence tests for key moves and unique-key swaps start their load after the copy has -landed. *Test obligation:* a +which the change lands and the catch-up converges). *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_collision_integration_test.go b/pkg/applier/catchup_collision_integration_test.go new file mode 100644 index 00000000..077c203c --- /dev/null +++ b/pkg/applier/catchup_collision_integration_test.go @@ -0,0 +1,133 @@ +package applier_test + +import ( + "fmt" + "math" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/applier" + "github.com/block/pg-sprite/pkg/copier" +) + +// A source change moves a unique value from a landed key to a key the copier +// has yet to copy, and the catch-up's flushes are held, so the landed row's +// shadow copy keeps the value when the copier reaches the other key: the +// chunk carrying it collides on the shadow's unique index. The copier runs +// the chunk again after a wait, the resumed catch-up flushes the landed +// row's half of the move, the next attempt lands, and the shadow converges +// (CO-4). Holding the flushes until the collision is seen makes the race +// happen on every run rather than when the timing allows it. +func TestCatchupConvergesWhenTheCopierCollidesWithAnUnflushedMove(t *testing.T) { + f := newCatchupFixture(t, 1500) + // The seeded rows hold uniq = id. Once the copier has landed key 10, + // uniq 10 moves to key 1000, far past the chunks in flight then. + move := &moveOnLanding{ + t: t, + f: f, + landed: 10, + statements: []string{ + `UPDATE %s SET uniq = 1000000 WHERE id = 10`, + `UPDATE %s SET uniq = 10 WHERE id = 1000`, + }, + } + run := startCatchupWith(t, f, catchupSetup{copierClock: move}) + move.shadow = pgx.Identifier{f.table.Schema, run.shadow.ShadowTable()}.Sanitize() + + resume, err := run.catchup.Pause(t.Context()) + require.NoError(t, err) + t.Cleanup(resume) + + copied := make(chan error, 1) + var wg sync.WaitGroup + wg.Go(func() { copied <- run.copier.Run(t.Context(), f.pool) }) + t.Cleanup(wg.Wait) + + const collisionDeadline = 15 * time.Second + require.Eventually(t, func() bool { return run.copier.Collisions() >= 1 }, collisionDeadline, 10*time.Millisecond, + "the chunk carrying key 1000 should collide on uniq 10, which landed key 10 still holds in the shadow") + require.NoError(t, move.result()) + + resume() + const copyDeadline = 30 * time.Second + select { + case err := <-copied: + require.NoError(t, err, "the collided chunk lands once the flush has moved uniq 10 off key 10") + case <-time.After(copyDeadline): + require.FailNow(t, "the copy did not finish after the flushes resumed") + } + assert.Equal(t, copier.NewWatermark(math.MaxInt64), run.copier.Position().Cut, "every key is landed") + run.awaitStatus(t, "flush the move's half on landed key 10", func(s applier.Status) bool { + return s.Applied > 0 && s.Buffered == 0 + }) + + run.stop(t) + f.assertConverged(t, run.shadow) +} + +// moveOnLanding is the copier's clock. On the first reading after the shadow +// holds key landed, it commits statements against the source in one +// transaction, once; each statement's %s is the source table. Every reading +// returns the wall clock. +type moveOnLanding struct { + t *testing.T + f catchupFixture + landed int64 + statements []string + // shadow is the quoted shadow table, set before the copier runs. + shadow string + + mu sync.Mutex + moved bool + err error +} + +func (m *moveOnLanding) Now() time.Time { + m.mu.Lock() + defer m.mu.Unlock() + if !m.moved && m.err == nil { + m.moved, m.err = m.tryMove() + } + return time.Now() +} + +func (m *moveOnLanding) tryMove() (bool, error) { + ctx := m.t.Context() + var landed bool + if err := m.f.pool.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM "+m.shadow+" WHERE id = $1)", m.landed).Scan(&landed); err != nil { + return false, fmt.Errorf("read whether key %d has landed: %w", m.landed, err) + } + if !landed { + return false, nil + } + err := pgx.BeginFunc(ctx, m.f.pool, func(tx pgx.Tx) error { + for _, statement := range m.statements { + if _, err := tx.Exec(ctx, fmt.Sprintf(statement, m.f.table.Qualified())); err != nil { + return err + } + } + return nil + }) + if err != nil { + return false, fmt.Errorf("move a unique value off landed key %d: %w", m.landed, err) + } + return true, nil +} + +// result reports whether the move was committed. +func (m *moveOnLanding) result() error { + m.mu.Lock() + defer m.mu.Unlock() + if m.err != nil { + return m.err + } + if !m.moved { + return fmt.Errorf("key %d never landed, so the move was not made", m.landed) + } + return nil +} diff --git a/pkg/applier/catchup_integration_test.go b/pkg/applier/catchup_integration_test.go index 2114f9f8..45501b61 100644 --- a/pkg/applier/catchup_integration_test.go +++ b/pkg/applier/catchup_integration_test.go @@ -127,6 +127,9 @@ type catchupSetup struct { // keys. The copy then runs to completion before the catch-up starts, so // the shadow holds every row, and the test moves the position by hand. positions applier.PositionSource + // copierClock, when set, is the copier's clock, which it reads as each + // chunk attempt starts and as each one commits. + copierClock progress.Clock } func startCatchup(t *testing.T, f catchupFixture, opts applier.CatchupOptions) *catchupRun { @@ -178,6 +181,7 @@ func startCatchupWith(t *testing.T, f catchupFixture, setup catchupSetup) *catch c, err := copier.NewCopier(f.target, shadow, lock, copier.Watermark{}, copier.Options{ Workers: 2, Chunker: copier.ChunkerOptions{InitialRows: 50, MaxRows: 50}, + Clock: setup.copierClock, }) require.NoError(t, err) run := &catchupRun{f: f, shadow: shadow, lock: lock, copier: c, done: make(chan error, 1)} @@ -348,9 +352,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{}) @@ -449,6 +454,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 aabf9c6b..43259352 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{} } @@ -95,6 +113,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 } @@ -130,6 +154,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 @@ -217,6 +244,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 @@ -285,12 +320,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) @@ -300,15 +337,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) +}