Let a running MoveNextAsync end when the replication connection is disposed - #6718

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

After a backend message bigger than the read buffer (e.g. a notice
longer than 8 KB), the connector reads into a bigger buffer whose array
comes from ArrayPool<byte>.Shared, and a replication stream follows it
there for the rest of the session. When the connector broke while the
consumer held a message (e.g. because a status update failed),
disposing the buffer returned the array to the pool, but the consumer
could still read the message's data from it. If something else rented
the array meanwhile, the consumer silently got that renter's bytes
instead.

The buffer of a broken connector now leaves its array to the GC instead
of returning it to the pool. Closing a connector, resetting a pooled
one and switching to an even bigger buffer still return it.

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>

…sposed

DisposeAsync cancelled the stream and then called MoveNextAsync on its
enumerator itself. If a MoveNextAsync of the consumer was still waiting
for the next message, that call re-entered the async iterator: it took
the result of the incomplete pooled ValueTask of the read the consumer
waited for, which returned that ValueTask's box to its pool. The
consumer's MoveNextAsync never completed, and when the read completed
(at the latest when DisposeAsync closed the connection), its
continuation ran on the recycled box. In Debug builds, where async state
machines are classes, that threw a NullReferenceException on an I/O
thread, which took down the process, e.g. the test host when a
replication mock test failed with a MoveNextAsync pending. In Release
builds, the consumer's call never completed, and with a backlog behind
a slow link, an exception on a thread-pool thread took down the process
as well. The same happened when the consumer called MoveNextAsync while
DisposeAsync stopped the stream (the call failed with
InvalidOperationException, and DisposeAsync never completed), and when
DisposeAsync came while the stream stopped because the consumer broke
out of its loop. A server that answered the cancellation at once didn't
prevent this, since DisposeAsync didn't wait for that answer.

The consumer's calls of the stream's enumerator and DisposeAsync now
take turns. DisposeAsync still cancels the stream first. If a call of
the consumer is running, it waits for that call to end: the call reads
until the server ends the command and fails with
OperationCanceledException, as before when the server answered in time.
Calls of the consumer that start while DisposeAsync stops the stream
wait until it is done, even if DisposeAsync gets called again
meanwhile, and find the stream ended: MoveNextAsync returns false, as it
did after DisposeAsync, and breaking out of the loop, which threw
NotSupportedException, waits too. If the consumer holds a message or
abandoned the enumeration, DisposeAsync stops the stream itself, as
before. Logical decoding plugins read a message's data inside the
consumer's MoveNextAsync but outside the stream's enumerator, so a
DisposeAsync while such a read waits still reads concurrently, as
before.

The wait is bounded by the stop's own timeouts plus a second, so that a
stop is never cut short: Cancellation Timeout, or with a Cancellation
Timeout of 0, WalReceiverTimeout (30 seconds if that's infinite). If the
call still runs by then, e.g. because the server keeps the stream going
without ending the command, DisposeAsync breaks the connection, which
makes the call fail, waits for the call once more, and then closes the
connection as before.

DisposeAsync didn't wait for a running call of the consumer when the
connection was already broken either, and disposed what that call still
needed: Debug builds failed an assertion in DisposeAsync, and in Release
builds the call failed with ObjectDisposedException instead of the
connection's failure. It now waits there too.

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