You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
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))
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.
Activity