Repository navigation
decode: stream the slot into change events with caller-confirmed feedback - #151
Conversation
Add the slot lifecycle to pkg/decode, the first use of the pinned jackc/pglogrepl dependency SAFETY.md records for this package. CreateSlot takes the copy-and-swap target preflight minted and creates both halves of the route's logical-decoding state under the derived name: the single-table publication on the caller's pool first, so a role that may create a slot but lacks CREATE on the database is refused with nothing to reap, then the logical pgoutput slot with EXPORT_SNAPSHOT on a dedicated replication connection. The connection is proven to be on the target's database (IDENTIFY_SYSTEM) before the slot exists anywhere; the pool is proven the same way before the publication. The returned Slot carries the exported snapshot name and the consistent point and keeps the replication connection open, because the snapshot lives only as long as that walsender's transaction; Close ends it and the slot persists. A logical slot of the derived name already in the target's database is the route's own earlier slot and is reported as *SlotExistsError for the caller to resume on or drop. A publication of the name that publishes anything but the target is ErrForeignDecodingState, never adopted. DropSlot drops only a logical slot of the pool's own database — any other slot of the name is ErrForeignDecodingState and a name outside the engine's pgsprite_<8hex> shape is ErrInvariantViolation — with DROP_REPLICATION_SLOT ... WAIT on its own replication connection, so a walsender still streaming is waited for under the caller's context alone; a wait the context ends is that error, never a report that the slot is gone. An absent slot is success, so the drop is idempotent. The publication goes with it. InspectSlot reads the slot's pg_replication_slots row: owning database, holder, wal_status as typed constants including the server's own 'lost' verdict for ST-4, restart and confirmed-flush positions, and the WAL the slot retains measured from restart_lsn to the write position. dbconn.ConnectReplication dials the replication-mode connection with the pool's TLS posture, connect timeout, startup parameters, and BeforeConnect hook, as a bare pgconn: a walsender accepts only the simple protocol, and replication commands are not bounded by statement_timeout at all. The depguard core rule gains a decode variant admitting pglogrepl there and nowhere else; pglogrepl is pinned to a commit as the module publishes no tags. Integration tests on a dedicated wal_level=logical server, PG14-18: the exported snapshot predates a row written after the slot and dies with the connection; a new slot has confirmed exactly its consistent point; the publication is exactly the target; a second create reports the route's own slot and reuses the publication; a role without CREATE on the database is refused before any slot exists; a replication connection or a pool on another database is refused; a quiesced target and the zero target are refused; drop is idempotent and removes the publication; drop waits while a START_REPLICATION holder streams and completes when it closes; a drop cut off by its context leaves slot and publication in place; a foreign database's slot and a physical slot of the name are refused and left; retained WAL grows with writes from an unmoving restart_lsn; wal_status reads 'lost' after the server discards the slot's WAL; a wider or FOR ALL TABLES publication is refused with no slot.
…back OpenStream decodes the route's slot with pgoutput on a dedicated replication connection proven to be on the target's database, from the consistent point or a checkpointed applied position, into one ChangeEvent per committed row change. A text value or NULL is a present column and the unchanged-TOAST marker an absent one; an UPDATE that moved the key carries OldKey under either admitted replica identity. A relation other than the target, a tuple that does not line up with it, or a change outside a transaction fails closed; a changed table shape is ErrSourceShapeChanged and TRUNCATE is ErrUnsupportedChange. The stream keeps Delivered — moved by a commit or by a keepalive between transactions, never past an unyielded change — apart from Confirmed, the one position it reports to the server. Confirm refuses a regression or a position beyond Delivered, and a keepalive reply carries only what the caller confirmed, so the slot's resume point moves on the caller's word alone.
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
morgo
left a comment
morgo
left a comment
There was a problem hiding this comment.
🤖 Automated adversarial review, posted on Morgan Tocker's behalf.
Approving. I reviewed this as the stack delta against kiran01bm/cs7-slot, not against main. I went looking specifically for a way to move the slot's confirmed position past something the caller never applied — the one failure here that destroys data irrecoverably, since the server cannot replay discarded WAL — and I could not construct one. All 16 checks green across PostgreSQL 14–18.
The position discipline is the part that had to be right, and it is. delivered rises in exactly two places, and both are safe for the reason the comments give:
- on
CommitMessage, toTransactionEndLSN, after every one of that transaction's changes has already been yielded; - on a keepalive, only when
!inTransaction.
The keepalive case is the one worth spelling out, because it rests on a server property the comment asserts without naming: a walsender's keepalive reports sentPtr, the position it has streamed to, not the server's WAL insert position. So ServerWALEnd can never name a position with an undelivered change below it — including during a large transaction, where protocol version 1 buffers until commit and sentPtr therefore stays put, making raiseDelivered a no-op. Holding the position inside a transaction is belt-and-braces on top of that, and correctly so.
Verified rather than assumed:
Confirm's bounds are the right two and they close the hole.lsn > s.deliveredis the one that matters, anddeliveredis initialised tofromrather than zero, so a caller cannot confirm above the start before any data arrives either.sendStatusreally does moveconfirmed_flush_lsndespite setting onlyWALWritePosition. pglogrepl fillsWALFlushPositionandWALApplyPositionfrom the write position when they are zero, so the wire message carries all three as the description says. Worth having checked, because logical slots advance on the flush position — had pglogrepl not defaulted,Confirmwould have been a silent no-op, and the integration test asserting the slot moves to exactly the confirmed position is what pins it.decodeColumnscannot turn an unchanged-TOAST marker into a value.'t'sets value and present,'n'sets present with a nil value,'u'falls through leavingPresentfalse, and every other data type including'b'is refused rather than defaulted. The column-count mismatch check in front of it means a stale relation cache cannot shift values by one position either.newRelationcannot leavekeyIndexdangling. It starts at-1, a missing PK column is refused before use, and the key column must carry the replica-identity flag — sodecodeKey's uncheckedtuple.Columns[rel.keyIndex]is safe for everyrelationthat exists.Next(ctx, 0)degrades safely rather than killing the stream. An already-expiredwaitCtxstill satisfies all three arms ofwaitElapsed(pgconn.Timeout,waitCtx.Err() == DeadlineExceeded,ctx.Err() == nil), so a caller whose wait is an unset config field gets an immediate empty progress delivery instead of a permanentlyfailedstream. That is the right direction and it is not obvious from readingNextalone.- Every
OpenStreamfailure path closes the connection viaerrors.Join(err, conn.Close(ctx)), including the database-identity refusal.
Non-blocking
Confirm advances s.confirmed before the send, so the field can outrun what the server was actually told.
s.confirmed = lsn
return s.sendStatus(ctx)If sendStatus fails, confirmed is already lsn and the stream is not failed, so the field now disagrees with its own doc comment — "Confirmed is the position last reported to the server as applied" — which is exactly what it is not.
Nothing is lost by this: confirmed is only read as the floor for later confirmations and as the value sent, any later successful send carries a position at or above the failed one, and the server never moves a slot backwards. The concrete cost is a legitimate caller getting refused. An applier that fails to confirm and then falls back to its last durable checkpoint — below the position whose send just failed — gets ErrInvariantViolation: confirm X below the Y already confirmed, naming a confirmation that never reached the server. That reads as a bug in the caller, and the remedy it suggests (don't go backwards) is the opposite of what the caller should do.
Setting s.confirmed only after sendStatus returns nil makes the field mean what it says and makes that retry work, at the cost of nothing — a failed send leaves the floor where it was, which is the conservative direction.
Related, smaller: the same sendStatus error is treated two ways. From handleKeepalive it propagates to Next, which latches it in fail and ends the stream permanently; from Confirm it is returned with the stream still live. Both are defensible in isolation, but a connection that cannot send a status update is in the same state either way, and one of the two callers decides that is terminal.
This PR is the first thing to produce Present: false, which makes the Value/Present ambiguity live. An unchanged-TOAST column and a SQL NULL both arrive with Value == nil; only Present separates them, and getting it wrong writes NULL over a value that was never touched — the precise corruption CO-8 exists to prevent. The producer side is right and the comment says so. The types predate this PR so the shape is not yours to change here, but since the applier is the next consumer and the compiler will not catch a switch on Value == nil, it may be worth Column growing an accessor that makes reading an absent value impossible rather than merely documented, before there is a second consumer to migrate.
|
🤖 Review of acf57cf (1/2): 1 blocking, 4 non-blocking. This is the adversarial correctness pass over the whole diff at The feedback path is sound where it matters most: nothing the stream sends can move the slot past a transaction the caller has not been handed. A keepalive reply carries only On invariants: ST-4 is extended, and with B1 open its new Enforced text states a property the stream does not have. CO-8 is extended to the decoder (presence carried; the TOAST and NULL mutants are killed). CO-4 is upheld ( BlockingB1. A change can arrive with an LSN below a position the stream already delivered and the caller confirmed, so the
Nothing is lost. The server replays by commit position, so a stream reopened from
Suggested fix:
Test case: the documented contract fails on acf57cf, and a replacement pins the real oneThe documented contract, as a test: func TestDeliveredCoversEveryChangeBelowIt(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
open, err := f.pool.Begin(t.Context())
require.NoError(t, err)
t.Cleanup(func() { _ = open.Rollback(context.WithoutCancel(t.Context())) })
_, err = open.Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s.ledger (id, note) VALUES (201, 'written first')`, f.schema))
require.NoError(t, err)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (202, 'committed first')`)
first := nextChange(t, stream)
awaitDelivered(t, stream, first.LSN+1)
confirmed := stream.Delivered()
require.NoError(t, stream.Confirm(t.Context(), confirmed))
require.NoError(t, open.Commit(t.Context()))
late := nextChange(t, stream)
assert.Greater(t, late.LSN, confirmed, "every change at or below Delivered has been yielded")
buf := applier.NewBuffer()
require.NoError(t, buf.Add(late))
oldest, _ := buf.OldestPending()
assert.NoError(t, stream.Confirm(t.Context(), min(stream.Delivered(), oldest)))
}
The contract the stream actually keeps, which passes on // A transaction that writes the target before another transaction commits,
// and commits after it, arrives after that commit with a change position
// below what the stream already delivered and the caller confirmed. The
// slot replays by commit position, so a stream reopened from the
// confirmation sees the change again; the position the caller may confirm
// while that change is unapplied is the Delivered it arrived with, never
// the change's own LSN.
func TestStreamDeliversAnInterleavedTransactionBelowTheConfirmedPosition(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
open, err := f.pool.Begin(t.Context())
require.NoError(t, err)
t.Cleanup(func() { _ = open.Rollback(context.WithoutCancel(t.Context())) })
_, err = open.Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s.ledger (id, note) VALUES (201, 'written first')`, f.schema))
require.NoError(t, err)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (202, 'committed first')`)
first := nextChange(t, stream)
require.EqualValues(t, 202, first.Key)
awaitDelivered(t, stream, first.LSN+1)
confirmed := stream.Delivered()
require.NoError(t, stream.Confirm(t.Context(), confirmed))
f.assertConfirmedFlushBecomes(t, slot, confirmed)
require.NoError(t, open.Commit(t.Context()))
late := nextChangeDelivery(t, stream) // nextChange, returning the whole Delivery
require.EqualValues(t, 201, late.Change.Key)
assert.Less(t, late.Change.LSN, confirmed, "the change was written below the confirmed position")
assert.GreaterOrEqual(t, late.Delivered, confirmed)
require.NoError(t, stream.Confirm(t.Context(), late.Delivered), "the delivery's own position is confirmable")
require.NoError(t, stream.Close(t.Context()))
replayed := nextChange(t, f.openStream(t, confirmed))
assert.Equal(t, *late.Change, replayed, "the slot replays the transaction that committed above it")
}
Non-blockingN1. An open stream that has not confirmed the server's send position holds a fast shutdown of the server open, until the stream closes. On shutdown, a logical walsender exits only once the client's reported flush position (its write position, while flush is zero) equals what the walsender has sent. Until then it keeps asking for replies. This stream answers each request with
This is latent until the orchestrator exists. It fires when a caller holds a stream open with
One trap for whoever fixes this in the stream rather than in the caller: do not report Probe (not a suite test): needs a server the test can stopRun against a dedicated Without confirms: With confirms: the stream ended after 52.98ms, and the server stopped. N2. A server that ends replication in an orderly way is reported as In the confirming variant of N1's probe, the walsender finished with N3. Nine refusals have no test that fails without them. Nine mutants survive the PR's tests. The proposed tests below pass on
The deadline mutant matters most. Without Test case: nine tests that pass on acf57cf and each kill a surviving mutantUnit tests (package // insertMessage encodes a pgoutput INSERT of a one-column row whose only
// column is the text key.
func insertMessage(relationID uint32, key string) []byte {
b := []byte{byte(pglogrepl.MessageTypeInsert)}
b = binary.BigEndian.AppendUint32(b, relationID)
b = append(b, 'N')
b = binary.BigEndian.AppendUint16(b, 1)
b = append(b, pglogrepl.TupleDataTypeText)
b = binary.BigEndian.AppendUint32(b, uint32(len(key)))
return append(b, key...)
}
// keyOnlyRelation is the stream's record of a one-column table keyed on id.
func keyOnlyRelation(id uint32) *relation {
return &relation{id: id, columns: []pglogrepl.RelationMessageColumn{{Flags: keyFlag, Name: "id"}}, keyIndex: 0}
}
// A row change the server sent outside BEGIN … COMMIT is refused rather
// than yielded: nothing would ever deliver a position covering it.
func TestStreamRefusesAChangeOutsideATransaction(t *testing.T) {
s := &Stream{delivered: 100, relation: keyOnlyRelation(7)}
d, yielded, err := s.handleWALData(walData(110, insertMessage(7, "101")))
require.ErrorIs(t, err, ErrInvariantViolation)
assert.False(t, yielded)
assert.Nil(t, d.Change)
}
// A row change naming a relation other than the one the stream described
// is refused rather than decoded against the target's columns (ST-3).
func TestStreamRefusesAChangeToAnotherRelation(t *testing.T) {
s := &Stream{delivered: 100, relation: keyOnlyRelation(7)}
_, _, err := s.handleWALData(walData(105, beginMessage(140)))
require.NoError(t, err)
d, yielded, err := s.handleWALData(walData(110, insertMessage(8, "101")))
require.ErrorIs(t, err, ErrInvariantViolation)
assert.False(t, yielded)
assert.Nil(t, d.Change)
}Integration tests (package // A stream is refused for a quiesced target, whose environment was never
// checked for decoding, and on a replication connection to another
// database, which would read another database's slot of the same name (ST-3).
func TestOpenStreamRefusesAQuiescedTargetAndAnotherDatabase(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
_, err := decode.OpenStream(t.Context(), f.cfg, f.mintTarget(t, f.pool, false), slot.ConsistentPoint())
require.ErrorIs(t, err, decode.ErrInvariantViolation)
other := dbconn.Config{URL: testutil.NewDatabase(t, f.serverURL)}
_, err = decode.OpenStream(t.Context(), other, f.target, slot.ConsistentPoint())
require.ErrorIs(t, err, decode.ErrInvariantViolation)
}
// A source renamed under the stream is no longer the target: its next
// change stops the stream rather than arriving under the old name.
func TestStreamStopsWhenTheSourceIsRenamed(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
f.exec(t, `ALTER TABLE %s.ledger RENAME TO ledger_moved`)
f.exec(t, `INSERT INTO %s.ledger_moved (id, note) VALUES (101, 'renamed')`)
require.ErrorIs(t, nextError(t, stream), decode.ErrInvariantViolation)
}
// A source whose key is no longer part of the replica identity cannot
// carry its key in an old tuple, so the stream stops at its description.
func TestStreamStopsWhenTheKeyLeavesTheReplicaIdentity(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
f.exec(t, `ALTER TABLE %s.ledger REPLICA IDENTITY NOTHING`)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (101, 'no identity')`)
require.ErrorIs(t, nextError(t, stream), decode.ErrInvariantViolation)
}
// The caller's own deadline ends the stream even when it falls inside the
// wait: only the wait running out is a quiet table.
func TestNextReturnsTheCallersDeadline(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond)
defer cancel()
until := time.Now().Add(streamDeadline)
for time.Now().Before(until) {
if _, err := stream.Next(ctx, 5*time.Second); err != nil {
require.ErrorIs(t, err, context.DeadlineExceeded)
return
}
}
require.FailNow(t, "Next kept yielding progress after the caller's deadline")
}
// A stopped stream refuses to confirm: the caller's position is no longer
// one the stream can vouch for.
func TestConfirmRefusesAfterTheStreamStopped(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
f.exec(t, `TRUNCATE %s.ledger`)
require.ErrorIs(t, nextError(t, stream), decode.ErrUnsupportedChange)
err := stream.Confirm(t.Context(), stream.Delivered())
require.ErrorIs(t, err, decode.ErrUnsupportedChange)
assert.Equal(t, decode.LSN(0), stream.Confirmed())
}
// A publication dropped under the stream fails the decoder on the server;
// the stream returns the server's error rather than a protocol violation.
func TestStreamReturnsTheDecodersError(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
_, err := f.pool.Exec(t.Context(), `DROP PUBLICATION `+slot.Name())
require.NoError(t, err)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (101, 'unpublished')`)
var pgErr *pgconn.PgError
require.True(t, errors.As(nextError(t, stream), &pgErr))
assert.Equal(t, "42704", pgErr.Code)
}All nine pass on N4. The "server never moves the slot backwards" assertion reads the slot before the server can have read the status update. The walsender applies a status update when it next reads the socket, which is why the file's own The verified list and the mutation table are in 2/2. This review was generated by Claude Code (claude-opus-5-5). |
|
🤖 Review of acf57cf (2/2): 0 blocking, 3 non-blocking. This comment covers the two lenses, OSS adoption and integration ease for SchemaBot and other importers, and then what I verified, including the mutation table. These are not correctness findings. For adoption, the stream is easy to pick up. The notes below are for the orchestrator that will drive this, and for the progress and resume surfaces an importer such as SchemaBot renders from it. Each is cheaper to shape now, before that caller exists. 1. The stream needs a second WAL sender, but preflight proves one. The description raises this as a decision to veto, and I would take the preflight change. 2. The server's position is read and then dropped, so a lag figure has to be projected. Every keepalive carries 3. A forwarded resume reports the position it asked for, not the one it got. When Verified
Mutants ran against
This review was generated by Claude Code (claude-opus-5-5). |
aparajon
left a comment
aparajon
left a comment
There was a problem hiding this comment.
🤖 Stamping with comments: 1 blocking, 7 non-blocking (see the review comments above).
This stamp was left by Claude Code (claude-opus-5-5).
…fusals
CreateSlot and DropSlot took a replication config and a pool as separate
inputs and tied them together by database name alone. A name identifies a
database only within one cluster, so a config pointing at a same-named
database on another cluster passed the check: CreateSlot created the slot
there, and DropSlot counted the other cluster's "no such slot" as a drop
while the pool's slot stayed, holding WAL, with its publication gone. Both
now prove the connection and the pool are sessions of one database on one
cluster — IDENTIFY_SYSTEM's system identifier against pg_control_system()
— before any slot command, and DropSlot re-reads the catalog after an
accepted drop and refuses to report success while the slot is still there.
The publication check counted tables in pg_publication_tables, which says
which tables and not which changes: a publication of the derived name
publishing inserts only, no truncate, a row filter, a column list, or a
whole schema was adopted as the route's own, and the changes it left out
never reached the stream. DropSlot skipped the check entirely and dropped
whatever publication wore the name. One predicate — every operation, no
root publishing, no filters, exactly one plain table membership, and on the
create path that table being the target — now gates both paths; DropSlot
refuses a publication that fails it and leaves it as found.
Refusals are typed so an orchestrator can route on them: *ForeignStateError
{Object, Name, Detail} wrapping ErrForeignDecodingState, and
*PublicationPrivilegeError naming the privilege and the database and
wrapping the server's error. A slot the server created but described
unusably is dropped before the error is returned, so no CreateSlot error
leaves a slot. SlotStatus.RetainedBytes becomes Retained{Bytes, Known} so
an unset restart_lsn is not read as zero bytes retained.
The derived name hashes the pgx-quoted database.schema.table instead of the
dotted join, so two tables whose dotted spellings coincide derive different
names.
Tests cover each: same-named database on a second cluster for both create
and drop, each narrower publication shape, another table's publication,
the foreign publication left by the drop, the context-cut create leaving
no slot, the typed errors, the name derivation, and the holder's PID in
the fixture. Docs: ST-3 and D11 state the server proof and the
publication-scoped drop.
…bm/cs7-pgoutput * origin/kiran01bm/cs7-slot: decode: prove the server, verify the publication's shape, type the refusals schemachange: re-derive the swapped-table proof from the catalog for the post-swap resume (#146) migrate: let only the caller's cancellation end the unknown-outcome test's lock wait (#145) applier: flush drained batches column-wise with the unique-move fallback (#149) # Conflicts: # SAFETY.md # docs/copy-and-swap-design.md # docs/invariants.md # pkg/decode/doc.go
…tream end Positions order transactions by their commit. pgoutput sends a transaction whole when it commits, so a change's own LSN can lie below a position the stream already delivered and the caller confirmed; the server replays by commit position, so nothing is lost, but the contract said otherwise and the applier's confirm bound was keyed on the change's LSN. Every ChangeEvent now carries the Delivered it arrived with, Buffer keys FirstLSN on it, and the Delivery, Confirm, package, ST-4, CO-5, SAFETY.md and design-doc text is restated in commit terms. Confirm sends first and records the position only when the server was told; a send that fails latches the stream, as it already did from a keepalive. sendStatus sets write, flush and apply explicitly. A server ending replication with the slot intact is ErrStreamEnded, not a violation. ServerWALEnd exposes the server's position for lag. Preflight now requires two free WAL senders for a copy-and-swap run: the slot's snapshot connection and the decoding stream hold one each. Tests: an interleaved transaction below the confirmed position composes with Buffer.OldestPending; Confirm after stop and after a failed send; the nine stream refusals (outside a transaction, another relation, quiesced database, renamed source, key leaving the identity, decoder error, the caller's deadline); the never-moves-backwards assertion waits for the walsender's flush position; preflight at max_wal_senders 1 and 2.
|
🤖 Adversarial review response — created by Kiran's code review agent (Amp, Claude Opus 4.6) — pull/151, follow-up commit Reviewed head was
Source: adversarial review comments on this PR; fixes in the follow-up commit. |
|
🤖 Delta review of 30f16d7 (1/2): 1 blocking, 3 non-blocking. This is the adversarial correctness pass over what changed since my review at Every earlier finding is fixed. The commit-ordering rework is right: On invariants: ST-4 is extended, and its Enforced text now matches what the stream does. N5 qualifies one sentence of the
BlockingB2. The doc says That is the doc's formula on This is blocking because it breaks the documented contract of a new exported accessor, and Suggested fix: take the position from keepalives only, and never report it below // ServerWALEnd is the walsender's position from its last keepalive, and
// never less than Delivered.
func (s *Stream) ServerWALEnd() LSN { return max(s.serverWALEnd, s.delivered) }Then delete line 271, and say in the doc, Tests that fail on 30f16d7 and pass with the fixUnit (package // For logical decoding the walsender writes the record's own position into
// XLogData's end field, so a change reports where it was written, not where
// the server is: only a keepalive moves ServerWALEnd, and it never reads
// below Delivered.
func TestServerWALEndIsNotMovedByAChange(t *testing.T) {
s := &Stream{delivered: 100, relation: keyOnlyRelation(7)}
_, _, err := s.handleKeepalive(t.Context(), pglogrepl.PrimaryKeepaliveMessage{ServerWALEnd: 200})
require.NoError(t, err)
_, _, err = s.handleWALData(walData(160, beginMessage(240)))
require.NoError(t, err)
_, _, err = s.handleWALData(walData(170, insertMessage(7, "101")))
require.NoError(t, err)
assert.Equal(t, LSN(200), s.ServerWALEnd())
_, _, err = s.handleWALData(walData(250, commitMessage(240, 250)))
require.NoError(t, err)
assert.Equal(t, LSN(250), s.ServerWALEnd(), "a commit past the last keepalive is the furthest position known")
}Integration (package // A transaction that wrote before another committed arrives after that
// commit. Its change carries the WAL position it was written at, which lies
// below what the stream already delivered, so the server's position the
// stream reports for lag must not follow it there.
func TestServerWALEndNeverTrailsDelivered(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
open, err := f.pool.Begin(t.Context())
require.NoError(t, err)
t.Cleanup(func() {
if err := open.Rollback(context.WithoutCancel(t.Context())); err != nil && !errors.Is(err, pgx.ErrTxClosed) {
t.Logf("roll back the open transaction: %v", err)
}
})
_, err = open.Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s.ledger (id, note) VALUES (201, 'written first')`, f.schema))
require.NoError(t, err)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (202, 'committed first')`)
first := nextChange(t, stream)
awaitDelivered(t, stream, first.LSN+1)
assert.GreaterOrEqual(t, stream.ServerWALEnd(), stream.Delivered(), "after a commit")
require.NoError(t, open.Commit(t.Context()))
late := nextChangeDelivery(t, stream)
require.EqualValues(t, 201, late.Change.Key)
assert.GreaterOrEqual(t, stream.ServerWALEnd(), stream.Delivered(), "after a change written below Delivered")
}On Non-blockingN5. On PostgreSQL 16, the slot does not keep the last confirmed position across the clean shutdown that The doc says "The slot keeps the position last confirmed". In the live shutdown above, the stream confirmed each Nothing is lost. A stream reopened from the caller's own checkpoint is honored, since the server forwards only a start that lies below the slot. A resume can still misread it, though. L3's follow-up reads N6. Four of the five places that key
Test case: passes on 30f16d7, and each of the four mutants fails it// Every kind of event keys its entry on the delivered position it arrived
// with, not on its own LSN: an UPDATE of a key the buffer does not hold, a
// DELETE, and a key move, whose moved image and old-key marker both owe the
// stream the same position.
func TestBufferKeysEveryEventKindOnItsDeliveredPosition(t *testing.T) {
oldKey := int64(1)
for name, ev := range map[string]decode.ChangeEvent{
"update": {Kind: decode.Update, LSN: 50, Delivered: 80, Key: 2, Columns: []decode.Column{col("label", "a")}},
"delete": {Kind: decode.Delete, LSN: 50, Delivered: 80, Key: 2},
"key move": {Kind: decode.Update, LSN: 50, Delivered: 80, Key: 2, OldKey: &oldKey, Columns: []decode.Column{col("label", "a")}},
} {
t.Run(name, func(t *testing.T) {
b := NewBuffer()
require.NoError(t, b.Add(ev))
oldest, ok := b.OldestPending()
require.True(t, ok)
assert.Equal(t, decode.LSN(80), oldest)
for key, e := range b.entries {
assert.Equal(t, decode.LSN(80), e.FirstLSN, "entry for key %d", key)
}
})
}
}
N7. #150 made This is latent: today only tests open streams. It fires when the resume path opens a stream from a config that The adoption lens, the verified list and the mutation table are in 2/2. This review was generated by Claude Code (claude-opus-5-5). |
|
🤖 Delta review of 30f16d7 (2/2): 0 blocking, 1 non-blocking. This comment covers the two lenses, OSS adoption and integration ease for SchemaBot and other importers, for what changed since For adoption, the delta makes the stream easier to build on.
Non-blockingA1. Even with B2 fixed, My earlier note 2 called the keepalive value the server's position. That was too strong. A logical walsender's keepalive carries its send position ( A progress surface that renders "lag" from this figure will show near zero during the catch-up an operator is waiting on. Two changes would let an importer show a number it can stand behind:
Verified
This review was generated by Claude Code (claude-opus-5-5). |
aparajon
left a comment
aparajon
left a comment
There was a problem hiding this comment.
🤖 Stamping with comments: 1 blocking, 4 non-blocking (see the review comments above).
This stamp was left by Claude Code (claude-opus-5-5).
ServerWALEnd was overwritten by every change with that change's own WAL position: for logical decoding the walsender writes the record's position into both XLogData fields, not its own send position. Under commit ordering a change can lie below Delivered, so the lag the doc defined wrapped below zero, and read as zero after every commit. Only a keepalive moves it now, it never reads below Delivered, and the doc, SAFETY.md, and the design row name it for what it is: the walsender's send position, a lower bound on lag — lag against the server's WAL end is a pool-side measurement. OpenStream now takes the pool and runs the same cluster-and-database proof as CreateSlot and DropSlot. The derived slot name hashes database, schema, and table, so the same table on two clusters shares a name and a database-name check alone cannot tell them apart. proveSameServer returns the proven identity so the target's database is checked without a second IDENTIFY_SYSTEM. ErrStreamEnded's doc no longer promises the slot keeps the last confirm across the shutdown: a confirm that only moves confirmed_flush is not always written out, and a resume starts from its own checkpoint. Tests: a unit and an integration test for ServerWALEnd, replacing the one that pinned the per-change overwrite; a two-cluster refusal for OpenStream; and a buffer test that keys an UPDATE of an unheld key, a DELETE, and a key move on Delivered.
|
🤖 Adversarial review response — created by Kiran's code review agent (Amp, Claude Opus 4.6) — pull/151, follow-up commit Delta review of
Source: delta review comments 1/2 and 2/2 on head |
|
🤖 Delta review of 8bcae74: 1 blocking, 2 non-blocking. This covers what changed since my review at
Invariants: ST-4 is weakened on PostgreSQL 18 until B3 is fixed:
BlockingB3. On PostgreSQL 18, a dropped publication makes the stream skip a change and deliver a position past it, and The red job (log) fails
Rerunning the job will not clear it. The failing test exposes a real fail-open. Through 17, pgoutput fails with So the change never arrives, a keepalive moves The job's other failure was Suggested fix: fail closed on a warning in copy-both mode. The walsender sends one only to say it will not send something. In my probe it said so once, when it reloaded publications. That makes the warning the stream's only chance to stop before a keepalive moves case *pgproto3.NoticeResponse:
// INV: ST-4 — a warning in copy-both mode is the walsender saying it
// will not send something, such as changes under a publication it
// skipped loading; stopping here keeps a later keepalive from
// delivering a position past them.
if msg.SeverityUnlocalized == "WARNING" {
pgErr := pgconn.ErrorResponseToPgError((*pgproto3.ErrorResponse)(msg))
return Delivery{}, false, fmt.Errorf("%w: ST-4: stream from slot %s: %w", ErrInvariantViolation, s.slotName, pgErr)
}
return Delivery{}, false, nil
case *pgproto3.ParameterStatus:
// An asynchronous message the connection has already recorded.
return Delivery{}, false, nilIf failing on every warning is too broad, matching SQLSTATE Test case: fails on 8bcae74 on PostgreSQL 18, passes with the fix on 16 and 18// A publication dropped under the stream ends it with the server's
// SQLSTATE: before 18 the decoder fails on the missing publication
// (42704); from 18 it skips loading it with a warning (55000) and sends
// nothing for the change, which the stream must refuse rather than step past.
func TestStreamReturnsTheDecodersError(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
stream := f.openStream(t, slot.ConsistentPoint())
_, err := f.pool.Exec(t.Context(), `DROP PUBLICATION `+slot.Name())
require.NoError(t, err)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (101, 'unpublished')`)
var version int
require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT current_setting('server_version_num')::int`).Scan(&version))
want := "42704"
if version >= 180000 {
want = "55000"
}
requireSQLState(t, nextError(t, stream), want)
}On Non-blockingN8. Two of The reply says putting the overwrite back fails both new tests. At this head it fails neither (E1). Every change in those tests carries a position at or below Separately, Test cases: pass on 8bcae74, and each fails its mutantUnit (package // A keepalive sent while a transaction is being replayed carries a position
// above the changes the walsender has yet to send from it, so a change that
// follows the keepalive leaves ServerWALEnd where the keepalive put it,
// even while that position is still above Delivered.
func TestServerWALEndKeepsAKeepaliveAboveALaterChange(t *testing.T) {
s := &Stream{delivered: 100, relation: keyOnlyRelation(7)}
_, _, err := s.handleWALData(walData(160, beginMessage(240)))
require.NoError(t, err)
_, _, err = s.handleKeepalive(t.Context(), pglogrepl.PrimaryKeepaliveMessage{ServerWALEnd: 230})
require.NoError(t, err)
require.Equal(t, LSN(100), s.Delivered(), "a keepalive inside a transaction is not delivered")
_, _, err = s.handleWALData(walData(210, insertMessage(7, "101")))
require.NoError(t, err)
assert.Equal(t, LSN(230), s.ServerWALEnd())
}Integration (package // The cluster proof ties the replication connection to the pool, not to the
// target: a connection and a pool that agree on another database of the
// same cluster are refused as the stream's own ST-3 violation, before the
// server is asked for a slot that database does not hold.
func TestOpenStreamRefusesAPoolOnAnotherDatabase(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
other := dbconn.Config{URL: testutil.NewDatabase(t, f.serverURL)}
otherPool, err := dbconn.NewPool(t.Context(), other)
require.NoError(t, err)
t.Cleanup(otherPool.Close)
stream, err := decode.OpenStream(t.Context(), other, otherPool, f.target, slot.ConsistentPoint())
require.ErrorIs(t, err, decode.ErrInvariantViolation)
assert.Nil(t, stream)
}Both pass on N9. ST-3's Enforced line still names only
This review was generated by Claude Code (claude-opus-5-5). |
aparajon
left a comment
aparajon
left a comment
There was a problem hiding this comment.
🤖 Stamping with comments: 1 blocking, 2 non-blocking (see the review comment above).
This stamp was left by Claude Code (claude-opus-5-5).
From PostgreSQL 18 a walsender that cannot load the slot's publication skips it with a WARNING instead of failing, and sends nothing for the changes under it. The stream dropped every notice, so a later keepalive moved Delivered -- and the caller's Confirm the slot -- past a change the slot will never resend. A warning in copy-both mode is the walsender saying it will withhold something, so handleNotice now ends the stream with ErrInvariantViolation and the server's SQLSTATE reachable; a notice below warning is still informational. The decoder-error test expects the server's SQLSTATE for its major: 42704 before 18, 55000 from 18. Two surviving mutants get tests: a change that follows a keepalive sent mid-transaction leaves ServerWALEnd at the keepalive's position, and a pool on another database of the same cluster is refused by OpenStream's own check rather than the server's missing-slot error. ST-3's enforced line names OpenStream among the entry points that run the cluster proof; the SAFETY.md and design decode rows and ST-4 record the warning rule.
|
🤖 Adversarial review response — created by Kiran's code review agent (Amp, Claude Opus 4.6) — pull/151, follow-up commit Delta review of
Not a finding, noted: Source: delta review comment on head |
|
🤖 Delta review of e1fb311: 1 blocking, 2 non-blocking. This covers N8 and N9 are fixed. B3 is partly fixed. The stream now stops on the warning, but only if the warning reaches it. If Invariants: ST-4 is extended by the warning rule, but it stays weakened on 18 for any session that filters warnings, until B4 is fixed. ST-3 is upheld, and its Enforced line now names
Does failing on every BlockingB4. A role, database or server that sets The walsender filters its messages through So the change never arrives, Suggested fix: have the replication connection request warnings itself. A startup parameter overrides role, database and server settings, and it also overrides a value in the URL, because it is assigned after parsing: connConfig.RuntimeParams["replication"] = "database"
// A walsender reports what it will withhold from the stream as a
// warning, so no role, database, or server setting may keep warnings
// from reaching the connection.
connConfig.RuntimeParams["client_min_messages"] = "warning"With this change, the probe stops with Test case: fails on e1fb311 on PostgreSQL 18, passes with the fix on 14, 16 and 18// A database that sends its sessions only errors cannot hide a withheld
// change from the stream: the replication connection asks for warnings
// itself, so a publication dropped under the stream still ends it with the
// server's SQLSTATE on every major.
func TestStreamStopsOnTheWarningWhenTheDatabaseSendsOnlyErrors(t *testing.T) {
f := newSlotFixture(t)
slot := f.createSlot(t)
var database string
require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT current_database()`).Scan(&database))
_, err := f.pool.Exec(t.Context(), `ALTER DATABASE `+pgx.Identifier{database}.Sanitize()+` SET client_min_messages = error`)
require.NoError(t, err)
stream := f.openStream(t, slot.ConsistentPoint())
_, err = f.pool.Exec(t.Context(), `DROP PUBLICATION `+slot.Name())
require.NoError(t, err)
f.exec(t, `INSERT INTO %s.ledger (id, note) VALUES (101, 'unpublished')`)
var version int
require.NoError(t, f.pool.QueryRow(t.Context(), `SELECT current_setting('server_version_num')::int`).Scan(&version))
err = nextError(t, stream)
if version >= 180000 {
require.ErrorIs(t, err, decode.ErrInvariantViolation)
requireSQLState(t, err, "55000")
return
}
requireSQLState(t, err, "42704")
}On Non-blockingN10. The red The one failure in the job is
The likely cause is how the matrix job runs: it puts every package on one long-lived server and database ( N11. Matching the localized severity survives the suite.
Test case: passes on e1fb311, fails F5// The stream reads the unlocalized severity: a server running with
// non-English lc_messages localizes Severity, and its warning must still
// stop the stream.
func TestStreamStopsOnALocalizedWarning(t *testing.T) {
s := &Stream{delivered: 100, confirmed: 90}
_, yielded, err := s.handleMessage(t.Context(), &pgproto3.NoticeResponse{
Severity: "WARNUNG", SeverityUnlocalized: "WARNING", Code: "55000", Message: "skipped loading publication",
})
require.ErrorIs(t, err, ErrInvariantViolation)
assert.False(t, yielded)
}
This review was generated by Claude Code (claude-opus-5-5). |
aparajon
left a comment
aparajon
left a comment
There was a problem hiding this comment.
🤖 Stamping with comments: 1 blocking, 2 non-blocking (see the review comment above).
This stamp was left by Claude Code (claude-opus-5-5).
3 participants
Why
pkg/decodecan create, drop, and inspect the route's slot, but nothing reads it:ChangeEventhas no producer, and the applier, the catch-up loop, and the cutover gate all wait on one. This PR lands the pgoutput stream reader that ST-3, ST-4, CO-4, and CO-8 describe: one connection proven to be a session of the pool's cluster on the target's database, decoding the slot from the consistent point (first run) or the checkpointed applied position (resume) into oneChangeEventper committed row change — with every column's presence carried so the server's unchanged-TOAST marker is never mistaken for a value, and the key a moved row had — and holding the slot's confirmed position where the caller says applied is, never where the server says WAL is. The reaper and the checkpoint-driven resume are the next PRs.Stacked on
kiran01bm/cs7-slot(#150); this PR's diff is the stream only, plus the preflight headroom it needs.Before / after
Target
app.ledgerin databaseshop; slot and publicationpgsprite_3f2a9c01.Positions order transactions by their commit. pgoutput sends a transaction whole when it commits, so a change's own
LSNcan lie below a position the stream already delivered and the caller confirmed:Nothing is lost: the server replays by commit position, so a stream reopened from
0/19278E0yields 201 again. What the caller may confirm while 201 is unapplied is theDeliveredit arrived with — carried on everyChangeEvent, and whatapplier.BufferkeysOldestPendingon — never 201's own LSN.What
pkg/decode/stream.go—OpenStream(takes the pool and runs the same cluster-and-database proof asCreateSlotandDropSlot),Stream{Start, Delivered, ServerWALEnd, Next, Close},Delivery,ErrUnsupportedChange,ErrStreamEnded. Message dispatch (handleCopyData→ keepalive / WAL data;handleNoticestops the stream on a warning), per-message decoders,fail(the first error is stored and returned from every later call).pkg/decode/feedback.go—Confirm,Confirmed,sendStatus: the one place a position is reported to the server, as write, flush, and apply alike.pkg/decode/relation.go— the singlerelationcache built from pgoutput'sRelationMessage: refuses a relation that is not the target (ST-3), records column names, key flags, and replica identity;sameShapedecidesErrSourceShapeChanged.pkg/decode/tuple.go—decodeColumns('t' present, 'n' present-NULL, 'u' absent, 'b' refused, count mismatch refused) anddecodeKey(thePKColumn()must be a flagged key column; parsed asint64).pkg/decode/types.go—ChangeEvent.Delivered, the stream's position when the change arrived.pkg/decode/server_identity.go—proveSameServerreturns the proven identity, soOpenStreamchecks the target's database against it without a secondIDENTIFY_SYSTEM.pkg/applier/buffer.go—FirstLSNkeyed on the event'sDelivered, soOldestPendingnever falls below what the stream confirmed.pkg/preflight/copy_swap_environment.go— a copy-and-swap run needs two free WAL senders; the refusal's detail names both holders.tuple_test.go,stream_test.go(hand-encoded Begin/Commit/Insert bytes: keepalive handling inside and between transactions, unpaired boundaries, a change outside a transaction or for another relation, theDeliveredstamp,ErrStreamEnded,ServerWALEndmoved by keepalives only,Confirmrefusals without a server),buffer_test.go(every event kind keys its entry onDelivered),stream_fixture_integration_test.go(sharedwal_level=logicalfixture, awal_sender_timeout=2svariant,assertWALSenderFlushBecomes),stream_integration_test.go,stream_refusal_integration_test.go,feedback_integration_test.go.pkg/decode/doc.go;SAFETY.mdpkg/decoderow; design package map (decode and applier rows);docs/invariants.mdCO-5, CO-8, ST-3 (OpenStreamjoins the cluster proof), and ST-4 (a warning stops the stream);docs/engine-role.md,docs/refusal-classes.md,docs/tcb-model.mdon sender headroom.Decisions to veto
OpenStreamdials its own connection; there is noSlot.Start. The slot's connection holds the exported snapshot the copier imports; streaming on it would end that transaction and the snapshot with it. A fresh connection means a resume (noSlotin hand) and a first run take the same path. Preflight therefore proves two free WAL senders, not one, so a server with exactly one never passes preflight, builds a shadow, and then fails to open the stream.Next(ctx, wait)is a wall-clock wait, not a context deadline. The caller's context cancels the stream;waitbounds one call so a quiet table never blocks the applier's checkpoint cadence. Implemented withpgconn's timeout-tolerant receive, so the connection is reusable after an elapsed wait.Deliveredmoves on a keepalive only between transactions. A keepalive'sServerWALEndcan lie inside a transaction the stream is mid-way through; advancing to it would let the caller confirm past a change not yet handed over. Inside a transaction the keepalive is answered but the position holds;ServerWALEnd()still reports it.Confirmedalone — zeros before the firstConfirm. The server treats a zero flush position as "no information" and leaves the slot where it is, which is the ST-4 behaviour wanted: the slot moves only on the caller's word.Confirmrecords the position only after the server was told, and a failed send ends the stream. A send that fails leavesConfirmedwhere it was, so a retry from a lower durable checkpoint is not refused; and since a connection that cannot carry a status report is in the same state whichever path tried to send one, the failure latches the stream fromConfirmas it already did from a keepalive reply.ConfirmbelowStartis accepted. A resumed stream may be handed the checkpoint's applied position, which is below the start it was reopened from after a forwarded resume; the server never moves a slot backwards, so the report is harmless and the test proves the slot stays put.Start()is the position asked for, not the one the server granted. The server forwards a start below the slot's confirmed position; the first delivery's position shows where decoding began. Readingconfirmed_flush_lsnbeforeSTART_REPLICATIONbelongs to the checkpoint-driven resume, which opens a stream without aSlotin hand, and is tracked as an internal follow-up there.RelationMessagefor the target with a different column list or replica identity isErrSourceShapeChanged, persistent for the stream; TRUNCATE isErrUnsupportedChange. Whether the route restarts the copy or refuses is the orchestrator's call.ErrStreamEnded, not a violation. The slot is intact with a position at or below the last confirm — a confirm that only movesconfirmed_flushis not always written out before a shutdown — and a stream reopened from the caller's own checkpoint continues, since the server forwards a start below the slot's position; a resume can tell this from slot loss.ServerWALEndis the walsender's send position from its last keepalive, never belowDelivered, and a lower bound on lag. A change'sXLogDatacarries the change's own position in both fields, which under commit ordering can lie belowDelivered, so a change does not move it;max(·, Delivered)keepsServerWALEnd − Deliveredfrom wrapping. Even so the keepalive reports how far the walsender has decoded, not how far the server has written, so the figure stays small while a backlog is undecoded; lag against the server's WAL end is a pool-side measurement againstpg_current_wal_lsn(), which the catch-up progress source takes up. The decode doc,SAFETY.md, and the design row say so.OpenStreamtakes the pool and runsproveSameServer. The derived slot name hashes database, schema, and table, so the same table on two clusters — shards, or a staging and a production copy — shares a slot name, and a config pointing at the other cluster would decode that cluster's slot as this target's changes. The database-name check fromIDENTIFY_SYSTEMalone cannot tell the two apart; the proofCreateSlotandDropSlotalready run can. The signature change reaches only test callers today and the stacked catch-up PR.42704); from 18 the walsender skips loading it with aWARNING(55000), sends nothing for the change, and the next keepalive would moveDelivered— and the caller'sConfirmthe slot — past a change the slot will never resend. The walsender sends a warning only to say it will not send something, so any warning isErrInvariantViolationwith the server's SQLSTATE reachable; a notice below warning is informational and ignored. Matching on55000alone would be narrower; failing on every warning is the fail-closed choice.FOR TABLE <target>; a second relation id is already an invariant violation, so a map would only hide the check.XLogData.WALStart, the change's own position, which can lie belowDeliveredandConfirmed(see above).Deliveredafter the commit isTransactionEndLSN. Adjacent transactions share a boundary (the nextWALStartcan equal the previousTransactionEndLSN), which the tests account for.ColumnkeepsValueandPresentas fields. An accessor that makes an absent value unreadable is tracked as an internal follow-up to land with the applier's flush path, its first caller.ErrSourceShapeChanged,ErrUnsupportedChange, andErrStreamEndedare library errors; the orchestrator that maps them to verdict reasons is wheredocs/refusal-classes.mdgrows.Verification
make lint0 issues;gofmt -lclean;go build ./...;SKIP_INTEGRATION=1 go test ./...green (docs guards included).go test -race -count=1 ./pkg/decode/ ./pkg/preflight/ ./pkg/applier/green on PG 16 (decode ~70 s, preflight ~100 s, applier ~54 s); after the review round,go test -race -count=1 ./pkg/decode/ ./pkg/applier/green again andscripts/test-flaky.sh 'TestServerWALEndNeverTrailsDelivered|TestOpenStreamRefusesAReplicationConnectionOnAnotherServer' 5 ./pkg/decode/5/5.serverWALEndoverwrite failsTestServerWALEndIsNotMovedByAChangeandTestServerWALEndNeverTrailsDelivered(the latter with the0/1927790 < 0/19278E0underflow case); revertingOpenStreamto the database-name check alone failsTestOpenStreamRefusesAReplicationConnectionOnAnotherServer, which then falls through to the other cluster's42704; each of the fourFirstLSNsites reverted toev.LSNfails its subtest ofTestBufferKeysEveryEventKindOnItsDeliveredPosition. Earlier rounds:PG_VERSION={14,15,17,18} go test -count=1 ./pkg/decode/ -run 'TestStream|TestOpenStream|TestConfirm|TestKeepaliveReplies|TestReopened'green on each.TestStreamReturnsTheDecodersErrorexpects42704before 18 andErrInvariantViolationcarrying55000from 18; with the warning handling removed it times out on 18 (16.9 s) and passes with it (1.7 s).TestStreamStopsOnAWarningFromTheServer(unit) fails with the handling removed;TestServerWALEndKeepsAKeepaliveAboveALaterChangefails with the per-change overwrite restored (expected 0xe6, actual 0xd2);TestOpenStreamRefusesAPoolOnAnotherDatabasefails with the target-database check removed (only the server's55000is left in the chain).PG_VERSION=18 go test -race -count=1 ./pkg/decode/andgo test -race -count=1 ./pkg/decode/ ./pkg/applier/on 16 green.go test -count=5 -run 'TestNextReturnsTheCallersDeadline|TestStreamDeliversAnInterleavedTransactionBelowTheConfirmedPosition' ./pkg/decode/— 5/5;scripts/test-flaky.sh 'TestKeepaliveRepliesDoNotMoveTheSlot|TestStreamDeliversATransactionsChangesBeforeMovingThePosition|TestReopenedStreamReplaysFromTheConfirmedPosition' 5 ./pkg/decode— 5/5 with-race.stream_integration_test.go): insert yields every column present; a NULL is present with a nil value; an unchanged out-of-line value (SET STORAGE EXTERNAL,repeat('x', 4000), another column updated) is absent; a changed out-of-line value is present; an UPDATE that moves the key carriesOldKeyunder default replica identity and underREPLICA IDENTITY FULL, and omits it when the key did not move; DELETE carries the key and nil columns; all of a transaction's changes are yielded beforeDeliveredreaches its commit; a rolled-back transaction yields nothing;ADD COLUMN→ErrSourceShapeChangedon the next change and from every later call;TRUNCATE→ErrUnsupportedChange; zero target and zero start position refused before any connection.stream_refusal_integration_test.go): a quiesced database and another database are refused (ST-3); a database of the target's name on a second cluster is refused beforeSTART_REPLICATION, by the stream's own proof rather than the other server's missing-slot error; a renamed source arrives as another relation and is refused; a key column leaving the replica identity is refused; a dropped publication surfaces as*pgconn.PgError42704before 18 and as an ST-4 violation carrying55000from 18; a pool on another database of the same cluster is refused by the stream's own proof;Nextreturns the caller's own context deadline.feedback_integration_test.go):Confirmmovesconfirmed_flush_lsnto exactly the confirmed position; five seconds of keepalive replies on awal_sender_timeout=2sserver leave the slot where it was whileDeliveredrises past writes to an unpublished table;ConfirmbeyondDeliveredrefused and the slot unmoved; a stream reopened from the confirmed position replays the unconfirmed transaction and not the confirmed one; reopened from a forwarded position;Confirmbelow start accepted and — after the walsender's flush position shows the report was read — the slot unmoved; an interleaved transaction arrives below the confirmed position andConfirm(min(Delivered, Buffer.OldestPending))succeeds;ServerWALEnd() ≥ Delivered()holds after a commit and after a change written belowDelivered;Confirmafter the stream stopped returns the stored error; aConfirmwhose send fails leavesConfirmedwhere it was and ends the stream.pkg/preflight): a free slot with exactly two free senders passes;max_wal_senders=1is refused with the two-sender detail; a live sender counts against the headroom.int64boundary with a non-integer, NULL, or missing key refused,sameShapeon id / columns / replica identity; keepalive movesDeliveredonly outside a transaction andServerWALEndalways, a change never movesServerWALEnd(including a change that follows a mid-transaction keepalive), it readsDelivereduntil the server reports, a NOTICE leaves the stream running and a WARNING stops it with its SQLSTATE reachable; an UPDATE of an unheld key, a DELETE, and a key move each keyFirstLSNonDelivered; Commit without Begin, Begin inside Begin, a change outside a transaction, and a change for another relation refused; a change carries theDeliveredit arrived with;CopyDoneends the stream withErrStreamEnded;Confirmregression and beyond-delivered refused; preflight refuses atmax_wal_senders− 1 free.