Read pgoutput messages from the connector's current read buffer - #6721

…ections

- TimelineHistory threw an InvalidCastException on PostgreSQL 14 and
  later, which declare the content column as text instead of bytea. The
  column carries the raw bytes of the history file either way, so it is
  now read as raw bytes. That also stops bytea unescaping from mangling
  backslashes in the content on older servers, and leaves the bytea
  parsing without a use, so it is removed.
- Show("all") returns many rows, which made the command helper expect
  CommandComplete, get another DataRow and break the connection. Further
  rows are now skipped, and Show throws NotSupportedException for results
  that aren't a single value, leaving the connection usable.
- The ReplicationConnection.CommandTimeout property was ignored for
  replication commands, since every command reset the timeouts to the
  connection string's command timeout. It now applies to all commands,
  and setting it before Open no longer throws. Timeouts that don't fit
  into an int of milliseconds (e.g. TimeSpan.MaxValue) mean no timeout.
- StartReplication(PhysicalReplicationSlot, CancellationToken) throws
  ArgumentNullException for a null slot instead of a
  NullReferenceException.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

PgTestCluster creates PostgreSQL clusters in temporary directories with
the binaries that NPGSQL_TEST_PG_BIN points to (initdb, pg_ctl,
pg_basebackup): a primary that allows physical and logical replication,
streaming standbys of it, promotion and shutdown in any mode. This
covers tests that can't run against the shared test server, like
timeline switches and server shutdowns. Tests using it are ignored when
the variable isn't set, so CI is unaffected.

The clusters listen on 127.0.0.1 only (no Unix sockets, whose default
directory isn't writable for unprivileged users on Debian-based
builds), pick a free port right before each start, retry when another
process took it, and limit every step to well below CI's 30 seconds per
test.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

- At the end of a historic timeline (e.g. after the standby we stream
  from got promoted), the server ends copy mode with CopyDone. Npgsql
  expected only CopyData and broke the connection ("Received backend
  message CopyDone while expecting CopyDataMessage"). It now answers
  with CopyDone, reads the next timeline the server sends, both
  CommandCompletes and ReadyForQuery, and ends the enumeration normally,
  leaving the connection usable. Status updates stop before our
  CopyDone, since the protocol doesn't allow CopyData after it. The next
  timeline is logged, and the StartReplication docs explain how to
  continue: IDENTIFY_SYSTEM returns the current timeline and
  TIMELINE_HISTORY the position where it branched off, which can be
  before the end of the WAL received on the old timeline.
- Asked to start at the end of a historic timeline, the server doesn't
  enter copy mode but sends the next timeline right away, which Npgsql
  treated as an unexpected message. The enumeration now ends right away.
- A walsender that exits after everything has been replicated (e.g. at
  server shutdown, npgsql#5752) sends CommandComplete, which also broke the
  connection with a misleading error. It now throws an NpgsqlException
  saying that the server ended the replication stream.
- In logical replication, unknown CopyData message types are skipped
  instead of breaking the connection, like the logical replication
  worker, which ignores them on purpose. Physical replication keeps
  rejecting them, like walreceiver, since it must not lose WAL.
  Keepalive fields that newer servers may add are skipped in both.
- In physical replication, the received position reported to the server
  was the server's own WAL end, not the end of the WAL received, so
  pg_stat_replication.write_lsn and synchronous_commit = remote_write
  treated WAL as written before it arrived. It is now the end of the
  data the consumer is done with, like walreceiver reports what it
  wrote.
- LastReceivedLsn starts over at the start position of each stream
  instead of keeping the maximum of all previous streams, which was
  wrong when continuing on a new timeline at a lower position.
- The XLogData stream is disposed after each message instead of later in
  cleanup, where it could skip data based on a stale position: a server
  error that ended the stream after a keepalive (e.g. after
  pg_cancel_backend) turned into a timeout and a broken connection.

The new cluster tests stream physically across a standby's promotion
and through a fast shutdown, using the throwaway clusters from
NPGSQL_TEST_PG_BIN.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

- Breaking out of a replication enumeration, or an exception in the
  consumer, left the inner enumerator undisposed. The connection stayed
  in replication mode with its feedback timers running, and every other
  command failed with NpgsqlOperationInProgressException. The
  enumerators are now disposed, which stops streaming: physical
  replication ends copy mode with CopyDone, which keeps the walsender
  ready for the next START_REPLICATION and keeps temporary slots;
  logical replication cancels the command, since a logical walsender
  can't stream again after CopyDone. Like any error, the cancellation
  drops the temporary slots of the session (documented on the logical
  StartReplication methods). Servers before PostgreSQL 10 ignore
  cancellation requests on replication connections, so with them the
  logical stop breaks the connection instead.
  A final status update reports the processed position first; for
  logical replication this is best effort, since the walsender may
  handle the cancellation before it reads the update. If stopping fails,
  the connection is broken; the exception that made the consumer stop is
  never replaced. On PostgreSQL 14-17, the server may not send
  pgoutput's Relation messages again in the next stream on the same
  connection after such a cancellation (on 14 after any error, on 15-17
  if the connection created the slot with an exported snapshot; fixed in
  PostgreSQL 18 by 80b727eb9de). This is documented on the pgoutput
  StartReplication, and the test avoids it.
- Status updates that were still on their way when a stream ended (e.g.
  from a timer or another thread) could be written after the end of the
  command, interleaved with the next one. They are now dropped.
- Whatever the consumer didn't read of the message it held is skipped
  first, also when a plugin like pgoutput reads the data directly from
  the buffer. Reading beyond the end of a message breaks the connection.
- After cancellation, messages that were already on their way were still
  handed out until the server's error arrived. They are now skipped.
  This makes the physical replication tests reliable, so they run again
  (some of them also used colliding slot names).
- Starting a second stream while one was in progress overwrote the state
  of the first one before failing, so disposing the connection didn't
  stop the first stream anymore. It now fails before touching anything,
  and a start that fails leaves no state behind.
- A message bigger than the read buffer (e.g. a long notice) made the
  connector switch to a bigger buffer while the stream kept reading from
  the old one. This is fixed for the replication stream itself; the
  pgoutput decoder still reads from the old buffer.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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

A backend message bigger than the read buffer (e.g. a notice longer
than 8 KB) makes the connector switch to a bigger buffer for the rest
of the replication session. The replication stream follows the switch,
but the pgoutput decoder kept the buffer it found when the enumeration
started, and its tuples, column enumerators and values kept the one
from when they were created. After such a message they read from the
old buffer, bypassing what the connector had already buffered in the
new one, and misread the stream from there, or waited for more data
although the next message had already arrived. The stream's check
whether a plugin read a message beyond its end measured the new
buffer, so it didn't notice.

The decoder now takes the connector's buffer for each message. The
tuples take it when they are populated for a message and pass it on to
their column enumerators and values. They don't look it up when the
consumer reads a row, since a broken connector no longer has a buffer
(e.g. after a failed status update while the consumer held the
message).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>