decode: stream the slot into change events with caller-confirmed feedback

.golangci.yml

@@ -38,7 +38,6 @@ linters:
             - "**/pkg/checksum/**"
             - "**/pkg/copier/**"
             - "**/pkg/applier/**"
-            - "**/pkg/decode/**"
             - "**/pkg/checkpoint/**"
             - "**/pkg/schemachange/**"
             - "**/pkg/internal/**"
@@ -59,6 +58,30 @@ linters:
             - github.com/block/pg-sprite/pkg/internal/chunksql
             - github.com/block/pg-sprite/pkg/statement
             - github.com/block/pg-sprite/pkg/schemadiff
+        # pkg/decode is the one core package admitted to import the pinned
+        # replication-protocol dependency (SAFETY.md's recorded decision on
+        # jackc/pglogrepl); every other core package is held to the list above.
+        decode:
+          files:
+            - "**/pkg/decode/**"
+            - "!$test"
+          allow:
+            - $gostd
+            - github.com/jackc/pgx/v5
+            - github.com/block/pg-sprite/pkg/dbconn
+            - github.com/block/pg-sprite/pkg/preflight
+            - github.com/block/pg-sprite/pkg/executor
+            - github.com/block/pg-sprite/pkg/progress
+            - github.com/block/pg-sprite/pkg/checksum
+            - github.com/block/pg-sprite/pkg/copier
+            - github.com/block/pg-sprite/pkg/applier
+            - github.com/block/pg-sprite/pkg/decode
+            - github.com/block/pg-sprite/pkg/checkpoint
+            - github.com/block/pg-sprite/pkg/schemachange
+            - github.com/block/pg-sprite/pkg/internal/chunksql
+            - github.com/block/pg-sprite/pkg/statement
+            - github.com/block/pg-sprite/pkg/schemadiff
+            - github.com/jackc/pglogrepl
     sloglint:
       static-msg: true
       key-naming-case: snake

SAFETY.md

@@ -17,16 +17,16 @@ The invariant registry (invariant IDs referenced below) lives in
 
 | Package | Core? | Status | Invariants enforced |
 | --- | --- | --- | --- |
