checkpoint: persist the one resume row per target with a guarded upsert and a typed load

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) |

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

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 <engine>` and `ALTER TABLE pgsprite.pgsprite_checkpoint OWNER TO
+<engine>`. `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

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,20 +523,37 @@ 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
 
 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