Stop replication streams properly when they are disposed early - #6716

Depends on #6689, #6690, #6709 and #6715, which aren't merged yet; their commits show up here until they are.

• Leaving the loop. Breaking out of await foreach over a replication stream, 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 session's temporary slots, which the logical StartReplication methods now document. 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, and the exception that made the consumer stop is never replaced.
• Late status updates. 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. The end of the stream now waits for an update in progress, and later ones are dropped.
• Unread data. 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.
• Cancellation. After cancellation, messages that were already on their way were still handed out until the server's error arrived. They are now skipped before they count as received, so the next START_REPLICATION from the reported position gets them again. This makes PhysicalReplicationTests reliable, so they run again (some of them also used colliding slot names).
• A second stream. 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.
• Oversize messages. 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, which Read pgoutput messages from the connector's current read buffer #6721 fixes.

How a real server gets there

Every case was run on PostgreSQL 18.6, most also on 14.24, against main, against #6715 where it matters, and against this PR:

• Leaving the loop, then using the connection: a consumer stops early because it has what it needs or because its own code throws (a failing sink), and then runs IDENTIFY_SYSTEM, a slot command or StartReplication again. On main IDENTIFY_SYSTEM threw NpgsqlOperationInProgressException ("The connection is already in state 'Replication'"), pg_stat_replication kept showing the walsender in catchup with a moving reply_time (the status timer kept running), the next StartReplication failed the same way, and the server logged "unexpected EOF on standby connection" at the disposal. With this PR IDENTIFY_SYSTEM worked, the walsender went back to its startup state, and a second START_REPLICATION on the same walsender worked; test_decoding and pgoutput behaved the same, with the server logging the cancellation and the release of the slot.
• Stopping inside a big message: a physical stream that catches up gets XLogData of up to 128 kB, 16 times the 8-kB read buffer. This PR skipped the rest of a 125,944-byte message of which the consumer had read 100 bytes, sent CopyDone and drained, and the connection worked; the same for 100 bytes of a 20,000-character pgoutput value.
• An unread pgoutput row: a consumer that filters by relation stops after a change without reading its row. Main's DisposeAsync read the row's bytes as backend messages ("There is not enough data left in the buffer" from ParameterDescriptionMessage.Load, swallowed). This PR skipped the row and sent the cancellation, and the slot was released.
• A second StartReplication: on main, "break, then StartReplication again" does exactly this, because the break leaves the first enumerator suspended. In Release, DisposeAsync then returned at once with both timers still set, and no cancellation reached the server ("unexpected EOF on standby connection"); a Debug build stopped the process on a Debug.Assert. With this PR the first stream was untouched, and DisposeAsync stopped it within about 25 ms.
• Cancellation with data in flight, as during a catch-up or on a busy logical stream: with about 6 MB of backlog, main handed out 3 more 131,072-byte messages after the cancellation, this PR none.
• The server goes away while the consumer throws (pg_terminate_backend, a shutdown, a lost connection): this PR logs event 50013, surfaces the consumer's exception and breaks the connection. Main has no stop path, so this test guards the new code.
• A status update still on its way when the stream ends (a timer callback, or SendStatusUpdate on another thread): forced by holding the feedback semaphore and suspending the walsender, so that the next command stays blocked in its write. On main and on Handle the end of a replication stream at timeline switches and shutdown #6715, the late update failed with "Did not write the amount of bytes the message length specified", the connection broke, and the server logged "incomplete message from client". With this PR the end of the stream waited, and the update went out first. Without forcing, the window is one status-update write per stream end, so this is rare; there is no test for it.
• Reading beyond a message: the only real case found is the stream abort that protocol version 4 with streaming = on misread before Fix pgoutput decoding of two-phase messages, stream aborts and bad data #6710. On main (18.6 and 16.15) it handed out garbage and then left the connector Ready while the walsender was still in copy mode, so that the next command was fatal on the server; with this PR it fails with "A replication message was read beyond its end." and breaks the connection. With that PR merged, the check is defensive hardening: it turns a silent misread into a clean break. No test.
• A long notice mid-stream: notices can come between CopyData messages at any time. A realistic trigger is a publication row filter (PostgreSQL 15+) with websearch_to_tsquery and a row whose text holds more than about 8 KB of whitespace, which raises a notice that repeats the whole input. On 18.6 and 15.19 main failed with "Unknown replication message code 'd'" after the first row; with this PR the replication stream handles the notice, and the failure moves to the pgoutput decoder. The mock test's physical variant needs a debug setting on a real server.

Tests

• ReplicationStopMockTests (7 cases, mock): on Handle the end of a replication stream at timeline switches and shutdown #6715's code, 6 of them fail (the stop tests time out because nothing ends copy mode, and messages are handed out after the cancellation), and Oversize_message_during_streaming hangs.
• Live: ReplicationStopTests<Logical|Physical>.Stop_early_and_go_on, PgOutputStopTests.Stop_in_the_middle_of_a_change (a 20,000-character row), and the 9 PhysicalReplicationTests, which were [Explicit("Flakiness")] and now run ([NonParallelizable]).
• The replication tests pass against PostgreSQL 18.6, in Debug and Release.

For the reviewer

• New public constants in NpgsqlEventId: 50008 (StoppingReplication) and 50013 (StoppingReplicationFailed), both at Debug level. The class suppresses RS0016, so PublicAPI.Unshipped.txt doesn't change; there is no other API change.
• On PostgreSQL 14 to 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 to 17 if the connection created the slot with an exported snapshot; fixed in PostgreSQL 18 by 80b727eb9de). The pgoutput StartReplication docs say to start the next stream on a new connection, and the test avoids it.
• Not changed: the logical stop doesn't wait for the server to read its final status update before it cancels. A busy walsender only reads replies once its output buffer fills, so stops would stall in large transactions.
• Behavior changes, also in the 11.0 release notes: leaving the loop stops the stream and leaves the connection usable; a logical stop drops the session's temporary slots, and its final status update is best effort; SendStatusUpdate after the loop throws InvalidOperationException (send it before leaving the loop, or pass the processed position as walLocation when starting again); a second StartReplication while one is active throws NpgsqlOperationInProgressException.

Dependencies and merge order

• Contains the commits of Add copy-both and replication support to the test server mock #6689, Fix TIMELINE_HISTORY, SHOW ALL and CommandTimeout on replication connections #6709, Add a throwaway primary/standby cluster for replication tests #6690 and Handle the end of a replication stream at timeline switches and shutdown #6715; merge those first. Only the last commit is this PR's own; the others drop out when the branch is rebased onto main after they land.
• Land it together with Handle the end of a replication stream at timeline switches and shutdown #6715, and not without Let a running MoveNextAsync end when the replication connection is disposed #6718: once leaving the loop disposes the inner enumerator, today's ReplicationConnection.DisposeAsync can run into it (a break while DisposeAsync stopped the stream threw NotSupportedException, and a DisposeAsync during the consumer's stop was still pending after 20 s), which that PR fixes. That PR in turn goes in with Don't return the read buffer of a broken connector to the ArrayPool #6717 or after it. So the stream PRs go in back to back: Handle the end of a replication stream at timeline switches and shutdown #6715, this one, Don't return the read buffer of a broken connector to the ArrayPool #6717, Let a running MoveNextAsync end when the replication connection is disposed #6718.

Fixes #5523
Part of #6699 (the oversize-message bullet; #6721 completes it)