[Rust] MultiplexedStream follow-ups

[danilonajkov-db]

Follow-ups from review of #188 (MultiplexedStream core). Both are agreed in unresolved review threads and should be addressed before the testing-gated feature is un-gated.

1. Poison/ack race — reserved-capacity ingest (must-do before un-gating)

There is a timing window where ingest_* returns Ok(MessageId) for a record that can never be acked:

• Between the second is_closed check and the actual enqueue, another task can poison the mux and flush past the record.
• wait_for_capacity only checks landing_zone.len() < max_inflight_requests without reserving the slot, so several concurrent callers can all pass that check and then queue past the limit / block inside LandingZone::add, which doesn't observe poisoning.

Agreed plan: add pub(crate) reserve-capacity ingest APIs on ZerobusStream so the mux can reserve a slot → take the ingest/stream-order lock → re-check closed → enqueue_reserved with no new blocking point. Sketch:

let slot = lane.stream.reserve_capacity().await?;
let _stream_order = lane.stream.ingest_lock().await;

let _admit = lane.admission.read().await;
self.check_closed()?;
lane.stream.check_open()?;

let off = lane.stream.enqueue_reserved(encoded_batch, slot);
Ok(MessageId::new(i, off))

Ref: #188 (comment)

2. Eager is_closed() detection of async sub-stream death

If an inner sub-stream dies asynchronously and no caller subsequently touches it (no ingest_*, flush, or wait_for_message_id to that stream), the outer is_closed() keeps returning false and round-robin writes that hit healthy sub-streams are still accepted.

Agreed approach: split into two methods —

• public is_closed() → self.is_closed.load(...) || self.streams.iter().any(|s| s.is_closed())
• private fast is_closed retaining the current lazy hot-path behavior.

Ref: #188 (comment)

---

Tracking issue requested on #188. cc @teodordelibasic-db