diff --git a/SAFETY.md b/SAFETY.md index 81fa011..54c7c28 100644 --- a/SAFETY.md +++ b/SAFETY.md @@ -25,7 +25,7 @@ The invariant registry (invariant IDs referenced below) lives in | `pkg/internal/chunksql` — the chunk insert statement the copier and the verifier's repair both run | ✅ core | exists; unexported from the module so no caller can run the statement outside the guard both packages wrap around it | CO-4 (copy SQL shape) | | `pkg/applier` — change apply, buffer, flush scheduling | ✅ core | package contract exists; applier planned | CO-4, CO-5, CO-6, CO-8, LK-3 | | `pkg/decode` — logical decoding, LSN/position accounting, per-column presence | ✅ core | contract types exist; decoder planned | ST-4, CO-4, CO-8 | -| `pkg/checkpoint` — durable resume state | ✅ core | checkpoint contract exists; persistence planned | ST-1, ST-2 | +| `pkg/checkpoint` — durable resume state | ✅ core | contract and persistence exist: `Store` over a `pkg/dbconn` pool creates `pgsprite.pgsprite_checkpoint` on first use under the engine's advisory key, creating only what is absent and refusing, typed, a schema or table another role owns (`Ensure`), writes one row per target in one guarded upsert — under the target's table lock session, confirmed from the write's own transaction — that refuses, typed, to overwrite a row carrying another statement's fingerprints or another row format (`Save`), reads it back telling a matching row, no row (`ErrNotFound`), a missing table (`ErrTableMissing`), an `IncompatibleError`, and a retried transient read error apart (`Load`), and removes it as the one explicit fresh start, matching the row identity the caller was shown (`Delete`); the resume state machine that drives it is planned | ST-1, ST-2, LK-1 | | slot lifecycle (in `pkg/decode`) — create, reap, lag ceiling | ✅ core | planned (Phase 8) | ST-3 | | `pkg/schemachange` — shadow builder, orchestrator, **cutover swap + fidelity gate** | ✅ core | shadow lifecycle exists (`BuildShadow` → `BuiltShadow`, `DropShadow`, `InspectShadow`, `SourceOfDerivedName`), the cutover fidelity gate exists (`GateCutover` → `CutoverReady`), and the swap exists (`Cutover` → `SwappedTable`, `DropOldTable`), every operation taking the `*dbconn.TableLockSession` it runs under; the orchestrator that chains them is planned | LK-1, LK-2, ST-5, ST-6, ST-7 (shadow build, drop, inspect); CO-1, ST-5, ST-6 (cutover gate); LK-1, LK-2, LK-4, RF-2, ST-5, ST-6 (swap and old-table drop); RF-1, RF-3..RF-6 at the orchestrator (planned) | | `pkg/statement`, `pkg/planner`, `pkg/schemadiff`, `pkg/router`, `pkg/plan`, `pkg/lint`, `pkg/suggest` — classify/diff/route/report | ❌ periphery¹ | `pkg/statement` (parse boundary), `pkg/schemadiff` (introspect/diff via scratch execute-and-introspect), `pkg/planner` (classifier), `pkg/router` (backend assignment + availability policy), `pkg/plan` (versioned dry-run plan report), `pkg/lint` (offline typed findings), and `pkg/suggest` (advisory rewrites with typed caveats) exist (Phases 2.1–2.5) | (CO-7 holds at the parse boundary) | diff --git a/docs/copy-and-swap-design.md b/docs/copy-and-swap-design.md index 5ac9244..2935985 100644 --- a/docs/copy-and-swap-design.md +++ b/docs/copy-and-swap-design.md @@ -87,8 +87,13 @@ needs separate resumability, validity, uniqueness, and resource controls. ### D3 — Store checkpoints in the target database -**Decision.** The engine role creates `pgsprite_checkpoint` in the target database on first use. -It contains one row per `(schema, table)`. +**Decision.** The engine role creates `pgsprite_checkpoint` in the target database on first use, +in an engine-owned `pgsprite` schema — the target's schemas are never written to, and `public` +is not assumed writable (PostgreSQL 15 revoked `CREATE` on it from `PUBLIC`). It contains one +row per `(schema, table)`, keyed on the row format version and the two model fingerprints +(ST-2); a write for a target whose row carries another statement's fingerprints or another +format is refused, typed, and a fresh start deletes the row explicitly first. Creating the +schema needs `CREATE` on the database, the same privilege the scratch schema already needs. **Why.** Ordinary table data follows the database through failover and gives each target one atomic, local resume record. @@ -406,7 +411,7 @@ decoding but adds write-path availability and amplification costs. | `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` and digests each in one read-only `REPEATABLE READ` transaction, so a chunk's two digests describe one snapshot and no snapshot outlives one chunk; each transaction runs the copier's guard — owner role, catalog-only `search_path`, `ACCESS SHARE` on both relations taken before the snapshot, lock confirmation, relation-OID check — and pins `extra_float_digits` to its maximum, since the digest hashes each row's text rendering and a database or role configured at zero or below would render two floats that differ only in their last digits the same. Both sides run the identical frozen statement — row count plus `md5` of the key-ordered concatenation of each row's `md5(ROW(col::shadow_type, …)::text)` over `pk BETWEEN $1 AND $2` — so only the data can differ, and the cast on every column is D7: a converted column hashes as the value the shadow holds. 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. `ParseDivergencePolicy` lets a caller refuse a configured policy before any pass. 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. | CO-1, CO-2, CO-3, CO-9, LK-1 | | `pkg/decode` | Produces `ChangeEvent`, including per-column presence and `OldKey` for an UPDATE that moved the primary key. | ST-3, ST-4, CO-4, CO-8 | | `pkg/applier` | Applies presence-aware events from the per-key buffer. | CO-4, CO-5, CO-6, CO-8, LK-3 | -| `pkg/checkpoint` | Produces `Checkpoint`. | ST-1, ST-2 | +| `pkg/checkpoint` | Produces `Checkpoint` and persists it: `Store` (over a `pkg/dbconn` pool) creates `pgsprite.pgsprite_checkpoint` on first use (`Ensure`, serialized under the engine's advisory key so concurrent first users never race the create; it creates only what is absent, so a pre-provisioned schema and table the engine owns need no database `CREATE`, and refuses with `ErrForeignObject` a schema or table another role owns), `Save` takes the target's `TableLockSession`, confirms it from the write's own transaction, and writes the target's one row in one `INSERT … ON CONFLICT DO UPDATE` whose update is guarded on the row's format version and fingerprints — zero rows updated is an `IncompatibleError`, never a silent overwrite — `Load` returns the row for a run with the same `Fingerprints`, `ErrNotFound` for no row, `ErrTableMissing` when `Ensure` never ran, an `IncompatibleError` naming the first disagreeing field (format, then source, then target fingerprint) and carrying the row's whole `Identity`, or, after bounded retries through transient errors (`dbconn.Retryable` plus a session the server ended from outside it, `57P01`/`57P02`/`57P03`) with an injected sleep, an error that is none of those; `Delete`, under the same lock, is the explicit fresh start and removes only the row whose `Identity` the caller was shown. The watermark column is `NULL` while nothing has landed, the LSN is a `pg_lsn`, and the phase is stored by its stable name. | ST-1, ST-2, LK-1 | | `pkg/schemachange` | Shadow builder (`BuildShadow` produces `BuiltShadow`: source and shadow OIDs, fingerprints, identity handoff, copy columns, fidelity snapshot; `SourceOfDerivedName` maps a derived name back to its table), orchestrator, and cutover. `BuildShadow`, `DropShadow`, and `InspectShadow` each take the `*dbconn.TableLockSession` for the table — a dedicated direct server session, distinct from the working pool, that `dbconn.AcquireTableLock` refuses to open through a transaction-pooling proxy: the operation runs under the session's `Bind` context so a lost lock cancels the statement in flight, and its transaction re-asserts from its own connection that the session's backend holds the lock before the first write, since the lock and the work are deliberately on different sessions. `InspectShadow` is the resume path `ErrShadowExists` points at: it re-derives the `BuiltShadow` proof from the catalog for the caller to compare with its checkpoint — `BuiltShadow.Proof()` is the plain, JSON-encodable view of that proof (`BuiltShadow` marshals as it), and nothing decodes back into a `BuiltShadow`, so only the builder and the inspection mint one; `DropShadow` is the D5 cleanup, dropping only a plain table the source's owner owns, without `CASCADE`. `GateCutover` is the ST-5 fidelity gate: handed a `BuiltShadow` and a `checksum.VerifiedShadow` for the same table, it re-reads both relations under the lock, refuses on a moved OID, a fingerprint or fidelity snapshot that drifted from the build's record, an invalid shadow index, or a swap name already taken, and otherwise mints `CutoverReady` — the pairing of every source index and extended statistics object with its shadow counterpart by catalog definition (not by name, since `LIKE` renames them; the default operator class and the column's own collation of a key column the statement retyped are set aside, since the server re-derives them for the new type; a written operator class or collation still has to agree, and a relaxed definition two indexes share pairs neither), and the sequences the swap must re-own; the swap itself consumes that proof. `Cutover` is the swap: one transaction under the lock session that takes `ACCESS EXCLUSIVE` on the source and the shadow under `lock_timeout` with bounded retry and backoff, runs the caller's `DrainFunc` (nil for a quiesced table) once per attempt once writers are excluded — a rolled-back attempt takes the drain's work with it, so the next attempt drains again from the same state — re-runs the copy-and-swap shape check (RF-2) and the gate's checklist so the pairing it renames by is as fresh as the lock, performs the D8 renames (every source dependent to its derived `_old` name, the source to `_old`, the shadow to the source's name, each paired shadow dependent to its partner's name), completes the D5 handoff (re-owning shared sequences; recreating each identity column with the source sequence's declared options — bounds following a widened column's type — under the source sequence's name, carrying the source sequence's grants, and `setval` to its `(last_value, is_called)`), re-reads the catalog to confirm every rename and handoff before committing, and mints `SwappedTable`; an attempt whose connection broke or whose context ended is resolved by following its backend to its exit and then reading from a fresh connection which OID bears the source name, never assumed (LK-4), and every attempt the catalog shows rolled back is reported wrapping `ErrCutoverRolledBack`. `DropOldTable` is the D9 drop: a separate bounded transaction that drops only the relation whose OID the `SwappedTable` recorded as the source, without `CASCADE`. Every refusal the six operations return is a `*RefusalError` carrying a `RefusalCause` from the closed set in [refusal-classes.md](refusal-classes.md#shadow-operation-refusals-keyed-on-refusalcause), so an importer routes on the cause rather than on message text. | LK-1, LK-2, LK-4, CO-1, ST-5, ST-6, ST-7 | Each producing package owns its types. `pkg/schemachange` imports every producer; no producer diff --git a/docs/engine-role.md b/docs/engine-role.md index dfac10e..f09f194 100644 --- a/docs/engine-role.md +++ b/docs/engine-role.md @@ -57,7 +57,18 @@ for execute-and-introspect; it does not add a higher privilege tier or require ` [the D1 decision](copy-and-swap-design.md#d1--no-durable-scratch-database). The scratch schema does need `CREATE` **on the database** (`CREATE SCHEMA` is a database-level privilege) — a requirement of every declarative plan, not only copy-and-swap, that the tier table does not yet -carry and preflight's privilege probe does not yet check; both are open follow-ups. +carry and preflight's privilege probe does not yet check; both are open follow-ups. The +checkpoint table ([D3](copy-and-swap-design.md#d3--store-checkpoints-in-the-target-database)) +lives in an engine-owned `pgsprite` schema created on first use, so a copy-and-swap run needs +the same database-level `CREATE` once, and `pkg/checkpoint` reports the server's refusal as is +when the engine lacks it. A deployment that keeps database-level `CREATE` away from the engine +role can pre-provision instead: have a role that holds `CREATE` run the engine's first use +once, which creates the `pgsprite` schema and the `pgsprite.pgsprite_checkpoint` table, then +`ALTER SCHEMA pgsprite OWNER TO ` and `ALTER TABLE pgsprite.pgsprite_checkpoint OWNER TO +`. `Ensure` creates only what is absent, so it issues no `CREATE` against objects that +already exist, and it refuses — rather than adopts — a schema or table that another role still +owns, because anything that role hangs off them would run with the engine's privileges on every +checkpoint write. Two cluster-level *facts* — settings, not grants — accompany Tier 3 and are checked by `preflight.CheckCopySwapEnvironment` once the privilege check has passed and the shape check diff --git a/docs/invariants.md b/docs/invariants.md index 0e64a6b..0f13554 100644 --- a/docs/invariants.md +++ b/docs/invariants.md @@ -321,7 +321,11 @@ gone-session, reported-loss, and mid-pass-loss tests); `pkg/schemachange` `GateC confirm from the swap's and the drop's own transactions that the session's backend holds the lock before the first rename or the drop (nil-session, rival-backend, and lost-mid-attempt tests for the swap; nil-session and rival-backend tests for the drop), so loss of the lock -aborts the change at every stage. +aborts the change at every stage; `pkg/checkpoint` `Store.Save` and `Store.Delete` require the +same session for the checkpoint's target, run under its `Bind` context, and confirm the lock +from the write's own transaction before the upsert or the delete (nil-session, wrong-table, +and rival-backend tests), so a run whose lock went to another engine cannot stamp its stale +checkpoint over the newer engine's row. *Source:* Spirit `pkg/dbconn/metadatalock.go` (stated pool invariants). This resolves the mutual-exclusion gap called out in the validation review. @@ -519,9 +523,11 @@ The [atomic RLS contract](atomic-row-security.md) defines these executor obligat The checkpoint table keeps **one row per `(schema, table)`** (upsert on that key) so a crash can never leave a partial pair for one target — its record is either the old or the new one. -Unbounded append-style checkpoint history is not used. *Planned enforcement (Phase 8):* -`pkg/checkpoint` write path (`INSERT … ON CONFLICT (schema_name, table_name) DO UPDATE`, the REPLACE -analog). *Source:* Spirit `pkg/checkpoint` (single-row REPLACE on `id=1`), scoped per target by +Unbounded append-style checkpoint history is not used. *Enforced:* `pkg/checkpoint` +`Store.Save` — one `INSERT … ON CONFLICT (schema_name, table_name) DO UPDATE` (the REPLACE +analog) per save, under the target's table lock (LK-1), so a reader sees the previous record or +the new one and never a mix; `TestSaveUpsertsTheOneRowPerTarget`, `TestSaveKeepsOneRowPerTarget`, +`TestSaveRefusesWhenAnotherBackendHoldsTheTable`. *Source:* Spirit `pkg/checkpoint` (single-row REPLACE on `id=1`), scoped per target by [copy-and-swap D3](copy-and-swap-design.md#d3--store-checkpoints-in-the-target-database). ### ST-2 — An incompatible checkpoint is distinguishable from a transient read error @@ -529,10 +535,25 @@ analog). *Source:* Spirit `pkg/checkpoint` (single-row REPLACE on `id=1`), scope Resume must tell apart: (a) a readable, matching checkpoint → resume; (b) a checkpoint written by an incompatible engine version or for a **different statement** → refuse to resume, start fresh (never mix state across versions/statements); (c) a *transient* read failure → retry, and never -trigger fresh-start recovery on a blip. *Enforced:* checkpoint read/validation path (version + -statement fingerprint stored with the watermark; the fingerprint will hash the -execute-and-introspect after-schema model, not SQL text, so textually-different-but-identical statements match and -cosmetic edits don't force a fresh start). *Source:* Spirit `checkpoint.IsIncompatible` + +trigger fresh-start recovery on a blip. *Enforced:* `pkg/checkpoint` `Store.Load` returns the +`Checkpoint` for (a), a typed `IncompatibleError` for (b) — the row format version and both +model fingerprints are stored with the watermark, and the fingerprints are +`pkg/schemachange`'s digests of the execute-and-introspect source and after-schema models, not +SQL text, so textually-different-but-identical statements match and cosmetic edits don't force +a fresh start — and for (c) retries through transient errors (what `dbconn.Retryable` names, plus a session the +server ended from outside it, `57P01`/`57P02`/`57P03`, which only a read may safely repeat) under bounded attempts +before returning an error that is neither `ErrNotFound` nor incompatible; `ErrNotFound` is +returned only for a completed read that found no row, and a database where the checkpoint table +was never created is the typed `ErrTableMissing`, never `ErrNotFound`. `Store.Save` applies the +same identity guard on the write path, so a run can never write its state over another +statement's row; `Delete` is the explicit fresh start, and it removes only the row whose +`Identity` the caller was shown (`IncompatibleError.Stored`), so a row another engine wrote in +between survives and surfaces as a new `IncompatibleError`. `TestLoadRetriesAcrossATerminatedBackend`, +`TestLoadReportsAnotherStatementsRowAsIncompatible`, +`TestLoadReportsAnotherFormatVersionAsIncompatible`, +`TestSaveRefusesToOverwriteAnotherStatementsRow`, `TestDeleteRefusesARowTheCallerWasNotShown`, +`TestWritesAndReadsWithoutEnsureReportTheMissingTable`, +`TestResumeFromCheckpointConvergesAfterAMidCopyKill`. *Source:* Spirit `checkpoint.IsIncompatible` + "resume requires the identical ALTER". ### ST-3 — Slot cleanup is guaranteed on success, failure, and crash diff --git a/pkg/checkpoint/checkpoint.go b/pkg/checkpoint/checkpoint.go index 47b3f4d..7e6bc96 100644 --- a/pkg/checkpoint/checkpoint.go +++ b/pkg/checkpoint/checkpoint.go @@ -53,17 +53,89 @@ func (p Phase) String() string { } } +// parsePhase reads a stable phase name back into its Phase. A name String +// never produces is an error: the store wrote the row, so an unknown name +// is a row the store cannot have written. +func parsePhase(name string) (Phase, error) { + for p := PhaseCopying; p <= PhaseFailed; p++ { + if p.String() == name { + return p, nil + } + } + return 0, fmt.Errorf("unknown phase %q", name) +} + +// Fingerprints identifies the statement a checkpoint belongs to: the digests +// of the source's and the shadow's introspected models. Two runs with the +// same pair are the same schema change; a row carrying another pair is +// another statement's state (ST-2). +type Fingerprints struct { + // Source is the digest of the source table's introspected model. + Source string + // Target is the digest of the shadow's introspected model after the + // statement ran on it: the after-schema, not the SQL text. + Target string +} + +// Identity is everything a stored row is guarded on: the row format it was +// written in and the statement it belongs to. Save refuses to write over a +// row with another Identity, and Delete removes a row only when its +// Identity is the one the caller was shown. +type Identity struct { + FormatVersion int32 + Fingerprints Fingerprints +} + // Checkpoint is the single durable resume record. type Checkpoint struct { - Schema string - Table string - ShadowTable string - SlotName string - PublicationName string - Watermark copier.Watermark - LastAppliedLSN decode.LSN + Schema string + Table string + ShadowTable string + SlotName string + PublicationName string + // Watermark is the copier's landed frontier; its zero value means + // nothing has landed yet. + Watermark copier.Watermark + LastAppliedLSN decode.LSN + // SourceFingerprint and TargetFingerprint are the Fingerprints the row + // is keyed on. SourceFingerprint string TargetFingerprint string Phase Phase - UpdatedAt time.Time + // UpdatedAt is when the record was last saved, from the store's clock. + UpdatedAt time.Time +} + +// Fingerprints returns the statement identity the checkpoint carries. +func (c Checkpoint) Fingerprints() Fingerprints { + return Fingerprints{Source: c.SourceFingerprint, Target: c.TargetFingerprint} +} + +// Identity returns the row identity a Save of this checkpoint writes: the +// current FormatVersion and the checkpoint's fingerprints. A run that wants +// to discard its own row hands it to Delete. +func (c Checkpoint) Identity() Identity { + return Identity{FormatVersion: FormatVersion, Fingerprints: c.Fingerprints()} +} + +// validate refuses a checkpoint the store must not persist: the key, the +// fingerprints, and the phase are what resume keys on, so none may be empty +// or unknown. The slot and publication names may be empty — a quiesced run +// decodes no WAL and has neither. +func (c Checkpoint) validate() error { + for _, field := range []struct{ name, value string }{ + {"schema", c.Schema}, + {"table", c.Table}, + {"shadow table", c.ShadowTable}, + {"source fingerprint", c.SourceFingerprint}, + {"target fingerprint", c.TargetFingerprint}, + } { + if field.value == "" { + return fmt.Errorf("%w: %s is empty", ErrInvalidCheckpoint, field.name) + } + } + if _, err := parsePhase(c.Phase.String()); err != nil { + return fmt.Errorf("%w: %w", ErrInvalidCheckpoint, err) + } + return nil } diff --git a/pkg/checkpoint/checkpoint_test.go b/pkg/checkpoint/checkpoint_test.go index 19635de..9105130 100644 --- a/pkg/checkpoint/checkpoint_test.go +++ b/pkg/checkpoint/checkpoint_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestPhaseString(t *testing.T) { @@ -19,3 +20,53 @@ func TestPhaseTerminal(t *testing.T) { assert.Equal(t, want, phase.Terminal(), "phase %s", phase) } } + +// Every phase name String produces parses back to its Phase; names String +// never produces, including the unknown-phase rendering, do not. +func TestParsePhaseInvertsString(t *testing.T) { + for p := PhaseCopying; p <= PhaseFailed; p++ { + got, err := parsePhase(p.String()) + require.NoError(t, err, p) + assert.Equal(t, p, got) + } + for _, bad := range []string{"", "Copying", "paused", "Phase(99)"} { + _, err := parsePhase(bad) + assert.Error(t, err, bad) + } +} + +func validCheckpoint() Checkpoint { + return Checkpoint{ + Schema: "app", Table: "orders", ShadowTable: "_pgsprite_orders_new", + SourceFingerprint: "src-a", TargetFingerprint: "tgt-a", Phase: PhaseCopying, + } +} + +// The key, the shadow, the fingerprints, and the phase are what resume keys +// on, so a checkpoint missing any of them is refused before it reaches the +// database; the slot and publication names may be empty because a quiesced +// run has none. +func TestCheckpointValidate(t *testing.T) { + require.NoError(t, validCheckpoint().validate()) + + cases := map[string]func(*Checkpoint){ + "empty schema": func(c *Checkpoint) { c.Schema = "" }, + "empty table": func(c *Checkpoint) { c.Table = "" }, + "empty shadow table": func(c *Checkpoint) { c.ShadowTable = "" }, + "empty source fingerprint": func(c *Checkpoint) { c.SourceFingerprint = "" }, + "empty target fingerprint": func(c *Checkpoint) { c.TargetFingerprint = "" }, + "zero phase": func(c *Checkpoint) { c.Phase = 0 }, + "unknown phase": func(c *Checkpoint) { c.Phase = Phase(99) }, + } + for name, mutate := range cases { + t.Run(name, func(t *testing.T) { + cp := validCheckpoint() + mutate(&cp) + assert.ErrorIs(t, cp.validate(), ErrInvalidCheckpoint) + }) + } +} + +func TestCheckpointFingerprints(t *testing.T) { + assert.Equal(t, Fingerprints{Source: "src-a", Target: "tgt-a"}, validCheckpoint().Fingerprints()) +} diff --git a/pkg/checkpoint/doc.go b/pkg/checkpoint/doc.go index 9039ed9..1eb914d 100644 --- a/pkg/checkpoint/doc.go +++ b/pkg/checkpoint/doc.go @@ -1,2 +1,6 @@ -// Package checkpoint defines durable resume contracts enforcing ST-1 and ST-2. +// Package checkpoint persists the one durable resume record per target — +// the Checkpoint — in the target database, enforcing ST-1 (one row per +// target, written atomically) and ST-2 (an incompatible record is +// distinguishable from a transient read error, and neither is mistaken for +// the absence of a record). package checkpoint diff --git a/pkg/checkpoint/ensure.go b/pkg/checkpoint/ensure.go new file mode 100644 index 0000000..4373a9d --- /dev/null +++ b/pkg/checkpoint/ensure.go @@ -0,0 +1,139 @@ +package checkpoint + +import ( + "context" + "errors" + "fmt" + + "github.com/jackc/pgx/v5" + + "github.com/block/pg-sprite/pkg/dbconn" +) + +// tableIdent is the schema-qualified checkpoint table as it appears in +// SQL: SchemaName and TableName quoted, which a unit test pins to what +// pgx.Identifier would render for them. Both names are the engine's own +// constants, never a caller's. +const tableIdent = `"` + SchemaName + `"."` + TableName + `"` + +// createSchemaSQL creates the engine schema. It runs only when the schema +// is absent; CREATE SCHEMA checks database-level CREATE before it checks +// for existence, so running it against a pre-provisioned schema would +// refuse an engine role that was deliberately never granted CREATE. +const createSchemaSQL = `CREATE SCHEMA "` + SchemaName + `"` + +// createTableSQL creates the checkpoint table. watermark is NULL while +// nothing has landed; phase holds the stable Phase name so an operator +// reading the row during an incident sees "copying", not a number. +const createTableSQL = "CREATE TABLE " + tableIdent + ` ( + schema_name text NOT NULL, + table_name text NOT NULL, + format_version integer NOT NULL, + shadow_table text NOT NULL, + slot_name text NOT NULL, + publication_name text NOT NULL, + watermark bigint, + last_applied_lsn pg_lsn NOT NULL, + source_fingerprint text NOT NULL, + target_fingerprint text NOT NULL, + phase text NOT NULL, + updated_at timestamptz NOT NULL, + PRIMARY KEY (schema_name, table_name) +)` + +// ensureLockSQL takes the engine's own advisory key for the checkpoint +// table — the key pkg/dbconn's table lock would derive for it — for the +// rest of the transaction, so two engines reaching the same database for +// the first time at once serialize their creates instead of racing them +// into a duplicate-key failure in the catalog, and so the owner read below +// cannot interleave with another engine's create. +const ensureLockSQL = "SELECT pg_advisory_xact_lock($1, hashtext(quote_ident($2) || '.' || quote_ident($3)))" + +// ownersSQL reads who owns the engine schema and whatever wears the +// checkpoint table's name inside it, alongside the role this session runs +// as. No row means no schema; a NULL relation means the schema exists and +// the table does not. +const ownersSQL = ` +SELECT current_user, pg_get_userbyid(n.nspowner), c.relkind::text, pg_get_userbyid(c.relowner) +FROM pg_namespace n +LEFT JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = $2 +WHERE n.nspname = $1` + +// relkindOrdinaryTable is pg_class.relkind for a plain table. +const relkindOrdinaryTable = "r" + +// engineObjects is what Ensure found under the engine's names: which of +// the two objects exist, and whether the engine owns them. +type engineObjects struct { + schemaExists bool + tableExists bool +} + +// Ensure makes the checkpoint table usable by this engine role and is a +// no-op once it is. It creates only what is absent: a pre-provisioned +// schema and table that the engine owns are accepted as they are, so a +// deployment that keeps database-level CREATE away from the engine role can +// create them for it beforehand. A schema or table that exists under +// another owner, or a relation under the table's name that is not a plain +// table, is refused with ErrForeignObject rather than adopted. +func (s *Store) Ensure(ctx context.Context) error { + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin create of %s: %w", tableIdent, err) + } + defer func() { + // Redundant safety closer: after a successful Commit this returns + // the guaranteed ErrTxClosed; on a failure path the server aborts + // the transaction with its session either way. + _ = tx.Rollback(context.WithoutCancel(ctx)) + }() + if _, err := tx.Exec(ctx, ensureLockSQL, dbconn.TableLockClassID, SchemaName, TableName); err != nil { + return fmt.Errorf("lock for create of %s: %w", tableIdent, err) + } + found, err := inspectEngineObjects(ctx, tx) + if err != nil { + return err + } + if !found.schemaExists { + if _, err := tx.Exec(ctx, createSchemaSQL); err != nil { + return fmt.Errorf("create schema %q: %w", SchemaName, err) + } + } + if !found.tableExists { + if _, err := tx.Exec(ctx, createTableSQL); err != nil { + return fmt.Errorf("create %s: %w", tableIdent, err) + } + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit create of %s: %w", tableIdent, err) + } + return nil +} + +// inspectEngineObjects reads the engine schema and table from the catalog +// and refuses, typed, anything under their names that is not this role's +// own plain table. +func inspectEngineObjects(ctx context.Context, tx pgx.Tx) (engineObjects, error) { + var self, schemaOwner string + var relkind, tableOwner *string + err := tx.QueryRow(ctx, ownersSQL, SchemaName, TableName).Scan(&self, &schemaOwner, &relkind, &tableOwner) + if errors.Is(err, pgx.ErrNoRows) { + return engineObjects{}, nil + } + if err != nil { + return engineObjects{}, fmt.Errorf("inspect %s: %w", tableIdent, err) + } + if schemaOwner != self { + return engineObjects{}, fmt.Errorf("%w: schema %q is owned by %s, this engine runs as %s", ErrForeignObject, SchemaName, schemaOwner, self) + } + if relkind == nil { + return engineObjects{schemaExists: true}, nil + } + if *relkind != relkindOrdinaryTable { + return engineObjects{}, fmt.Errorf("%w: %s is a relation of kind %q, not a table", ErrForeignObject, tableIdent, *relkind) + } + if *tableOwner != self { + return engineObjects{}, fmt.Errorf("%w: %s is owned by %s, this engine runs as %s", ErrForeignObject, tableIdent, *tableOwner, self) + } + return engineObjects{schemaExists: true, tableExists: true}, nil +} diff --git a/pkg/checkpoint/errors.go b/pkg/checkpoint/errors.go new file mode 100644 index 0000000..9368480 --- /dev/null +++ b/pkg/checkpoint/errors.go @@ -0,0 +1,104 @@ +package checkpoint + +import ( + "errors" + "fmt" + + "github.com/jackc/pgx/v5/pgconn" +) + +// ErrNotFound reports that the store holds no row for the target. It is a +// positive answer from the database — the read completed and found nothing +// — and the only Load outcome from which a caller may start fresh (ST-2). +var ErrNotFound = errors.New("checkpoint: no row for target") + +// ErrTableMissing reports that the checkpoint table does not exist in the +// database: Ensure has never run there. It is distinct from ErrNotFound — +// a database with no table has no checkpoints to read, but that is not a +// positive "no row for this target" and resume must not start fresh from it +// — and it still carries the server's error for callers that want it. +var ErrTableMissing = errors.New("checkpoint: checkpoint table does not exist; Ensure has not run") + +// ErrForeignObject reports that the engine schema or the checkpoint table +// exists but is not the engine's: another role owns it, or the name is +// taken by a relation that is not a plain table. Ensure refuses rather +// than adopt it, because whatever hangs off a foreign object — a trigger, +// a view — would run with the engine's privileges on every Save. +var ErrForeignObject = errors.New("checkpoint: engine object exists but is not the engine's") + +// ErrInvalidCheckpoint reports a Checkpoint the store refuses to persist: +// a missing identifier, an empty fingerprint, or an unknown phase. +var ErrInvalidCheckpoint = errors.New("checkpoint: invalid checkpoint") + +// ErrInvalidOptions reports Options the store cannot run under. +var ErrInvalidOptions = errors.New("checkpoint: invalid options") + +// ErrInvariantViolation reports a state the store's own invariants rule out: +// a row the database returned that no Save could have written. +var ErrInvariantViolation = errors.New("checkpoint: invariant violation") + +// Mismatch names which identity field of a stored row disagrees with the +// run asking for it. +type Mismatch uint8 + +const ( + // MismatchFormat means the row was written in another row format: an + // engine whose checkpoint columns mean something else. + MismatchFormat Mismatch = iota + 1 + // MismatchSource means the row was written for a source table whose + // introspected model differs from the current one. + MismatchSource + // MismatchTarget means the row was written for a different statement: + // its after-schema model differs from the current one. + MismatchTarget +) + +// String returns the stable mismatch name. +func (m Mismatch) String() string { + switch m { + case MismatchFormat: + return "format_version" + case MismatchSource: + return "source_fingerprint" + case MismatchTarget: + return "target_fingerprint" + default: + return fmt.Sprintf("Mismatch(%d)", m) + } +} + +// IncompatibleError reports a stored row the current run must not resume +// from or write over: it belongs to another row format or another statement +// (ST-2). Mismatch names the first disagreeing field, Have is its stored +// value and Want the current run's; Stored is the whole identity of the row +// as it was read, which a caller hands back to Delete so that a fresh start +// removes exactly the row it was shown and no later one. +type IncompatibleError struct { + Schema string + Table string + Mismatch Mismatch + Have string + Want string + Stored Identity +} + +// Error names the target, the disagreeing field, and both values. +func (e *IncompatibleError) Error() string { + return fmt.Sprintf("checkpoint for %s.%s is incompatible: stored %s %q, this run has %q", + e.Schema, e.Table, e.Mismatch, e.Have, e.Want) +} + +// undefinedTable is the SQLSTATE the server reports when the checkpoint +// table does not exist. +const undefinedTable = "42P01" + +// missingTable turns the server's undefined_table into ErrTableMissing, +// keeping the server's error reachable through errors.As; any other error +// is returned as it came. +func missingTable(err error) error { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == undefinedTable { + return fmt.Errorf("%w: %w", ErrTableMissing, err) + } + return err +} diff --git a/pkg/checkpoint/load.go b/pkg/checkpoint/load.go new file mode 100644 index 0000000..a25e348 --- /dev/null +++ b/pkg/checkpoint/load.go @@ -0,0 +1,210 @@ +package checkpoint + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/dbconn" + "github.com/block/pg-sprite/pkg/decode" +) + +// Load reads the target's checkpoint for a run identified by want and tells +// its four outcomes apart (ST-2): a row carrying want's format and +// fingerprints is returned; no row is ErrNotFound, the one outcome a caller +// may start fresh from; a row carrying another format version or another +// statement's fingerprints is an IncompatibleError and must not be resumed +// from or written over; a read that did not complete is retried through +// transient errors under the store's bounded attempts and, once they are +// spent, returned as an error that is none of the above — a blip never +// reads as "no checkpoint". A database where Ensure has never run has no +// table to read, which is ErrTableMissing: also none of the above, so a +// status reader can report "no checkpoint table" without resume mistaking +// it for ErrNotFound. +func (s *Store) Load(ctx context.Context, schema, table string, want Fingerprints) (Checkpoint, error) { + return s.load(ctx, s.pool, schema, table, want) +} + +func (s *Store) load(ctx context.Context, q dbconn.RowQuerier, schema, table string, want Fingerprints) (Checkpoint, error) { + // INV: ST-2 + var stored row + read := func(ctx context.Context) error { + var err error + stored, err = readRow(ctx, q, schema, table) + return err + } + if err := retryTransient(ctx, s.opts.LoadAttempts, s.opts.LoadBackoff, read, s.opts.Sleep); err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return Checkpoint{}, fmt.Errorf("%w: %s.%s", ErrNotFound, schema, table) + } + return Checkpoint{}, fmt.Errorf("load checkpoint for %s.%s: %w", schema, table, missingTable(err)) + } + if incompatible := stored.incompatibility(want); incompatible != nil { + return Checkpoint{}, incompatible + } + cp, err := stored.checkpoint() + if err != nil { + return Checkpoint{}, fmt.Errorf("load checkpoint for %s.%s: %w", schema, table, err) + } + return cp, nil +} + +// SQLSTATE codes a backend reports when the server ends its session from +// outside the session — pg_terminate_backend, a shutdown, or a restart still +// in recovery — which is how a failover looks from the client. A write +// interrupted this way has an ambiguous outcome and must not be retried, +// which is why dbconn.Retryable does not list them; Load is a read, so +// repeating it is safe and these count as transient for it alone. +const ( + codeAdminShutdown = "57P01" + codeCrashShutdown = "57P02" + codeCannotConnectNow = "57P03" +) + +// readRetryable reports whether a failed read of the checkpoint row may be +// repeated: anything the engine treats as transient, plus a session the +// server ended from outside it. +func readRetryable(err error) bool { + if dbconn.Retryable(err) { + return true + } + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) { + return false + } + switch pgErr.Code { + case codeAdminShutdown, codeCrashShutdown, codeCannotConnectNow: + return true + } + return false +} + +// retryTransient runs attempt up to attempts times, waiting between tries +// with a linearly growing backoff, and retries only errors readRetryable +// classifies as transient. Any other error, including no row, returns at +// once; the context's end wins over a pending wait. +func retryTransient(ctx context.Context, attempts int, backoff time.Duration, attempt func(context.Context) error, sleep SleepFunc) error { + var last error + for i := 1; i <= attempts; i++ { + last = attempt(ctx) + if last == nil { + return nil + } + if !readRetryable(last) { + return last + } + if i == attempts { + break + } + if err := sleep(ctx, backoff*time.Duration(i)); err != nil { + return fmt.Errorf("%w (last read error: %w)", err, last) + } + } + return fmt.Errorf("retries exhausted after %d attempts: %w", attempts, last) +} + +// row is one checkpoint row as the database returns it, before it is +// checked against the run and decoded into a Checkpoint. +type row struct { + schema, table string + formatVersion int32 + shadowTable string + slotName string + publicationName string + watermark pgtype.Int8 + lastAppliedLSN string + sourceFingerprint string + targetFingerprint string + phase string + updatedAt time.Time +} + +// loadSQL reads the target's row. The LSN comes back in its text form; pgx +// has no codec for pg_lsn and decode.ParseLSN is the one reader of it. +const loadSQL = `SELECT format_version, shadow_table, slot_name, publication_name, + watermark, last_applied_lsn::text, source_fingerprint, target_fingerprint, phase, updated_at +FROM ` + tableIdent + ` WHERE schema_name = $1 AND table_name = $2` + +// readRow performs one read of the target's row and returns pgx.ErrNoRows +// unwrapped when there is none, so callers can tell absence from failure. +func readRow(ctx context.Context, q dbconn.RowQuerier, schema, table string) (row, error) { + r := row{schema: schema, table: table} + err := q.QueryRow(ctx, loadSQL, schema, table).Scan( + &r.formatVersion, &r.shadowTable, &r.slotName, &r.publicationName, + &r.watermark, &r.lastAppliedLSN, &r.sourceFingerprint, &r.targetFingerprint, &r.phase, &r.updatedAt) + if err != nil { + return row{}, err + } + return r, nil +} + +// identity is the stored row's guarded fields. +func (r row) identity() Identity { + return Identity{ + FormatVersion: r.formatVersion, + Fingerprints: Fingerprints{Source: r.sourceFingerprint, Target: r.targetFingerprint}, + } +} + +// incompatibility reports the first identity field on which the stored row +// disagrees with the run, format version first: a row in another format +// cannot be trusted to carry fingerprints that mean the same thing. It is +// nil for a row the run may resume from. +func (r row) incompatibility(want Fingerprints) *IncompatibleError { + return r.incompatibleWith(Identity{FormatVersion: FormatVersion, Fingerprints: want}) +} + +// incompatibleWith reports the first field on which the stored row's +// identity differs from want, in the same order; nil when they agree. +func (r row) incompatibleWith(want Identity) *IncompatibleError { + // INV: ST-2 + mismatch := &IncompatibleError{Schema: r.schema, Table: r.table, Stored: r.identity()} + switch { + case r.formatVersion != want.FormatVersion: + mismatch.Mismatch, mismatch.Have, mismatch.Want = MismatchFormat, fmt.Sprint(r.formatVersion), fmt.Sprint(want.FormatVersion) + case r.sourceFingerprint != want.Fingerprints.Source: + mismatch.Mismatch, mismatch.Have, mismatch.Want = MismatchSource, r.sourceFingerprint, want.Fingerprints.Source + case r.targetFingerprint != want.Fingerprints.Target: + mismatch.Mismatch, mismatch.Have, mismatch.Want = MismatchTarget, r.targetFingerprint, want.Fingerprints.Target + default: + return nil + } + return mismatch +} + +// checkpoint decodes a compatible row. A value Save could not have written +// — an LSN that does not parse, a phase name String never produces — is an +// invariant violation, not a resume point. +func (r row) checkpoint() (Checkpoint, error) { + lsn, err := decode.ParseLSN(r.lastAppliedLSN) + if err != nil { + return Checkpoint{}, fmt.Errorf("%w: %w", ErrInvariantViolation, err) + } + phase, err := parsePhase(r.phase) + if err != nil { + return Checkpoint{}, fmt.Errorf("%w: %w", ErrInvariantViolation, err) + } + var watermark copier.Watermark + if r.watermark.Valid { + watermark = copier.NewWatermark(r.watermark.Int64) + } + return Checkpoint{ + Schema: r.schema, + Table: r.table, + ShadowTable: r.shadowTable, + SlotName: r.slotName, + PublicationName: r.publicationName, + Watermark: watermark, + LastAppliedLSN: lsn, + SourceFingerprint: r.sourceFingerprint, + TargetFingerprint: r.targetFingerprint, + Phase: phase, + UpdatedAt: r.updatedAt, + }, nil +} diff --git a/pkg/checkpoint/load_integration_test.go b/pkg/checkpoint/load_integration_test.go new file mode 100644 index 0000000..5fa9587 --- /dev/null +++ b/pkg/checkpoint/load_integration_test.go @@ -0,0 +1,101 @@ +package checkpoint_test + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/checkpoint" + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/dbconn" +) + +// A target with no row is ErrNotFound: a completed read that found nothing, +// the one outcome a run may start fresh from. +func TestLoadReportsNotFoundForATargetWithNoRow(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + _, err := f.store.Load(t.Context(), "app", "orders", ordersCheckpoint().Fingerprints()) + assert.ErrorIs(t, err, checkpoint.ErrNotFound) + var incompatible *checkpoint.IncompatibleError + assert.False(t, errors.As(err, &incompatible)) +} + +// A row written for another statement — here a different after-schema +// digest — is the typed mismatch, never a Checkpoint and never ErrNotFound. +func TestLoadReportsAnotherStatementsRowAsIncompatible(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + require.NoError(t, f.store.Save(t.Context(), f.ordersLock(t), ordersCheckpoint())) + + _, err := f.store.Load(t.Context(), "app", "orders", checkpoint.Fingerprints{Source: "src-a", Target: "tgt-after-edit"}) + var incompatible *checkpoint.IncompatibleError + require.ErrorAs(t, err, &incompatible) + assert.Equal(t, checkpoint.MismatchTarget, incompatible.Mismatch) + assert.Equal(t, "tgt-a", incompatible.Have) + assert.Equal(t, "tgt-after-edit", incompatible.Want) + assert.NotErrorIs(t, err, checkpoint.ErrNotFound) +} + +// A row written in another row format is incompatible before its +// fingerprints are consulted, even though they match. +func TestLoadReportsAnotherFormatVersionAsIncompatible(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + require.NoError(t, f.store.Save(t.Context(), f.ordersLock(t), ordersCheckpoint())) + _, err := f.pool.Exec(t.Context(), "UPDATE pgsprite.pgsprite_checkpoint SET format_version = format_version + 1") + require.NoError(t, err) + + _, err = f.store.Load(t.Context(), "app", "orders", ordersCheckpoint().Fingerprints()) + var incompatible *checkpoint.IncompatibleError + require.ErrorAs(t, err, &incompatible) + assert.Equal(t, checkpoint.MismatchFormat, incompatible.Mismatch) + assert.Equal(t, "2", incompatible.Have) + assert.Equal(t, "1", incompatible.Want) +} + +// A backend terminated from outside the session — what a failover looks +// like from the client — is a transient read error: Load retries on a fresh +// connection and returns the row, and the blip is never mistaken for a +// missing checkpoint (ST-2). The pool is held to one connection so the +// terminated one is the one the first read must use. +func TestLoadRetriesAcrossATerminatedBackend(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + require.NoError(t, f.store.Save(t.Context(), f.ordersLock(t), ordersCheckpoint())) + + single, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: f.url, MaxConns: 1, MinConns: 1}) + require.NoError(t, err) + t.Cleanup(single.Close) + waits := 0 + store, err := checkpoint.NewStore(single, checkpoint.Options{ + LoadAttempts: 3, + LoadBackoff: time.Millisecond, + Sleep: func(context.Context, time.Duration) error { + waits++ + return nil + }, + }) + require.NoError(t, err) + + var pid int + require.NoError(t, single.QueryRow(t.Context(), "SELECT pg_backend_pid()").Scan(&pid)) + var terminated bool + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT pg_terminate_backend($1)", pid).Scan(&terminated)) + require.True(t, terminated) + const goneDeadline = 10 * time.Second + require.Eventually(t, func() bool { + var alive bool + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE pid = $1)", pid).Scan(&alive)) + return !alive + }, goneDeadline, 20*time.Millisecond, "the terminated backend should leave pg_stat_activity") + + got, err := store.Load(t.Context(), "app", "orders", ordersCheckpoint().Fingerprints()) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(1000), got.Watermark) + assert.Equal(t, 1, waits, "exactly one read failed on the dead connection before the retry succeeded") +} diff --git a/pkg/checkpoint/load_test.go b/pkg/checkpoint/load_test.go new file mode 100644 index 0000000..1a15529 --- /dev/null +++ b/pkg/checkpoint/load_test.go @@ -0,0 +1,343 @@ +package checkpoint + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgtype" + "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/decode" +) + +// recordingSleep stands in for the wait between attempts and records every +// backoff it was asked for. +type recordingSleep struct{ waits []time.Duration } + +func (s *recordingSleep) sleep(_ context.Context, d time.Duration) error { + s.waits = append(s.waits, d) + return nil +} + +// A session the server ended from outside it is transient for the +// checkpoint read alone: the engine-wide classifier keeps those codes +// terminal, because a write they interrupt has an ambiguous outcome. +func TestReadRetryableWidensTheEngineClassifierForServerEndedSessions(t *testing.T) { + for _, tc := range []struct { + name string + err error + read bool + }{ + {"admin_shutdown (terminated backend)", &pgconn.PgError{Code: "57P01"}, true}, + {"crash_shutdown", &pgconn.PgError{Code: "57P02"}, true}, + {"cannot_connect_now (server in recovery)", &pgconn.PgError{Code: "57P03"}, true}, + {"database_dropped", &pgconn.PgError{Code: "57P04"}, false}, + {"query_canceled", &pgconn.PgError{Code: "57014"}, false}, + {"undefined_table", &pgconn.PgError{Code: "42P01"}, false}, + {"plain error", errors.New("boom"), false}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.read, readRetryable(tc.err), "read classification") + assert.False(t, dbconn.Retryable(tc.err), "the engine-wide classifier must not retry a write through this") + }) + } + for _, tc := range []struct { + name string + code string + }{ + {"lock_not_available", "55P03"}, + {"connection_exception", "08006"}, + } { + t.Run(tc.name+" stays transient for reads too", func(t *testing.T) { + err := &pgconn.PgError{Code: tc.code} + assert.True(t, dbconn.Retryable(err)) + assert.True(t, readRetryable(err)) + }) + } +} + +// A transient error is retried with a linearly growing backoff until the +// read succeeds; the attempt that succeeds is not followed by a wait. +func TestRetryTransientRetriesThroughTransientErrors(t *testing.T) { + sleep := &recordingSleep{} + calls := 0 + err := retryTransient(t.Context(), 5, 10*time.Millisecond, func(context.Context) error { + calls++ + if calls < 3 { + return &pgconn.PgError{Code: "57P01"} + } + return nil + }, sleep.sleep) + require.NoError(t, err) + assert.Equal(t, 3, calls) + assert.Equal(t, []time.Duration{10 * time.Millisecond, 20 * time.Millisecond}, sleep.waits) +} + +// No row is not a transient error: the read completed. It returns after one +// attempt, unwrapped, so Load can turn it into ErrNotFound. +func TestRetryTransientReturnsNoRowsAtOnce(t *testing.T) { + sleep := &recordingSleep{} + calls := 0 + err := retryTransient(t.Context(), 5, 10*time.Millisecond, func(context.Context) error { + calls++ + return pgx.ErrNoRows + }, sleep.sleep) + assert.ErrorIs(t, err, pgx.ErrNoRows) + assert.Equal(t, 1, calls) + assert.Empty(t, sleep.waits) +} + +// A permanent error — the table is missing — is not retried either. +func TestRetryTransientReturnsPermanentErrorAtOnce(t *testing.T) { + sleep := &recordingSleep{} + calls := 0 + err := retryTransient(t.Context(), 5, 10*time.Millisecond, func(context.Context) error { + calls++ + return &pgconn.PgError{Code: "42P01"} + }, sleep.sleep) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr) + assert.Equal(t, "42P01", pgErr.Code) + assert.Equal(t, 1, calls) + assert.Empty(t, sleep.waits) +} + +// Exhausting the attempts returns the last transient error, still +// classifiable, after attempts-1 waits. +func TestRetryTransientExhaustsItsAttempts(t *testing.T) { + sleep := &recordingSleep{} + calls := 0 + err := retryTransient(t.Context(), 3, time.Millisecond, func(context.Context) error { + calls++ + return &pgconn.PgError{Code: "08006"} + }, sleep.sleep) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr) + assert.Equal(t, "08006", pgErr.Code) + assert.Equal(t, 3, calls) + assert.Equal(t, []time.Duration{time.Millisecond, 2 * time.Millisecond}, sleep.waits) + assert.NotErrorIs(t, err, pgx.ErrNoRows) +} + +// A context that ends during a wait stops the retries with the context's +// error, carrying the last read error for the operator. +func TestRetryTransientStopsWhenTheContextEnds(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + calls := 0 + err := retryTransient(ctx, 5, time.Millisecond, func(context.Context) error { + calls++ + return &pgconn.PgError{Code: "57P01"} + }, func(ctx context.Context, _ time.Duration) error { + cancel() + return ctx.Err() + }) + assert.ErrorIs(t, err, context.Canceled) + var pgErr *pgconn.PgError + assert.ErrorAs(t, err, &pgErr) + assert.Equal(t, 1, calls) +} + +// fakeRow answers one Scan from a fixed value list or with an error. +type fakeRow struct { + values []any + err error +} + +func (r fakeRow) Scan(dest ...any) error { + if r.err != nil { + return r.err + } + if len(dest) != len(r.values) { + return errors.New("fakeRow: column count mismatch") + } + for i, d := range dest { + switch d := d.(type) { + case *int32: + *d = r.values[i].(int32) + case *string: + *d = r.values[i].(string) + case *pgtype.Int8: + *d = r.values[i].(pgtype.Int8) + case *time.Time: + *d = r.values[i].(time.Time) + default: + return errors.New("fakeRow: unsupported destination") + } + } + return nil +} + +// fakeQuerier serves a scripted sequence of rows, one per QueryRow call. +type fakeQuerier struct { + rows []fakeRow + calls int +} + +func (q *fakeQuerier) QueryRow(context.Context, string, ...any) pgx.Row { + row := q.rows[q.calls] + q.calls++ + return row +} + +var ( + stampedAt = time.Date(2026, time.October, 1, 12, 0, 0, 0, time.UTC) + wantFP = Fingerprints{Source: "src-a", Target: "tgt-a"} +) + +// storedRow is the row the fakes serve: a copy stopped at key 1000 with +// changes applied through 16/B374D848. +func storedRow() fakeRow { + return fakeRow{values: []any{ + FormatVersion, "_pgsprite_orders_new", "pgsprite_0a1b2c3d", "pgsprite_0a1b2c3d", + pgtype.Int8{Int64: 1000, Valid: true}, "16/B374D848", wantFP.Source, wantFP.Target, "copying", stampedAt, + }} +} + +func newTestStore(t *testing.T, sleep SleepFunc) *Store { + t.Helper() + return &Store{opts: Options{Clock: fixedClock{stampedAt}, LoadAttempts: 3, LoadBackoff: time.Millisecond, Sleep: sleep}} +} + +// Two terminated-backend errors followed by a good read resume from the +// stored row; every column comes back typed. +func TestLoadRetriesATerminatedBackendAndDecodesTheRow(t *testing.T) { + sleep := &recordingSleep{} + q := &fakeQuerier{rows: []fakeRow{ + {err: &pgconn.PgError{Code: "57P01"}}, + {err: &pgconn.PgError{Code: "08006"}}, + storedRow(), + }} + cp, err := newTestStore(t, sleep.sleep).load(t.Context(), q, "app", "orders", wantFP) + require.NoError(t, err) + assert.Equal(t, 3, q.calls) + assert.Len(t, sleep.waits, 2) + assert.Equal(t, Checkpoint{ + Schema: "app", Table: "orders", ShadowTable: "_pgsprite_orders_new", + SlotName: "pgsprite_0a1b2c3d", PublicationName: "pgsprite_0a1b2c3d", + Watermark: copier.NewWatermark(1000), LastAppliedLSN: decode.LSN(0x00000016b374d848), + SourceFingerprint: "src-a", TargetFingerprint: "tgt-a", Phase: PhaseCopying, UpdatedAt: stampedAt, + }, cp) +} + +// A blip that outlasts the attempts is an error that is neither ErrNotFound +// nor IncompatibleError: the caller must not start fresh on it (ST-2). +func TestLoadExhaustedRetriesIsNeitherNotFoundNorIncompatible(t *testing.T) { + sleep := &recordingSleep{} + q := &fakeQuerier{rows: []fakeRow{ + {err: &pgconn.PgError{Code: "57P01"}}, + {err: &pgconn.PgError{Code: "57P01"}}, + {err: &pgconn.PgError{Code: "57P01"}}, + }} + _, err := newTestStore(t, sleep.sleep).load(t.Context(), q, "app", "orders", wantFP) + require.Error(t, err) + assert.NotErrorIs(t, err, ErrNotFound) + var incompatible *IncompatibleError + assert.False(t, errors.As(err, &incompatible)) + var pgErr *pgconn.PgError + assert.ErrorAs(t, err, &pgErr) + assert.Equal(t, 3, q.calls) +} + +// No row is ErrNotFound after a single read. +func TestLoadReportsNoRowAsNotFound(t *testing.T) { + sleep := &recordingSleep{} + q := &fakeQuerier{rows: []fakeRow{{err: pgx.ErrNoRows}}} + _, err := newTestStore(t, sleep.sleep).load(t.Context(), q, "app", "orders", wantFP) + assert.ErrorIs(t, err, ErrNotFound) + assert.Equal(t, 1, q.calls) + assert.Empty(t, sleep.waits) +} + +// The format version is judged before the fingerprints: a row in another +// format is incompatible even when its fingerprint columns happen to match, +// because their meaning is not trusted. Among the fingerprints, the source +// is judged before the target. +func TestRowIncompatibilityOrdersFormatThenSourceThenTarget(t *testing.T) { + base := row{schema: "app", table: "orders", formatVersion: FormatVersion, sourceFingerprint: "src-a", targetFingerprint: "tgt-a"} + + assert.Nil(t, base.incompatibility(wantFP)) + + other := base + other.formatVersion = FormatVersion + 1 + other.sourceFingerprint = "src-b" + assert.Equal(t, &IncompatibleError{Schema: "app", Table: "orders", Mismatch: MismatchFormat, Have: "2", Want: "1", + Stored: Identity{FormatVersion: 2, Fingerprints: Fingerprints{Source: "src-b", Target: "tgt-a"}}}, + other.incompatibility(wantFP)) + + other = base + other.sourceFingerprint = "src-b" + other.targetFingerprint = "tgt-b" + assert.Equal(t, &IncompatibleError{Schema: "app", Table: "orders", Mismatch: MismatchSource, Have: "src-b", Want: "src-a", + Stored: Identity{FormatVersion: 1, Fingerprints: Fingerprints{Source: "src-b", Target: "tgt-b"}}}, + other.incompatibility(wantFP)) + + other = base + other.targetFingerprint = "tgt-b" + assert.Equal(t, &IncompatibleError{Schema: "app", Table: "orders", Mismatch: MismatchTarget, Have: "tgt-b", Want: "tgt-a", + Stored: Identity{FormatVersion: 1, Fingerprints: Fingerprints{Source: "src-a", Target: "tgt-b"}}}, + other.incompatibility(wantFP)) +} + +// Delete judges the row against the identity the caller was shown, so a +// row in the caller's format but written by another statement is reported +// with that statement's fingerprint; the identity Save writes for a +// checkpoint is the one Delete accepts for its own row. +func TestRowIncompatibleWithJudgesTheWholeIdentity(t *testing.T) { + stored := row{schema: "app", table: "orders", formatVersion: FormatVersion, sourceFingerprint: "src-a", targetFingerprint: "tgt-a"} + assert.Nil(t, stored.incompatibleWith(Identity{FormatVersion: FormatVersion, Fingerprints: wantFP})) + + cp := Checkpoint{SourceFingerprint: "src-a", TargetFingerprint: "tgt-a"} + assert.Nil(t, stored.incompatibleWith(cp.Identity())) + + shown := Identity{FormatVersion: FormatVersion, Fingerprints: Fingerprints{Source: "src-a", Target: "tgt-old"}} + assert.Equal(t, &IncompatibleError{Schema: "app", Table: "orders", Mismatch: MismatchTarget, Have: "tgt-a", Want: "tgt-old", + Stored: stored.identity()}, stored.incompatibleWith(shown)) + + shown.FormatVersion = FormatVersion + 1 + assert.Equal(t, MismatchFormat, stored.incompatibleWith(shown).Mismatch) +} + +// Load judges compatibility on the row it read, so a different statement's +// row surfaces as a typed IncompatibleError, not as a Checkpoint. +func TestLoadReportsAnotherStatementsRowAsIncompatible(t *testing.T) { + q := &fakeQuerier{rows: []fakeRow{storedRow()}} + _, err := newTestStore(t, (&recordingSleep{}).sleep).load(t.Context(), q, "app", "orders", Fingerprints{Source: "src-a", Target: "tgt-other"}) + var incompatible *IncompatibleError + require.ErrorAs(t, err, &incompatible) + assert.Equal(t, MismatchTarget, incompatible.Mismatch) + assert.Equal(t, "tgt-a", incompatible.Have) + assert.Equal(t, "tgt-other", incompatible.Want) + assert.NotErrorIs(t, err, ErrNotFound) +} + +// A NULL watermark decodes to the zero Watermark: nothing has landed. +func TestRowCheckpointDecodesNullWatermarkAsNothingLanded(t *testing.T) { + r := row{formatVersion: FormatVersion, lastAppliedLSN: "0/0", phase: "copying"} + cp, err := r.checkpoint() + require.NoError(t, err) + assert.False(t, cp.Watermark.Valid()) + assert.Equal(t, decode.LSN(0), cp.LastAppliedLSN) +} + +// A value Save cannot have written is an invariant violation, not a resume +// point: a phase name String never produces, or an LSN that does not parse. +func TestRowCheckpointRefusesValuesSaveCannotHaveWritten(t *testing.T) { + bad := row{formatVersion: FormatVersion, lastAppliedLSN: "0/0", phase: "paused"} + _, err := bad.checkpoint() + assert.ErrorIs(t, err, ErrInvariantViolation) + + bad = row{formatVersion: FormatVersion, lastAppliedLSN: "not-an-lsn", phase: "copying"} + _, err = bad.checkpoint() + assert.ErrorIs(t, err, ErrInvariantViolation) +} + +type fixedClock struct{ now time.Time } + +func (c fixedClock) Now() time.Time { return c.now } diff --git a/pkg/checkpoint/lock.go b/pkg/checkpoint/lock.go new file mode 100644 index 0000000..6f4670e --- /dev/null +++ b/pkg/checkpoint/lock.go @@ -0,0 +1,56 @@ +package checkpoint + +import ( + "context" + "errors" + "fmt" + + "github.com/block/pg-sprite/pkg/dbconn" +) + +// requireTableLock refuses a checkpoint write without the per-table lock +// session that keeps a second engine instance off the same target: the +// session must exist, be for schema.table, and not have reported loss. It +// runs before any connection is opened, so a refusal costs nothing. +func requireTableLock(lock *dbconn.TableLockSession, schema, table string) error { + // INV: LK-1 + if lock == nil { + return fmt.Errorf("%w (LK-1): checkpoint writes for %s.%s require a table lock session", ErrInvariantViolation, schema, table) + } + held := lock.Lock() + if held.Schema() != schema || held.Table() != table { + return fmt.Errorf("%w (LK-1): table lock is for %s.%s, checkpoint is for %s.%s", ErrInvariantViolation, held.Schema(), held.Table(), schema, table) + } + if err := lock.Err(); err != nil { + return fmt.Errorf("%w (LK-1): table lock was lost before the checkpoint write: %w", ErrInvariantViolation, err) + } + return nil +} + +// confirmTableLock asks the server, on the connection about to write, +// whether the lock session's own backend holds the target. A server that +// answers "no one" or "another backend" has contradicted the session the +// store trusts, and that is the store's invariant violation; a lookup that +// got no answer is the connection's error, reported as such. +func confirmTableLock(ctx context.Context, conn dbconn.AdvisoryLockHolder, lock *dbconn.TableLockSession) error { + err := lock.Confirm(ctx, conn) + if err == nil { + return nil + } + if lockDenied(err) { + // INV: LK-1 + return fmt.Errorf("%w (LK-1): %w", ErrInvariantViolation, err) + } + return fmt.Errorf("confirm the table lock from the checkpoint transaction: %w", err) +} + +// lockDenied reports whether a Confirm error is the server's word that the +// session does not hold the table, in either of the two forms dbconn gives +// it. +func lockDenied(err error) bool { + if errors.Is(err, dbconn.ErrTableLockNotHeld) { + return true + } + var heldElsewhere *dbconn.TableLockHeldError + return errors.As(err, &heldElsewhere) +} diff --git a/pkg/checkpoint/resume_integration_test.go b/pkg/checkpoint/resume_integration_test.go new file mode 100644 index 0000000..f396eca --- /dev/null +++ b/pkg/checkpoint/resume_integration_test.go @@ -0,0 +1,218 @@ +package checkpoint_test + +import ( + "context" + "fmt" + "math" + "strings" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/internal/testutil" + "github.com/block/pg-sprite/pkg/checkpoint" + "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" + "github.com/block/pg-sprite/pkg/statement" +) + +// copyFixture puts the real shadow builder and copier in front of the +// store, in a throwaway schema of the store's database, so the resume +// below goes through the same objects a run would checkpoint. +type copyFixture struct { + storeFixture + cfg dbconn.Config + schema string +} + +func newCopyFixture(t *testing.T) copyFixture { + t.Helper() + f := newStoreFixture(t) + f.ensure(t) + return copyFixture{storeFixture: f, cfg: dbconn.Config{URL: f.url}, schema: testutil.NewSchema(t, f.pool)} +} + +// exec runs SQL with %s standing for the fixture schema. +func (f copyFixture) exec(t *testing.T, sql string) { + t.Helper() + _, err := f.pool.Exec(t.Context(), strings.ReplaceAll(sql, "%s", f.schema)) + require.NoError(t, err) +} + +// prepare creates an orders table holding keys 1..rows, proves it, takes +// its table lock, and builds the shadow that drops the note column. +func (f copyFixture) prepare(t *testing.T, rows int64) (preflight.CopySwapTarget, *dbconn.TableLockSession, schemachange.BuiltShadow) { + t.Helper() + f.exec(t, ` + CREATE TABLE %s.orders ( + id bigint PRIMARY KEY, + qty integer NOT NULL, + note text + )`) + f.exec(t, fmt.Sprintf(` + INSERT INTO %%s.orders (id, qty, note) + SELECT n, n, 'order ' || n FROM generate_series(1, %d) AS n`, rows)) + + role, err := preflight.CheckPrivileges(t.Context(), f.pool, f.schema, "orders", preflight.Requirement{Tier: preflight.TierCopyAndSwap}) + require.NoError(t, err) + target, err := preflight.CheckCopySwapShape(t.Context(), f.pool, f.schema, "orders", role) + require.NoError(t, err) + + lock, err := dbconn.AcquireTableLock(t.Context(), f.cfg, f.schema, "orders") + require.NoError(t, err) + t.Cleanup(func() { assert.NoError(t, lock.Release(context.WithoutCancel(t.Context()))) }) + + alter, err := statement.ParseOne(fmt.Sprintf(`ALTER TABLE %s DROP COLUMN note`, pgx.Identifier{f.schema, "orders"}.Sanitize())) + require.NoError(t, err) + shadow, err := schemachange.BuildShadow(t.Context(), f.pool, lock, target, alter, schemachange.Options{}) + require.NoError(t, err) + return target, lock, shadow +} + +// hookedClock runs hook on its first reading. The copier reads its clock +// after the resume clear and the first claim, before the first chunk +// transaction, so a hook there meets the copy as a concurrent writer. +type hookedClock struct { + once sync.Once + hook func() +} + +func (c *hookedClock) Now() time.Time { + c.once.Do(c.hook) + return time.Now() +} + +// pinShadowKey inserts key into the shadow in a transaction left open at +// the copier's first clock reading, so the chunk holding key waits on the +// unique index and every other chunk lands around it; the run is then +// cancelled with that one chunk in flight, which is the mid-copy kill the +// checkpoint must survive. +func (f copyFixture) pinShadowKey(t *testing.T, shadow schemachange.BuiltShadow, key int64) (clock *hookedClock, pinned chan struct{}, release func()) { + t.Helper() + pinned = make(chan struct{}) + var tx pgx.Tx + var pinErr error + clock = &hookedClock{hook: func() { + defer close(pinned) + tx, pinErr = f.pool.Begin(t.Context()) + if pinErr != nil { + return + } + _, pinErr = tx.Exec(t.Context(), "INSERT INTO "+pgx.Identifier{f.schema, shadow.ShadowTable()}.Sanitize()+" (id, qty) VALUES ($1, 0)", key) + }} + var once sync.Once + release = func() { + once.Do(func() { + <-pinned + require.NoError(t, pinErr) + assert.NoError(t, tx.Rollback(context.WithoutCancel(t.Context()))) + }) + } + t.Cleanup(release) + return clock, pinned, release +} + +// A copy killed mid-way resumes from the checkpoint and converges: the +// first run's committed watermark is saved with the shadow's fingerprints, +// a new run loads it with the same fingerprints, hands it to the copier, +// and the copier clears above it and copies the rest — including a change +// made above the watermark while nothing ran — until the shadow equals the +// source. The checkpoint recorded only committed work: the resumed copier +// inserts exactly the rows above the watermark, including the pinned chunk +// that never landed. +func TestResumeFromCheckpointConvergesAfterAMidCopyKill(t *testing.T) { + f := newCopyFixture(t) + const rows = 2000 + target, lock, shadow := f.prepare(t, rows) + clock, pinned, release := f.pinShadowKey(t, shadow, 1050) + + first, err := copier.NewCopier(target, shadow, lock, copier.Watermark{}, copier.Options{ + Workers: 4, + LockTimeout: 30 * time.Second, + Chunker: copier.ChunkerOptions{InitialRows: 100, MaxRows: 100}, + Clock: clock, + }) + require.NoError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + results := make(chan error, 1) + var wg sync.WaitGroup + wg.Go(func() { results <- first.Run(ctx, f.pool) }) + t.Cleanup(wg.Wait) + + const pinDeadline = 15 * time.Second + select { + case <-pinned: + case <-time.After(pinDeadline): + t.Fatalf("the copier did not reach its first chunk within %s", pinDeadline) + } + const landedDeadline = 15 * time.Second + require.Eventually(t, func() bool { + pos := first.Position() + return pos.Cut == copier.NewWatermark(math.MaxInt64) && len(pos.InFlight) == 1 + }, landedDeadline, 20*time.Millisecond, "every chunk but the pinned one should land") + cancel() + const stopDeadline = 15 * time.Second + select { + case err := <-results: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(stopDeadline): + t.Fatalf("copy did not stop within %s of cancellation", stopDeadline) + } + killed := first.Position() + require.Equal(t, copier.NewWatermark(1000), killed.Watermark, "the watermark stops below the pinned chunk") + + // The run checkpoints what it committed, keyed on the shadow it built. + require.NoError(t, f.store.Save(t.Context(), lock, checkpoint.Checkpoint{ + Schema: target.Schema(), + Table: target.Table(), + ShadowTable: shadow.ShadowTable(), + Watermark: killed.Watermark, + SourceFingerprint: shadow.SourceFingerprint(), + TargetFingerprint: shadow.TargetFingerprint(), + Phase: checkpoint.PhaseCopying, + })) + release() + + // While nothing runs, the source moves above the watermark. + f.exec(t, "UPDATE %s.orders SET qty = 999 WHERE id = 1500") + + // A new run resumes from the row, under the same fingerprints. + loaded, err := f.store.Load(t.Context(), target.Schema(), target.Table(), + checkpoint.Fingerprints{Source: shadow.SourceFingerprint(), Target: shadow.TargetFingerprint()}) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(1000), loaded.Watermark) + assert.Equal(t, shadow.ShadowTable(), loaded.ShadowTable) + assert.Equal(t, checkpoint.PhaseCopying, loaded.Phase) + + resumed, err := copier.NewCopier(target, shadow, lock, loaded.Watermark, copier.Options{Workers: 2}) + require.NoError(t, err) + require.NoError(t, resumed.Run(t.Context(), f.pool)) + testutil.AssertConverged(t, f.pool, + testutil.RelationRef{Schema: f.schema, Table: shadow.SourceTable()}, + testutil.RelationRef{Schema: f.schema, Table: shadow.ShadowTable()}, + testutil.ConvergeOptions{IgnoreColumns: []string{"note"}}) + assert.Equal(t, int64(rows-1000), resumed.Position().RowsInserted, "everything above the watermark was cleared and copied again") + + // The completed copy checkpoints the complete watermark over the same row. + require.NoError(t, f.store.Save(t.Context(), lock, checkpoint.Checkpoint{ + Schema: target.Schema(), + Table: target.Table(), + ShadowTable: shadow.ShadowTable(), + Watermark: resumed.Position().Watermark, + SourceFingerprint: shadow.SourceFingerprint(), + TargetFingerprint: shadow.TargetFingerprint(), + Phase: checkpoint.PhaseVerifying, + })) + final, err := f.store.Load(t.Context(), target.Schema(), target.Table(), loaded.Fingerprints()) + require.NoError(t, err) + assert.True(t, final.Watermark.Complete()) + assert.Equal(t, checkpoint.PhaseVerifying, final.Phase) + assert.Equal(t, int64(1), f.rowCount(t)) +} diff --git a/pkg/checkpoint/store.go b/pkg/checkpoint/store.go new file mode 100644 index 0000000..003020f --- /dev/null +++ b/pkg/checkpoint/store.go @@ -0,0 +1,256 @@ +package checkpoint + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/jackc/pgx/v5/pgxpool" + + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/dbconn" + "github.com/block/pg-sprite/pkg/progress" +) + +const ( + // FormatVersion is the row format Save writes and Load accepts. A row + // carrying another value was written by an engine whose columns mean + // something else, and is incompatible whatever its fingerprints say. + FormatVersion int32 = 1 + // SchemaName is the engine-owned schema the checkpoint table lives in. + // It is the engine's, not the target's, so one table holds every + // target's row (D3) and no target schema is written to. + SchemaName = "pgsprite" + // TableName is the checkpoint table. + TableName = "pgsprite_checkpoint" + // DefaultLoadAttempts bounds how many times Load reads through a + // transient error before it gives up. + DefaultLoadAttempts = 5 + // DefaultLoadBackoff is the first wait between Load's attempts; later + // waits grow linearly. + DefaultLoadBackoff = 200 * time.Millisecond +) + +// SleepFunc waits for d or until ctx is done, returning ctx's error in the +// latter case. Tests inject one that records instead of waiting. +type SleepFunc func(ctx context.Context, d time.Duration) error + +// Options bounds the store. Zero values take the defaults: the wall clock, +// DefaultLoadAttempts, DefaultLoadBackoff, and a sleep that waits. +type Options struct { + // Clock stamps UpdatedAt on every Save. + Clock progress.Clock + // LoadAttempts bounds how many reads Load makes through transient + // errors (ST-2). + LoadAttempts int + // LoadBackoff is the wait after the first failed read; the n-th wait is + // n times it. + LoadBackoff time.Duration + // Sleep performs the waits between Load's attempts. + Sleep SleepFunc +} + +func (o Options) withDefaults() Options { + if o.Clock == nil { + o.Clock = progress.WallClock{} + } + if o.LoadAttempts == 0 { + o.LoadAttempts = DefaultLoadAttempts + } + if o.LoadBackoff == 0 { + o.LoadBackoff = DefaultLoadBackoff + } + if o.Sleep == nil { + o.Sleep = sleep + } + return o +} + +// validate runs after withDefaults, so every field is set. +func (o Options) validate() error { + if o.LoadAttempts < 1 { + return fmt.Errorf("%w: load attempts %d is below one", ErrInvalidOptions, o.LoadAttempts) + } + if o.LoadBackoff < 0 { + return fmt.Errorf("%w: load backoff %s is negative", ErrInvalidOptions, o.LoadBackoff) + } + return nil +} + +// sleep is the default SleepFunc. +func sleep(ctx context.Context, d time.Duration) error { + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +// Store persists one Checkpoint per target in the target database (D3). It +// runs over a pkg/dbconn pool, so every statement it issues is already +// under the session's lock_timeout and statement_timeout. +type Store struct { + pool *pgxpool.Pool + opts Options +} + +// NewStore prepares a store over pool. It refuses options that cannot bound +// a read. +func NewStore(pool *pgxpool.Pool, opts Options) (*Store, error) { + if pool == nil { + return nil, fmt.Errorf("%w: pool is nil", ErrInvalidOptions) + } + opts = opts.withDefaults() + if err := opts.validate(); err != nil { + return nil, err + } + return &Store{pool: pool, opts: opts}, nil +} + +// saveSQL is the one statement Save runs: insert the row, or update the +// existing row for the same target — but only when that row carries this +// run's format version and fingerprints. A row that carries another's is +// left untouched and the statement reports zero rows, which Save turns +// into IncompatibleError (ST-1, ST-2). +const saveSQL = "INSERT INTO " + tableIdent + ` ( + schema_name, table_name, format_version, shadow_table, slot_name, publication_name, + watermark, last_applied_lsn, source_fingerprint, target_fingerprint, phase, updated_at +) VALUES ($1, $2, $3, $4, $5, $6, $7, $8::pg_lsn, $9, $10, $11, $12) +ON CONFLICT (schema_name, table_name) DO UPDATE SET + shadow_table = EXCLUDED.shadow_table, + slot_name = EXCLUDED.slot_name, + publication_name = EXCLUDED.publication_name, + watermark = EXCLUDED.watermark, + last_applied_lsn = EXCLUDED.last_applied_lsn, + phase = EXCLUDED.phase, + updated_at = EXCLUDED.updated_at +WHERE ` + tableIdent + `.format_version = EXCLUDED.format_version + AND ` + tableIdent + `.source_fingerprint = EXCLUDED.source_fingerprint + AND ` + tableIdent + `.target_fingerprint = EXCLUDED.target_fingerprint` + +// Save writes cp as the target's one row, atomically: a reader sees the +// previous record or this one, never a mix (ST-1). It refuses, with +// IncompatibleError, to write over a row that another statement or another +// row format owns; a run that means to start fresh calls Delete first +// (ST-2). The write runs under the target's table lock session and +// confirms, from its own transaction, that the session still holds the +// table (LK-1): a run whose lock went to another engine cannot stamp its +// stale state over the newer engine's row. UpdatedAt is taken from the +// store's clock, not from cp. +func (s *Store) Save(ctx context.Context, lock *dbconn.TableLockSession, cp Checkpoint) error { + // INV: ST-1 + if err := cp.validate(); err != nil { + return err + } + if err := requireTableLock(lock, cp.Schema, cp.Table); err != nil { + return err + } + ctx, unbind := lock.Bind(ctx) + defer unbind() + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin save of checkpoint for %s.%s: %w", cp.Schema, cp.Table, err) + } + defer func() { + // Redundant safety closer: see Ensure. + _ = tx.Rollback(context.WithoutCancel(ctx)) + }() + if err := confirmTableLock(ctx, tx, lock); err != nil { + return fmt.Errorf("save checkpoint for %s.%s: %w", cp.Schema, cp.Table, err) + } + tag, err := tx.Exec(ctx, saveSQL, + cp.Schema, cp.Table, FormatVersion, cp.ShadowTable, cp.SlotName, cp.PublicationName, + watermarkColumn(cp.Watermark), cp.LastAppliedLSN.String(), + cp.SourceFingerprint, cp.TargetFingerprint, cp.Phase.String(), s.opts.Clock.Now()) + if err != nil { + return fmt.Errorf("save checkpoint for %s.%s: %w", cp.Schema, cp.Table, missingTable(err)) + } + if tag.RowsAffected() == 0 { + // INV: ST-2 + // The conflicting row is locked by the statement above for the rest + // of this transaction, so the read sees exactly the row that refused + // the update. + stored, err := readRow(ctx, tx, cp.Schema, cp.Table) + if err != nil { + return fmt.Errorf("save checkpoint for %s.%s: refused by an existing row that could not be read back: %w", cp.Schema, cp.Table, err) + } + if incompatible := stored.incompatibility(cp.Fingerprints()); incompatible != nil { + return incompatible + } + return fmt.Errorf("%w: save of checkpoint for %s.%s updated no row although the existing row is compatible", ErrInvariantViolation, cp.Schema, cp.Table) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit save of checkpoint for %s.%s: %w", cp.Schema, cp.Table, err) + } + return nil +} + +// watermarkColumn maps the copier's frontier onto the nullable bigint +// column: NULL when nothing has landed. +func watermarkColumn(w copier.Watermark) pgtype.Int8 { + return pgtype.Int8{Int64: w.Value(), Valid: w.Valid()} +} + +// deleteSQL removes the target's row, but only the row the caller was +// shown: one carrying the format version and fingerprints it names. A row +// that has since been replaced by another statement's is left in place and +// the statement reports zero rows. +const deleteSQL = "DELETE FROM " + tableIdent + ` WHERE schema_name = $1 AND table_name = $2 + AND format_version = $3 AND source_fingerprint = $4 AND target_fingerprint = $5` + +// Delete removes the target's row so a fresh run can Save its own. It is +// the one deliberate step between an IncompatibleError and a fresh start, +// and it removes only the row with the Identity the caller was shown — +// IncompatibleError.Stored, or a loaded Checkpoint's Identity — so a row +// another engine wrote in the meantime survives and comes back as a new +// IncompatibleError (ST-2). It runs under the target's table lock like +// Save (LK-1), and it is idempotent: a target with no row is already in +// the state Delete produces. +func (s *Store) Delete(ctx context.Context, lock *dbconn.TableLockSession, schema, table string, stored Identity) error { + if err := requireTableLock(lock, schema, table); err != nil { + return err + } + ctx, unbind := lock.Bind(ctx) + defer unbind() + tx, err := s.pool.Begin(ctx) + if err != nil { + return fmt.Errorf("begin delete of checkpoint for %s.%s: %w", schema, table, err) + } + defer func() { + // Redundant safety closer: see Ensure. + _ = tx.Rollback(context.WithoutCancel(ctx)) + }() + if err := confirmTableLock(ctx, tx, lock); err != nil { + return fmt.Errorf("delete checkpoint for %s.%s: %w", schema, table, err) + } + tag, err := tx.Exec(ctx, deleteSQL, schema, table, stored.FormatVersion, stored.Fingerprints.Source, stored.Fingerprints.Target) + if err != nil { + return fmt.Errorf("delete checkpoint for %s.%s: %w", schema, table, missingTable(err)) + } + if tag.RowsAffected() == 0 { + // INV: ST-2 + // Either there is no row, which is the state Delete produces, or + // the row is not the one the caller was shown. + current, err := readRow(ctx, tx, schema, table) + if errors.Is(err, pgx.ErrNoRows) { + return nil + } + if err != nil { + return fmt.Errorf("delete checkpoint for %s.%s: the row that refused the delete could not be read back: %w", schema, table, err) + } + if incompatible := current.incompatibleWith(stored); incompatible != nil { + return incompatible + } + return fmt.Errorf("%w: delete of checkpoint for %s.%s removed no row although the existing row matches", ErrInvariantViolation, schema, table) + } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("commit delete of checkpoint for %s.%s: %w", schema, table, err) + } + return nil +} diff --git a/pkg/checkpoint/store_integration_test.go b/pkg/checkpoint/store_integration_test.go new file mode 100644 index 0000000..9128640 --- /dev/null +++ b/pkg/checkpoint/store_integration_test.go @@ -0,0 +1,505 @@ +package checkpoint_test + +import ( + "context" + "errors" + "net/url" + "sync" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/internal/testutil" + "github.com/block/pg-sprite/pkg/checkpoint" + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/dbconn" + "github.com/block/pg-sprite/pkg/decode" +) + +// storeFixture is a store over a throwaway database of its own: the +// checkpoint table is one per database, so a fresh database is what makes +// "first use" observable. +type storeFixture struct { + url string + pool *pgxpool.Pool + store *checkpoint.Store + clock *fixedClock +} + +type fixedClock struct{ now time.Time } + +func (c *fixedClock) Now() time.Time { return c.now } + +var stampedAt = time.Date(2026, time.October, 1, 12, 0, 0, 0, time.UTC) + +func newStoreFixture(t *testing.T) storeFixture { + t.Helper() + url := testutil.NewDatabase(t, testutil.StartPostgres(t)) + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: url}) + require.NoError(t, err) + t.Cleanup(pool.Close) + clock := &fixedClock{now: stampedAt} + store, err := checkpoint.NewStore(pool, checkpoint.Options{Clock: clock}) + require.NoError(t, err) + return storeFixture{url: url, pool: pool, store: store, clock: clock} +} + +// ensure creates the table and asserts it is there. +func (f storeFixture) ensure(t *testing.T) { + t.Helper() + require.NoError(t, f.store.Ensure(t.Context())) + assert.True(t, f.tableExists(t)) +} + +func (f storeFixture) tableExists(t *testing.T) bool { + t.Helper() + var exists bool + require.NoError(t, f.pool.QueryRow(t.Context(), + "SELECT to_regclass($1) IS NOT NULL", checkpoint.SchemaName+"."+checkpoint.TableName).Scan(&exists)) + return exists +} + +// lock takes the per-table lock session every checkpoint write requires +// and releases it when the test ends. +func (f storeFixture) lock(t *testing.T, schema, table string) *dbconn.TableLockSession { + t.Helper() + lock, err := dbconn.AcquireTableLock(t.Context(), dbconn.Config{URL: f.url}, schema, table) + require.NoError(t, err) + t.Cleanup(func() { + // A test that deliberately loses the lock has already seen Release's + // invariant error through Err; a clean test releases cleanly. + if lock.Err() == nil { + assert.NoError(t, lock.Release(context.WithoutCancel(t.Context()))) + } + }) + return lock +} + +// ordersLock is the lock session for the app.orders target every test here +// checkpoints. +func (f storeFixture) ordersLock(t *testing.T) *dbconn.TableLockSession { + t.Helper() + return f.lock(t, "app", "orders") +} + +func (f storeFixture) rowCount(t *testing.T) int64 { + t.Helper() + var n int64 + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT count(*) FROM pgsprite.pgsprite_checkpoint").Scan(&n)) + return n +} + +// exec runs SQL as the fixture's superuser. +func (f storeFixture) exec(t *testing.T, sql string) { + t.Helper() + _, err := f.pool.Exec(t.Context(), sql) + require.NoError(t, err) +} + +// rolePassword is the password every throwaway login role here is created +// with. +const rolePassword = "checkpoint-test-password" + +// roleURL is the fixture database's URL with role as the user. +func (f storeFixture) roleURL(t *testing.T, role string) string { + t.Helper() + u, err := url.Parse(f.url) + require.NoError(t, err, "parse database URL") + u.User = url.UserPassword(role, rolePassword) + return u.String() +} + +// engineRole creates a login role with no privilege beyond CONNECT — the +// least an engine role can be — and a store over a pool connected as it. +func (f storeFixture) engineRole(t *testing.T) (string, *checkpoint.Store) { + t.Helper() + role := testutil.NewRole(t, f.pool, "LOGIN PASSWORD '"+rolePassword+"'") + pool, err := dbconn.NewPool(t.Context(), dbconn.Config{URL: f.roleURL(t, role)}) + require.NoError(t, err, "connect as %s", role) + t.Cleanup(pool.Close) + store, err := checkpoint.NewStore(pool, checkpoint.Options{Clock: f.clock}) + require.NoError(t, err) + return role, store +} + +// ordersCheckpoint is the record a copy of app.orders stopped at key 1000 +// would save: a quiesced run, so no slot and no publication. +func ordersCheckpoint() checkpoint.Checkpoint { + return checkpoint.Checkpoint{ + Schema: "app", + Table: "orders", + ShadowTable: "_pgsprite_orders_new", + Watermark: copier.NewWatermark(1000), + LastAppliedLSN: decode.LSN(0x00000016b374d848), + SourceFingerprint: "src-a", + TargetFingerprint: "tgt-a", + Phase: checkpoint.PhaseCopying, + } +} + +// Ensure creates the engine schema and the table on first use and is a +// no-op afterwards; nothing else is needed before the first Save. +func TestEnsureCreatesTheTableOnceAndIsIdempotent(t *testing.T) { + f := newStoreFixture(t) + assert.False(t, f.tableExists(t), "a fresh database has no checkpoint table") + f.ensure(t) + f.ensure(t) + assert.Equal(t, int64(0), f.rowCount(t)) +} + +// A second engine's Ensure that arrives while the first engine's create is +// still uncommitted waits on the engine's advisory key, then finds the +// schema committed and succeeds. Without the key its CREATE SCHEMA would +// race the uncommitted one into a duplicate-key failure in pg_namespace. +func TestEnsureWaitsForAConcurrentFirstCreate(t *testing.T) { + f := newStoreFixture(t) + first, err := f.pool.Begin(t.Context()) + require.NoError(t, err) + t.Cleanup(func() { _ = first.Rollback(context.WithoutCancel(t.Context())) }) + _, err = first.Exec(t.Context(), "SELECT pg_advisory_xact_lock($1, hashtext(quote_ident($2) || '.' || quote_ident($3)))", + dbconn.TableLockClassID, checkpoint.SchemaName, checkpoint.TableName) + require.NoError(t, err) + _, err = first.Exec(t.Context(), `CREATE SCHEMA "pgsprite"`) + require.NoError(t, err) + + second := make(chan error, 1) + var wg sync.WaitGroup + wg.Go(func() { second <- f.store.Ensure(t.Context()) }) + t.Cleanup(wg.Wait) + const waitingDeadline = 5 * time.Second + require.Eventually(t, func() bool { + var waiting bool + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT EXISTS (SELECT 1 FROM pg_locks WHERE NOT granted)").Scan(&waiting)) + return waiting + }, waitingDeadline, 10*time.Millisecond, "the second Ensure should be waiting behind the first create") + require.NoError(t, first.Commit(t.Context())) + const ensureDeadline = 5 * time.Second + select { + case err := <-second: + require.NoError(t, err) + case <-time.After(ensureDeadline): + t.Fatalf("the second Ensure did not finish within %s of the first commit", ensureDeadline) + } + assert.True(t, f.tableExists(t)) +} + +// A schema and table the engine role owns but did not create itself — a +// DBA pre-provisioned them so the engine never needs CREATE on the +// database — are accepted as they are: Ensure creates nothing, and Save +// lands. The engine role here has LOGIN and nothing else, so the CREATE +// SCHEMA that the privilege check would refuse is never issued. +func TestEnsureAcceptsAPreProvisionedTableWithoutDatabaseCreate(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + engine, engineStore := f.engineRole(t) + f.exec(t, `ALTER SCHEMA "pgsprite" OWNER TO `+pgx.Identifier{engine}.Sanitize()) + f.exec(t, `ALTER TABLE "pgsprite"."pgsprite_checkpoint" OWNER TO `+pgx.Identifier{engine}.Sanitize()) + var canCreate bool + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT has_database_privilege($1, current_database(), 'CREATE')", engine).Scan(&canCreate)) + require.False(t, canCreate, "the engine role must lack database CREATE for this test to mean anything") + + require.NoError(t, engineStore.Ensure(t.Context())) + lock, err := dbconn.AcquireTableLock(t.Context(), dbconn.Config{URL: f.roleURL(t, engine)}, "app", "orders") + require.NoError(t, err) + t.Cleanup(func() { assert.NoError(t, lock.Release(context.WithoutCancel(t.Context()))) }) + require.NoError(t, engineStore.Save(t.Context(), lock, ordersCheckpoint())) + assert.Equal(t, int64(1), f.rowCount(t)) +} + +// An engine schema another role owns is refused, typed, rather than +// adopted: whatever that role hangs off the schema or its table would run +// with the engine's privileges on every Save. +func TestEnsureRefusesASchemaAnotherRoleOwns(t *testing.T) { + f := newStoreFixture(t) + other := testutil.NewRole(t, f.pool, "NOLOGIN") + f.exec(t, `CREATE SCHEMA "pgsprite" AUTHORIZATION `+pgx.Identifier{other}.Sanitize()) + + err := f.store.Ensure(t.Context()) + assert.ErrorIs(t, err, checkpoint.ErrForeignObject) + assert.False(t, f.tableExists(t), "nothing is created inside a foreign schema") +} + +// A checkpoint table another role owns is refused the same way, even when +// the engine owns the schema around it. +func TestEnsureRefusesATableAnotherRoleOwns(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + other := testutil.NewRole(t, f.pool, "NOLOGIN") + f.exec(t, `ALTER TABLE "pgsprite"."pgsprite_checkpoint" OWNER TO `+pgx.Identifier{other}.Sanitize()) + + err := f.store.Ensure(t.Context()) + assert.ErrorIs(t, err, checkpoint.ErrForeignObject) +} + +// A view wearing the checkpoint table's name is not the checkpoint table, +// whoever owns it: only a plain table is accepted. +func TestEnsureRefusesAViewUnderTheTableName(t *testing.T) { + f := newStoreFixture(t) + f.exec(t, `CREATE SCHEMA "pgsprite"`) + f.exec(t, `CREATE VIEW "pgsprite"."pgsprite_checkpoint" AS SELECT 1 AS schema_name`) + + err := f.store.Ensure(t.Context()) + assert.ErrorIs(t, err, checkpoint.ErrForeignObject) +} + +// Save then Load round-trips every field: the watermark, the LSN through +// pg_lsn, the phase by name, and UpdatedAt from the store's clock rather +// than from the record handed in. +func TestSaveThenLoadRoundTripsTheRecord(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + cp := ordersCheckpoint() + cp.UpdatedAt = time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC) + require.NoError(t, f.store.Save(t.Context(), f.ordersLock(t), cp)) + + got, err := f.store.Load(t.Context(), "app", "orders", cp.Fingerprints()) + require.NoError(t, err) + assert.True(t, stampedAt.Equal(got.UpdatedAt), "UpdatedAt is the store clock's reading, got %s", got.UpdatedAt) + got.UpdatedAt = time.Time{} + cp.UpdatedAt = time.Time{} + assert.Equal(t, cp, got) +} + +// A second Save for the same target replaces the one row rather than adding +// another (ST-1): the table still holds one row and Load returns the later +// watermark, phase, and clock reading. +func TestSaveUpsertsTheOneRowPerTarget(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + lock := f.ordersLock(t) + first := ordersCheckpoint() + require.NoError(t, f.store.Save(t.Context(), lock, first)) + + later := stampedAt.Add(time.Minute) + f.clock.now = later + second := first + second.Watermark = copier.NewWatermark(5000) + second.LastAppliedLSN = decode.LSN(0x00000016b374ffff) + second.Phase = checkpoint.PhaseCatchingUp + second.SlotName = "pgsprite_0a1b2c3d" + second.PublicationName = "pgsprite_0a1b2c3d" + second.ShadowTable = "_pgsprite_orders_new2" + require.NoError(t, f.store.Save(t.Context(), lock, second)) + + assert.Equal(t, int64(1), f.rowCount(t)) + got, err := f.store.Load(t.Context(), "app", "orders", first.Fingerprints()) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(5000), got.Watermark) + assert.Equal(t, decode.LSN(0x00000016b374ffff), got.LastAppliedLSN) + assert.Equal(t, checkpoint.PhaseCatchingUp, got.Phase) + assert.Equal(t, "pgsprite_0a1b2c3d", got.SlotName) + assert.Equal(t, "pgsprite_0a1b2c3d", got.PublicationName) + assert.Equal(t, "_pgsprite_orders_new2", got.ShadowTable) + assert.True(t, later.Equal(got.UpdatedAt)) +} + +// Rows for different targets live side by side; a save for one never +// touches the other. +func TestSaveKeepsOneRowPerTarget(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + orders := ordersCheckpoint() + customers := ordersCheckpoint() + customers.Table = "customers" + customers.ShadowTable = "_pgsprite_customers_new" + customers.Watermark = copier.Watermark{} + require.NoError(t, f.store.Save(t.Context(), f.ordersLock(t), orders)) + require.NoError(t, f.store.Save(t.Context(), f.lock(t, "app", "customers"), customers)) + assert.Equal(t, int64(2), f.rowCount(t)) + + got, err := f.store.Load(t.Context(), "app", "customers", customers.Fingerprints()) + require.NoError(t, err) + assert.False(t, got.Watermark.Valid(), "a run that has landed nothing reads back the zero watermark") + assert.Equal(t, "_pgsprite_customers_new", got.ShadowTable) +} + +// A Save for a target whose row carries another statement's fingerprints is +// refused with the typed mismatch and leaves the stored row exactly as it +// was (ST-2): the statement's guard updated nothing. Delete is the explicit +// fresh start, after which the new statement's Save lands. +func TestSaveRefusesToOverwriteAnotherStatementsRow(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + lock := f.ordersLock(t) + original := ordersCheckpoint() + require.NoError(t, f.store.Save(t.Context(), lock, original)) + + other := ordersCheckpoint() + other.TargetFingerprint = "tgt-other" + other.Watermark = copier.NewWatermark(7) + other.Phase = checkpoint.PhaseVerifying + err := f.store.Save(t.Context(), lock, other) + var incompatible *checkpoint.IncompatibleError + require.ErrorAs(t, err, &incompatible) + assert.Equal(t, &checkpoint.IncompatibleError{ + Schema: "app", Table: "orders", Mismatch: checkpoint.MismatchTarget, Have: "tgt-a", Want: "tgt-other", + Stored: original.Identity(), + }, incompatible) + + kept, err := f.store.Load(t.Context(), "app", "orders", original.Fingerprints()) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(1000), kept.Watermark, "the refused save changed nothing") + assert.Equal(t, checkpoint.PhaseCopying, kept.Phase) + assert.Equal(t, int64(1), f.rowCount(t)) + + require.NoError(t, f.store.Delete(t.Context(), lock, "app", "orders", incompatible.Stored)) + require.NoError(t, f.store.Save(t.Context(), lock, other)) + got, err := f.store.Load(t.Context(), "app", "orders", other.Fingerprints()) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(7), got.Watermark) + assert.Equal(t, int64(1), f.rowCount(t)) +} + +// Delete removes only the row the caller was shown (ST-2): a row another +// statement wrote after the caller read its IncompatibleError is not the +// one it agreed to discard, so the row stays and the caller is shown the +// new one. +func TestDeleteRefusesARowTheCallerWasNotShown(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + lock := f.ordersLock(t) + shown := ordersCheckpoint() + require.NoError(t, f.store.Save(t.Context(), lock, shown)) + newer := ordersCheckpoint() + newer.TargetFingerprint = "tgt-newer" + newer.Watermark = copier.NewWatermark(42) + require.NoError(t, f.store.Delete(t.Context(), lock, "app", "orders", shown.Identity())) + require.NoError(t, f.store.Save(t.Context(), lock, newer)) + + err := f.store.Delete(t.Context(), lock, "app", "orders", shown.Identity()) + var incompatible *checkpoint.IncompatibleError + require.ErrorAs(t, err, &incompatible) + assert.Equal(t, &checkpoint.IncompatibleError{ + Schema: "app", Table: "orders", Mismatch: checkpoint.MismatchTarget, Have: "tgt-newer", Want: "tgt-a", + Stored: newer.Identity(), + }, incompatible) + kept, err := f.store.Load(t.Context(), "app", "orders", newer.Fingerprints()) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(42), kept.Watermark, "the newer row is untouched") +} + +// A checkpoint write without the target's lock session, or with a session +// for another table, is refused before any connection is opened (LK-1): +// the store never writes resume state for a table it does not hold. +func TestSaveAndDeleteRequireTheTargetsTableLock(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + cp := ordersCheckpoint() + + assert.ErrorIs(t, f.store.Save(t.Context(), nil, cp), checkpoint.ErrInvariantViolation) + assert.ErrorIs(t, f.store.Delete(t.Context(), nil, "app", "orders", cp.Identity()), checkpoint.ErrInvariantViolation) + + customers := f.lock(t, "app", "customers") + assert.ErrorIs(t, f.store.Save(t.Context(), customers, cp), checkpoint.ErrInvariantViolation) + assert.ErrorIs(t, f.store.Delete(t.Context(), customers, "app", "orders", cp.Identity()), checkpoint.ErrInvariantViolation) + assert.Equal(t, int64(0), f.rowCount(t)) +} + +// A Save whose lock session the server no longer counts as the holder is +// refused as the store's invariant violation (LK-1), and the newer holder's +// row is left as it was. The stale session's backend is terminated under a +// keepalive too long to notice during the test, so the session still +// believes it holds the table and only the confirmation from the save's own +// transaction stands between it and the row. This is the stale-writer case: +// same statement, same fingerprints, older state. +func TestSaveRefusesWhenAnotherBackendHoldsTheTable(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + stale, err := dbconn.AcquireTableLock(t.Context(), dbconn.Config{URL: f.url}, "app", "orders", dbconn.WithTableLockKeepalive(time.Hour)) + require.NoError(t, err) + t.Cleanup(func() { + assert.Error(t, stale.Release(context.WithoutCancel(t.Context())), "releasing a dead session reports it") + }) + var terminated bool + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT pg_terminate_backend($1)", stale.BackendPID()).Scan(&terminated)) + require.True(t, terminated) + const backendExitDeadline = 10 * time.Second + require.Eventually(t, func() bool { + var alive bool + require.NoError(t, f.pool.QueryRow(t.Context(), "SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE pid = $1)", stale.BackendPID()).Scan(&alive)) + return !alive + }, backendExitDeadline, 50*time.Millisecond, "the terminated backend should leave pg_stat_activity") + require.NoError(t, stale.Err(), "the stale session has not noticed its loss") + + takeover := f.ordersLock(t) + newer := ordersCheckpoint() + newer.Watermark = copier.NewWatermark(1800) + newer.Phase = checkpoint.PhaseVerifying + require.NoError(t, f.store.Save(t.Context(), takeover, newer)) + + err = f.store.Save(t.Context(), stale, ordersCheckpoint()) + assert.ErrorIs(t, err, checkpoint.ErrInvariantViolation) + var heldElsewhere *dbconn.TableLockHeldError + assert.ErrorAs(t, err, &heldElsewhere, "the server names the backend that holds the table now") + got, err := f.store.Load(t.Context(), "app", "orders", newer.Fingerprints()) + require.NoError(t, err) + assert.Equal(t, copier.NewWatermark(1800), got.Watermark, "the stale save changed nothing") + assert.Equal(t, checkpoint.PhaseVerifying, got.Phase) +} + +// Delete of a target with no row is not an error: the target is already in +// the state Delete produces. +func TestDeleteIsIdempotent(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + lock := f.ordersLock(t) + cp := ordersCheckpoint() + require.NoError(t, f.store.Delete(t.Context(), lock, "app", "orders", cp.Identity())) + require.NoError(t, f.store.Save(t.Context(), lock, cp)) + require.NoError(t, f.store.Delete(t.Context(), lock, "app", "orders", cp.Identity())) + require.NoError(t, f.store.Delete(t.Context(), lock, "app", "orders", cp.Identity())) + _, err := f.store.Load(t.Context(), "app", "orders", cp.Fingerprints()) + assert.ErrorIs(t, err, checkpoint.ErrNotFound) +} + +// An invalid checkpoint never reaches the database. +func TestSaveRefusesAnInvalidCheckpointBeforeTheDatabase(t *testing.T) { + f := newStoreFixture(t) + cp := ordersCheckpoint() + cp.TargetFingerprint = "" + err := f.store.Save(t.Context(), f.ordersLock(t), cp) + assert.ErrorIs(t, err, checkpoint.ErrInvalidCheckpoint) + var pgErr *pgconn.PgError + assert.False(t, errors.As(err, &pgErr), "the database was never asked") + assert.False(t, f.tableExists(t), "Save does not create the table; Ensure does") +} + +// Without Ensure the table is missing, and Save, Delete, and Load all report +// the typed ErrTableMissing — still carrying the server's undefined_table — +// rather than creating anything on their own; Load's answer is never +// ErrNotFound, so resume cannot start fresh over an unprovisioned database. +func TestWritesAndReadsWithoutEnsureReportTheMissingTable(t *testing.T) { + f := newStoreFixture(t) + lock := f.ordersLock(t) + cp := ordersCheckpoint() + for name, err := range map[string]error{ + "save": f.store.Save(t.Context(), lock, cp), + "delete": f.store.Delete(t.Context(), lock, "app", "orders", cp.Identity()), + } { + assert.ErrorIs(t, err, checkpoint.ErrTableMissing, name) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr, name) + assert.Equal(t, "42P01", pgErr.Code, name) + } + _, err := f.store.Load(t.Context(), "app", "orders", cp.Fingerprints()) + assert.ErrorIs(t, err, checkpoint.ErrTableMissing) + assert.NotErrorIs(t, err, checkpoint.ErrNotFound) + assert.False(t, f.tableExists(t)) +} + +// A cancelled context stops Save before any write. +func TestSaveHonoursACancelledContext(t *testing.T) { + f := newStoreFixture(t) + f.ensure(t) + lock := f.ordersLock(t) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + err := f.store.Save(ctx, lock, ordersCheckpoint()) + assert.ErrorIs(t, err, context.Canceled) + assert.Equal(t, int64(0), f.rowCount(t)) +} diff --git a/pkg/checkpoint/store_test.go b/pkg/checkpoint/store_test.go new file mode 100644 index 0000000..5a7cd5a --- /dev/null +++ b/pkg/checkpoint/store_test.go @@ -0,0 +1,77 @@ +package checkpoint + +import ( + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgtype" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/block/pg-sprite/pkg/copier" + "github.com/block/pg-sprite/pkg/progress" +) + +// Zero options take the defaults; a store cannot be built over no pool or +// under options that would never read. +func TestNewStoreOptions(t *testing.T) { + _, err := NewStore(nil, Options{}) + assert.ErrorIs(t, err, ErrInvalidOptions) + + opts := Options{}.withDefaults() + assert.Equal(t, progress.WallClock{}, opts.Clock) + assert.Equal(t, DefaultLoadAttempts, opts.LoadAttempts) + assert.Equal(t, DefaultLoadBackoff, opts.LoadBackoff) + assert.NotNil(t, opts.Sleep) + require.NoError(t, opts.validate()) + + assert.ErrorIs(t, Options{LoadAttempts: -1}.withDefaults().validate(), ErrInvalidOptions) + assert.ErrorIs(t, Options{LoadBackoff: -time.Second}.withDefaults().validate(), ErrInvalidOptions) +} + +// The table identifier is spelled as a constant so the statements can be +// constants too; it must stay what pgx would render for the two names. +func TestTableIdentMatchesPgxQuoting(t *testing.T) { + assert.Equal(t, pgx.Identifier{SchemaName, TableName}.Sanitize(), tableIdent) + assert.Equal(t, "CREATE SCHEMA "+pgx.Identifier{SchemaName}.Sanitize(), createSchemaSQL) + assert.True(t, strings.HasPrefix(createTableSQL, "CREATE TABLE "+tableIdent+" ("), createTableSQL) +} + +// Delete's statement matches the whole row identity, so a row that another +// statement wrote after the caller read its IncompatibleError is never the +// one removed (ST-2). +func TestDeleteSQLMatchesTheRowIdentity(t *testing.T) { + _, where, found := strings.Cut(deleteSQL, " WHERE ") + require.True(t, found, "the delete is conditional") + for _, column := range []string{"schema_name", "table_name", "format_version", "source_fingerprint", "target_fingerprint"} { + assert.Contains(t, where, column+" = $") + } +} + +// The copier's zero watermark is the NULL column, and a valid one carries +// its key; the complete watermark fits the bigint column. +func TestWatermarkColumn(t *testing.T) { + assert.Equal(t, pgtype.Int8{}, watermarkColumn(copier.Watermark{})) + assert.Equal(t, pgtype.Int8{Int64: 1000, Valid: true}, watermarkColumn(copier.NewWatermark(1000))) + assert.Equal(t, pgtype.Int8{Int64: 1<<63 - 1, Valid: true}, watermarkColumn(copier.NewWatermark(1<<63-1))) +} + +// Save's one statement upserts on the target key and updates only a row +// that carries this run's identity: every identity column is in the +// conflict guard and none is in the update list, so a Save can never move +// a row from one statement to another (ST-1, ST-2). +func TestSaveSQLGuardsTheRowIdentity(t *testing.T) { + assert.Contains(t, saveSQL, `ON CONFLICT (schema_name, table_name) DO UPDATE SET`) + _, guard, found := strings.Cut(saveSQL, "\nWHERE ") + require.True(t, found, "the upsert has a conflict guard") + for _, column := range []string{"format_version", "source_fingerprint", "target_fingerprint"} { + assert.Contains(t, guard, `"pgsprite"."pgsprite_checkpoint".`+column+" = EXCLUDED."+column) + } + update, _, _ := strings.Cut(saveSQL, "\nWHERE ") + _, update, _ = strings.Cut(update, "DO UPDATE SET") + for _, column := range []string{"format_version", "source_fingerprint", "target_fingerprint"} { + assert.NotContains(t, update, column+" = EXCLUDED."+column, "identity columns are never updated") + } +} diff --git a/pkg/decode/types.go b/pkg/decode/types.go index a9fb32a..cf49cb4 100644 --- a/pkg/decode/types.go +++ b/pkg/decode/types.go @@ -1,6 +1,10 @@ package decode -import "fmt" +import ( + "fmt" + "strconv" + "strings" +) // ChangeKind identifies a decoded row operation. type ChangeKind uint8 @@ -34,6 +38,25 @@ type LSN uint64 // String renders the PostgreSQL X/Y representation. func (l LSN) String() string { return fmt.Sprintf("%X/%X", uint64(l)>>32, uint64(l)&0xffffffff) } +// ParseLSN reads the PostgreSQL X/Y representation — the text form of a +// pg_lsn value — back into an LSN. Both halves are hexadecimal and each +// must fit in 32 bits; anything else is an error naming the input. +func ParseLSN(s string) (LSN, error) { + high, low, ok := strings.Cut(s, "/") + if !ok { + return 0, fmt.Errorf("parse LSN %q: want the X/Y form", s) + } + hi, err := strconv.ParseUint(high, 16, 32) + if err != nil { + return 0, fmt.Errorf("parse LSN %q: high half: %w", s, err) + } + lo, err := strconv.ParseUint(low, 16, 32) + if err != nil { + return 0, fmt.Errorf("parse LSN %q: low half: %w", s, err) + } + return LSN(hi<<32 | lo), nil +} + // Column is one decoded column. Present=false means pgoutput omitted an // unchanged TOAST value, so an applier must leave the target column untouched. type Column struct { diff --git a/pkg/decode/types_test.go b/pkg/decode/types_test.go index 2446125..d5d8932 100644 --- a/pkg/decode/types_test.go +++ b/pkg/decode/types_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestChangeKindString(t *testing.T) { @@ -13,6 +14,26 @@ func TestChangeKindString(t *testing.T) { assert.Equal(t, "ChangeKind(99)", ChangeKind(99).String()) } func TestLSNString(t *testing.T) { assert.Equal(t, "16/B374D848", LSN(0x00000016b374d848).String()) } + +// ParseLSN inverts String: the pg_lsn text form round-trips, including the +// zero LSN and a low half with leading zeros that String would not print. +func TestParseLSN(t *testing.T) { + cases := map[string]LSN{ + "16/B374D848": 0x00000016b374d848, + "0/0": 0, + "0/00000001": 1, + "FFFFFFFF/FFFFFFFF": 0xffffffffffffffff, + } + for text, want := range cases { + got, err := ParseLSN(text) + require.NoError(t, err, text) + assert.Equal(t, want, got, text) + } + for _, bad := range []string{"", "16", "16/", "/B374D848", "1/FFFFFFFFF", "100000000/0", "0x16/1", "16/B374D848/0"} { + _, err := ParseLSN(bad) + assert.Error(t, err, bad) + } +} func TestPresentColumns(t *testing.T) { e := ChangeEvent{Columns: []Column{{Name: "blob", Present: false}, {Name: "label", Value: "x", Present: true}}} assert.Equal(t, []Column{{Name: "label", Value: "x", Present: true}}, e.PresentColumns())