Stream Coordinator can leave a stream in `stopped` in a 2/3 node loss disaster

[the-mikedavis]

Describe the bug

When a stream has its writer member on node A and replicas on nodes B and C, when nodes B and C are removed from the cluster in sequence (by forget_cluster_node), the writer on node A is terminated as a side effect of the first delete_replica and is not re-elected. The writer can't be healed by anything in the current stream coordinator API:

1. evaluate_stream/3 has no branch that re-acts on a writer already in {stopped, _, _}
2. the only path to {ready, NextEpoch} is the select_leader/2 call inside member_stopped's is_quorum branch, and once the first member_stopped event has fired for the writer, no later event re-evaluates the quorum predicate
3. the filter_command/3 clause for delete_replica rejects any further delete as last_stream_member

The log is still intact on A, but the writer can't be recovered. This sequence is actually OK as long as the writer was originally on B or C during the disaster event: then the stream can be recovered with a sequence of force-shrinks of Khepri, the Stream Coordinator (and any QQs), and then normal stream membership change commands.

Reproduction steps

I found this behavior by simulating a disaster scenario where 2/3 nodes of a cluster are lost.

1. 3-node cluster with a stream declared.
2. Identify the node running the stream writer, call it A.
3. Kill nodes B and C (SIGTERM, simultaneous or staggered). Wait for rabbit on node B/C down on A. (Treat B and C as permanently lost now, all commands are against A from here on.)
4. Force-shrink Khepri: rabbitmqctl force_standalone_khepri_boot
5. Force-shrink the stream coordinator: rabbitmqctl eval 'ra_server_proc:force_shrink_members_to_current_member({rabbit_stream_coordinator, node()}).'
6. Issue delete_replica for B and then C. (forget_cluster_node also does this.) Note that the writer terminates on A (osiris_writer:terminate/2 ... reason:shutdown) when the first delete_replica is applied, and no started writer ... in 2 is logged.

The condition for triggering the bug is precisely 'the stream's writer was on the surviving node at the time delete_replica is first issued'. Streams whose writer was on one of the dead nodes recover normally via the {down, Pid, Reason} -> {down, Epoch} path in update_stream0/3.

add_replica invoked on the stuck stream fails with:

[warning] queue '<Q>' ... failed to add replica on 'rabbit@<D>':
   {noproc,{gen_batch_server,call,[<OLD_WRITER_PID>,query_replication_state,5000]}}

where <OLD_WRITER_PID> is the PID that terminated at the osiris_writer:terminate line above.

rabbitmq-streams restart_stream <Q> hangs indefinitely and never returns. The command is applied to the Ra state machine, sets target = stopped on the single existing member (a no-op since it is already stopped), but evaluate_stream/3 finds no actionable state transition and never emits a reply effect, so the reply_to Pid waits forever.

Expected behavior

The stream coordinator should either not leave the writer member in a stopped state, or have an API to recover it. I would lean towards the former since that's how it behaves if the writer is on one of the lost nodes.

Additional context

This was reproduced against main from a few weeks ago, 2bd3449. Like QQ/Khepri/Coordinator force-shrinking, we can't guarantee that there won't be data loss in the disaster-recovery, but we can provide tools to recover availability. I think we might be able to fix it simply by re-evaluating is_quorum in member_deleted.