You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Fix pgoutput decoding of two-phase messages, stream aborts and bad data - #6710
A slot with two-phase decoding enabled (PostgreSQL 15+) sends prepared-transaction messages (b/P/K/r, and p when streaming) whatever protocol version the client requests, but the message instances were only allocated for protocol version 3 and later, so version 1 and 2 sessions failed with a NullReferenceException. The same happened at version 1 with stream aborts, which PostgreSQL 18.0 to at least 18.6 can send even without streaming and at any protocol version (PostgreSQL bug #19616). All message instances are now allocated.
Version 4 sessions read every stream abort as if it carried the abort LSN and timestamp, which the server only sends with streaming = parallel. With streaming = on, and with streaming off when PostgreSQL 18 sent a stream abort anyway, this misread the stream. The layout now follows the streaming option, as on the server.
GIDs of prepared transactions were read with synchronous I/O when they weren't fully buffered, which blocked a thread.
Unknown message codes, unknown tuple types, unknown column kinds and changes for relations without a preceding Relation message now break the connection, the way PostgreSQL's own apply worker errors out. Before, an Insert with an unknown tuple type was read as a new row, and in the other cases the exception left the connection in the Replication state, since nothing disposed the suspended replication enumerator: later commands on it threw NpgsqlOperationInProgressException, and the walsender kept the slot active until the connection was disposed. The errors for Delete and Truncate messages no longer say that they happened in an Update message, and ReplicationConnection.DisposeAsync also cleans up a replication enumerator that was left suspended when the connection broke.
Reading a ReplicationValue after it had been consumed (e.g. the last column after the column enumeration ended) read whatever came next on the wire; it now throws InvalidOperationException, like TupleEnumerator.Current before the first MoveNextAsync. Reading a NULL as a nullable type no longer fails in text mode.
The linked CancellationTokenSource created for each pgoutput and test_decoding enumeration is now disposed.
PgOutputReplicationOptions.Equals and GetHashCode compared the list of publication names by reference; they now compare the names.
How a real server gets there
Each case ran against main and this PR:
Two-phase messages at version 1 or 2: a slot with two_phase enabled, from CreatePgOutputReplicationSlot(twoPhase: true) or flipped for good by any earlier START_REPLICATION with two_phase 'on', read with proto_version 1 or 2 and without the two_phase option. On 15.19, 17.11 and 18.6 main threw the NullReferenceException at the first message and left the connection in the Replication state; this PR delivered BeginPrepare, Relation, Insert, Prepare and CommitPrepared, and with streaming also StreamPrepare.
Stream aborts without streaming (PostgreSQL bug #19616): on PostgreSQL 18.0 to 18.6 with streaming off, a transaction that has a subtransaction with changes, exceeds logical_decoding_work_mem and is already rolled back when the walsender evicts it, which is typical for a consumer that lags behind. On 18.6, main threw the NullReferenceException at version 1; at version 4 it yielded a ParallelStreamAbortMessage built from the next CopyData header and then hung. This PR decoded the 8-byte abort and the following transaction. 17.11, the control, sends no abort. The fix is on PostgreSQL's REL_18_STABLE (caa463e3f8d) but in no release yet.
Version 4 with streaming = on (PostgreSQL 16+): a transaction larger than logical_decoding_work_mem is streamed while in progress and then rolled back. On 16.15 and 18.6 main yielded a ParallelStreamAbortMessage with garbage (LSN 64000000/32770000, a timestamp in 2000) and then hung, or, in a Debug build, threw "Unknown message code: 0" with the connector back in Ready while the walsender was still streaming, so that the next command was fatal on the server ("invalid standby message type"). This PR decoded the abort and the following transaction.
GIDs: a GID that isn't fully buffered yet, e.g. split across TCP segments. With a 163-byte GID behind a proxy that held back the second half of the message for 5 s, main's MoveNextAsync returned after 4.6 s, already completed, so the caller's thread was blocked; this PR's returned pending after 1 ms. The data was right either way. The mock test does the same with a 199-byte GID, the longest PostgreSQL allows: it sends the message up to the middle of the GID and the rest only once MoveNextAsync has returned, which main's call doesn't do before the rest arrives. Sent in one piece, the message passes on main too.
Changes for unknown relations: PostgreSQL 14 to 17 can leave out the Relation messages in the next stream on the same connection after a stream that ended with an error (on 15 to 17 only after a snapshot-exporting CREATE_REPLICATION_SLOT on that connection; fixed in 18 by 80b727eb9de), e.g. for a client that creates its slot, streams, stops with its CancellationToken and starts again. On 17.11 main then threw NpgsqlOperationInProgressException at the next command, and starting again on a new connection failed with 55006 (the slot is active) while the old connection object lived. With this PR the connection is broken, and a new one can start at once.
Values outside their column: consumer code that keeps a ReplicationValue and reads it later, reads Current before MoveNextAsync, or reads a NULL as int?. On 18.6 main returned a value for Current before MoveNextAsync, threw InvalidCastException for the NULL, and returned the next CopyData header's bytes for a value read after the enumeration, after which the stream hung or misread. This PR throws InvalidOperationException for the first and the last, and returns null for the NULL.
The token sources: an application that passes a long-lived token, such as its stopping token, and restarts often. Main kept one registration per run on that token (counted over 10 runs on 18.6); this PR none.
Unknown message codes, tuple types and column kinds: no server sends them; a real server only gets the decoder there after a client-side misalignment of the stream. The breaks are defensive hardening, in line with PostgreSQL's apply worker, which errors on unknown messages too. In an Insert, pgoutput always sends a new tuple ('N'), and PostgreSQL's own subscriber rejects anything else (proto.c on master).
Tests
PgOutputMockTests (22 cases, mock): on main 16 of them fail. The other six are controls that pass on main as well: an 8-byte abort at version 2, the 24-byte abort with streaming = parallel, and the four GID messages sent in one piece.
PgOutputReplicationTests gets a fixture for protocol version 4 with streaming = on, and the two-phase test now also runs at versions 1 and 2 (PostgreSQL 15+).
All pgoutput tests, mock and live, pass against PostgreSQL 18.6.
For the reviewer
No public API change: with streaming = parallel a stream abort still yields ParallelStreamAbortMessage, otherwise StreamAbortMessage, and their summaries now say so instead of naming protocol versions.
Behavior changes, also in the 11.0 release notes: malformed pgoutput data breaks the connection, values can't be read after they were consumed, and TupleEnumerator.Current throws before the first MoveNextAsync.
The fixes for the NullReferenceExceptions and the misread aborts are meant to go to the 10.0 hotfix branch once this is merged.
Dependencies and merge order
It depends on no other PR. #6707 adds a using at the same place in PgOutputReplicationOptions.cs; whichever of the two is merged second keeps both lines.
The reason will be displayed to describe this comment to others. Learn more.
Recommendation
Keep the protocol fixes: they address reproducible failures with valid server output. I found no correctness blocker. I would separate the independent publication-options equality repair and make the small API-documentation correction noted inline.
Problem and reproduction
Prepared-transaction messages depend on persistent slot state (PostgreSQL implementation), not just protocol version. Create a publication, create this slot, then execute BEGIN; INSERT ...; PREPARE TRANSACTION 'gid' on a SQL connection before starting enumeration:
The merge base throws NullReferenceException; HEAD delivers the prepared/commit-prepared sequence when the producer resolves the transaction. V2 has the same result.
Separately, stream and roll back a transaction larger than logical_decoding_work_mem using:
The wire format includes the extra abort fields only for parallel streaming. The merge base overreads the short abort, yields ParallelStreamAbortMessage, and loses framing. HEAD yields StreamAbortMessage and correctly reads the following transaction. PostgreSQL's bug #19616 fix also confirms unexpected aborts with streaming disabled on affected PostgreSQL 18 servers.
Expected approach and fit
I would expect a small C# decoder repair: allocate messages independently of negotiation, distinguish abort layouts, read GIDs asynchronously, and add wire-level and live regressions. No public API expansion or runtime abstraction is warranted. The implementation follows this approach; most added code is tests/mock infrastructure.
The value-state guards, NULL handling, token-source disposal, and terminating unusable streams are small, defensible repairs. The options equality change is correct but unrelated to decoding and is the clearest scope reduction, particularly for backports.
Gaps
The abort types' XML summaries still promise a version-based distinction; update them to distinguish nonparallel and parallel streaming. No broader malformed-packet hardening project is needed here.
Manual validation
I ran a standalone C# consumer against an isolated PostgreSQL 18.4 instance, linked first to the merge base and then HEAD, reproducing the V1/V2 and V4 failures above and their fixes. I also exercised the value API: the merge base rejected Get<int?>() on SQL NULL and returned "d" (the next CopyData header byte) for an expired one-character value. HEAD returned NULL and threw InvalidOperationException, respectively, and rejected Current before MoveNextAsync.
Additionally, 24 mock, 16 live two-phase, and 24 live logical-message cases passed. Fragmented GIDs/malformed messages were covered by mocks; the streaming-disabled PostgreSQL 18 abort was verified against upstream source and mocks, not manually reproduced live. No tracked files were changed.
I've kept the equality fix here: it's 16 lines and one unit test. The backport to the 10.0 hotfix branch will take only the decoding fixes from this PR, so it doesn't widen what goes back.
The reason will be displayed to describe this comment to others. Learn more.
🔵 Needs a closer look
Notice and error fields still use ASCII serialization with UTF-8 byte lengths, corrupting non-ASCII content and producing invalid message lengths.
No open issues 🎉
1 previously missed issue
In code that hasn't changed since last review
Encode notice and error fields as UTF-8
test/Npgsql.Tests/Support/PgServerMock.cs:547
The fields below still use NpgsqlWriteBuffer.WriteNullTerminatedString(string), which encodes ASCII and asserts ASCII-only input in Debug, while these lengths count UTF-8 bytes. For example, WriteNoticeResponse(..., message: "café") writes caf? and advertises a message one byte longer than the serialized payload. Please write each field as UTF-8 followed by its null terminator so non-ASCII notices and errors retain their contents and have valid lengths.
- A slot with two-phase decoding enabled (PostgreSQL 15+) sends prepared
transaction messages (b/P/K/r, and p when streaming) whatever the
requested protocol version, but the message instances were only
allocated for protocol version 3 and later, so protocol version 1
and 2 sessions failed with a NullReferenceException. The same happened
with protocol version 1 for stream aborts, which PostgreSQL 18.0 to at
least 18.6 can send even without streaming and with any protocol
version (bug #19616). All message instances are now allocated.
- Protocol version 4 sessions read every stream abort as if it carried
the abort LSN and timestamp, which the server only sends with
streaming = parallel. With streaming = on, and with streaming off when
PostgreSQL 18 sent a stream abort anyway (bug #19616), this misparsed
the stream. The layout is now chosen by the message length.
- GIDs of prepared transactions were read with synchronous I/O when they
didn't fit into the buffered data, blocking a thread.
- Unknown message codes, unknown tuple types, unknown column kinds and
changes for relations without a preceding Relation message now break
the connection, like PostgreSQL's own apply worker errors out. Before,
an Insert with an unknown tuple type was read as a new row, and in the
other cases the exception left the connection in the Replication
state, since nothing disposed the suspended replication enumerator:
later commands on it threw NpgsqlOperationInProgressException, and the
walsender kept the slot active until the connection was disposed. The
errors for Delete and Truncate messages no longer say that they
happened in an Update message. ReplicationConnection.DisposeAsync now
also cleans up a replication enumerator that was left suspended when
the connection broke.
- Reading a ReplicationValue after it had been consumed (e.g. the last
column after the column enumeration ended) read whatever came next on
the wire; it now throws InvalidOperationException, like
TupleEnumerator.Current before the first MoveNextAsync. Reading a NULL
as a nullable type no longer fails in text mode.
- The linked CancellationTokenSource created for each pgoutput and
test_decoding enumeration is now disposed.
- PgOutputReplicationOptions.Equals and GetHashCode compared the list of
publication names by reference; they now compare the names.
The pgoutput tests get a protocol version 4 fixture with streaming = on,
and the two-phase test now also runs with protocol versions 1 and 2.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
- The layout of a stream abort message follows our streaming option, as the
server's does (pgoutput.c's write_abort_info): with streaming = parallel it
has the abort LSN and timestamp, otherwise not. This replaces telling them
apart by the message length.
- The summaries of StreamAbortMessage and ParallelStreamAbortMessage say so,
instead of naming protocol versions.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
The reason will be displayed to describe this comment to others. Learn more.
🟢 Approval recommended
The review identifies no active issues with the pgoutput decoding and replication cleanup changes.
No open issues 🎉
Review effort: Balanced
This branch has not been deployed
No deployments
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
ReplicationConnection.DisposeAsyncalso cleans up a replication enumerator that was left suspended when the connection broke.ReplicationValueafter it had been consumed (e.g. the last column after the column enumeration ended) read whatever came next on the wire; it now throws InvalidOperationException, likeTupleEnumerator.Currentbefore the firstMoveNextAsync. Reading a NULL as a nullable type no longer fails in text mode.PgOutputReplicationOptions.EqualsandGetHashCodecompared the list of publication names by reference; they now compare the names.How a real server gets there
Each case ran against main and this PR:
CreatePgOutputReplicationSlot(twoPhase: true)or flipped for good by any earlier START_REPLICATION with two_phase 'on', read with proto_version 1 or 2 and without the two_phase option. On 15.19, 17.11 and 18.6 main threw the NullReferenceException at the first message and left the connection in the Replication state; this PR delivered BeginPrepare, Relation, Insert, Prepare and CommitPrepared, and with streaming also StreamPrepare.MoveNextAsyncreturned after 4.6 s, already completed, so the caller's thread was blocked; this PR's returned pending after 1 ms. The data was right either way. The mock test does the same with a 199-byte GID, the longest PostgreSQL allows: it sends the message up to the middle of the GID and the rest only onceMoveNextAsynchas returned, which main's call doesn't do before the rest arrives. Sent in one piece, the message passes on main too.ReplicationValueand reads it later, readsCurrentbeforeMoveNextAsync, or reads a NULL asint?. On 18.6 main returned a value forCurrentbeforeMoveNextAsync, threw InvalidCastException for the NULL, and returned the next CopyData header's bytes for a value read after the enumeration, after which the stream hung or misread. This PR throws InvalidOperationException for the first and the last, and returns null for the NULL.Tests
PgOutputMockTests(22 cases, mock): on main 16 of them fail. The other six are controls that pass on main as well: an 8-byte abort at version 2, the 24-byte abort with streaming = parallel, and the four GID messages sent in one piece.PgOutputReplicationTestsgets a fixture for protocol version 4 with streaming = on, and the two-phase test now also runs at versions 1 and 2 (PostgreSQL 15+).For the reviewer
ParallelStreamAbortMessage, otherwiseStreamAbortMessage, and their summaries now say so instead of naming protocol versions.TupleEnumerator.Currentthrows before the firstMoveNextAsync.Dependencies and merge order
It depends on no other PR. #6707 adds a
usingat the same place inPgOutputReplicationOptions.cs; whichever of the two is merged second keeps both lines.Fixes #6696