-| `pkg/dbconn` — pool defaults, terminate-blockers, retries, RDS TLS, advisory table lock | ✅ core | table-lock primitive exists; wiring into executing modes planned | LK-1 primitive; LK-2 primitives; CO-9 (session hook and `LocalSearchPath`) |
+| `pkg/dbconn` — pool defaults, terminate-blockers, retries, RDS TLS, advisory table lock, replication connection | ✅ core | table-lock primitive exists; `ConnectReplication` dials one replication-mode connection with the pool's TLS, timeouts, and `BeforeConnect` hook; wiring into executing modes planned | LK-1 primitive; LK-2 primitives; CO-9 (session hook and `LocalSearchPath`) |
 | `pkg/preflight` — precondition verifier, refusals | ✅ core | exists; copy-and-swap target proof (shape and dependents) and cluster/volume environment check exist, not yet wired into a route | ST-6, RF-1..RF-5 |
 | `pkg/executor` — bounded optimistic attempt; native concurrent index build with invalid-index recovery; native sequence executor for the safer idioms | ✅ core | exists (Phase 1: attempt-under-budget; Phase 3.1: concurrent index build; Phase 3.2: sequence executor) | LK-2 (attempt bound + the CONCURRENTLY wait-policy exception), CO-9 (qualified proof reads), ST-9 (create owner verified, never repaired) |
 | `pkg/checksum` — chunk verifier, divergence policy, repair, continuous checker | ✅ core | the chunk `Verifier` exists (one read-only `REPEATABLE READ` transaction per chunk in which the chunk is cut and both digests read, so all three share a snapshot and the cut is bounded, every column cast to the shadow's type on both sides, the same guard as the copier — owner role, catalog-only `search_path`, `ACCESS SHARE` on both relations, lock confirmation, relation-OID check — and a `Report` of mismatched chunks up to the landed watermark); `Check` runs the pass under an explicit `DivergencePolicy` (zero value refused; `ParseDivergencePolicy` for configuration), aborts with the shadow untouched or recopies every differing chunk with the copier's chunk statement in one guarded transaction — every chunk's rows deleted before any are put back, so a unique value the source moved between two chunks lands — then rereads each chunk, returning the committed repairs with any refusal, and mints `CleanWatermark` and `VerifiedShadow` (private constructors) only from a pass that found nothing and repaired nothing; the pass is the progress tracker's `WorkSource` while it runs (chunks compared, rows hashed, chunks mismatched, chunks repaired, all from memory); the continuous checker is planned | CO-1, CO-2, CO-3, CO-9, LK-1, LK-2 |
 | `pkg/copier` — shadow-table chunked copy | ✅ core | contract types and the keyset `Chunker` exist (row-count chunks over the proven key, first chunk open below and last open above, a cut frontier for the applier's discard rule, time-targeted sizing) and the parallel `Copier` (one bounded never-overwriting insert per chunk under the table lock session, each in a guarded transaction — owner role, catalog-only `search_path`, `ACCESS SHARE` on both relations, lock confirmation, relation-OID check — frontier-ordered in-flight registry, a resume that first clears the shadow above the watermark, `Position.Classify` for the applier, and the copy step's `progress.WorkSource` — rows from committed chunks, the source's catalog row count, both tables' measured sizes; every cut and measurement in a bounded read-only transaction with the catalog alone on its `search_path`) exist | CO-4 (chunk coverage, copy SQL shape, in-flight registry), CO-9, LK-1, LK-2 (every statement on the caller's pool bounds itself), LK-3 |
 | `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 | `Buffer` (merge, drain against `copier.Position`, `Requeue` of a refused batch) and `Flusher` (one guarded read-write transaction per batch under the table lock session — owner role, catalog-only `search_path`, `ACCESS SHARE` on both relations, lock confirmation, relation-OID check — completing moved marker-bearing images first, then column-wise delete / upsert / present-columns UPDATE, with the delete-all-then-insert-all fallback on a unique violation and `ErrBatchDeferred` when the fallback is refused too) built; flush scheduling against the copier planned | CO-4, CO-5, CO-6, CO-8, LK-1, LK-2, 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/decode` — logical decoding, LSN/position accounting, per-column presence | ✅ core | exists (`OpenStream` decodes the slot with pgoutput on a dedicated replication connection proven to be on the target's database, into one `ChangeEvent` per committed row change: a text value or NULL is a present column, the unchanged-TOAST marker is an absent one, and an UPDATE that moved the key carries `OldKey`; a relation that is not the target, a tuple that does not line up with it, or a change outside a transaction fails closed, a changed table shape is `ErrSourceShapeChanged`, and TRUNCATE is `ErrUnsupportedChange`; `Delivered` moves on a commit or on a keepalive between transactions and never names a position a transaction not yet yielded in full committed at or below — positions order transactions by commit, so a change's own LSN can lie below one, and each `ChangeEvent` carries the `Delivered` it arrived with as the position confirmable while it is unapplied; `Confirm` is the only standby-status report that carries a position, refuses a regression or a position beyond `Delivered`, and records nothing the server was not told; keepalive replies carry only what the caller confirmed; `ServerWALEnd` is the server's own position for lag; the server ending replication with the slot intact is `ErrStreamEnded`, not a violation) | ST-4, CO-4, CO-8 |
 | `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 |
+| slot lifecycle (in `pkg/decode`) — create, drop, inspect, reap, lag ceiling | ✅ core | create, drop, and inspect exist (`CreateSlot` makes the single-table publication first and then the logical slot, named as preflight derived, on a dedicated replication connection from `pkg/dbconn` proven to be on the target's database, with an exported snapshot and the consistent point; `DropSlot` waits out a walsender holding the slot under the caller's context alone, never reports a cut-off wait as a drop, refuses a slot of the name another database owns, and is idempotent; `InspectSlot` reads `wal_status`, retained WAL, and the holder); the reaper and the lag ceiling are planned | 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) |
 | `pkg/verdict` — structured outcome contract, rendering, exit codes | ❌ periphery | exists (Phase 1) | — |
@@ -112,7 +112,8 @@ The short version — the full rules live in [docs/tcb-model.md](docs/tcb-model.
   Recorded decision: `jackc/pglogrepl` (pinned) is admitted to the core for `pkg/decode` because
   the streaming-replication protocol and `pgoutput` message decoding are load-bearing
   wire-protocol expertise, under the same rubric as the parser; it is confined to `pkg/decode`
-  and is not added to `go.mod` until that package's implementation lands.
+  (the depguard `decode` rule admits it there and nowhere else in the core) and is pinned to a
+  commit in `go.mod`, as the module publishes no tags.
   Recorded decision: the AWS SDK (`aws-sdk-go-v2`) is a test-harness-only dependency, confined
   behind the `ministack` build tag in `internal/testutil` — it never appears in the core, in
   `cmd/pg-sprite`, or in any ordinary build; a plain `go build ./...` / `go test ./...` never