datum/actor/

stream_ref_proto.rs

//! Protobuf StreamRefs protocol and transport-agnostic endpoint drivers.
//!
//! This module mirrors Akka's StreamRefs protocol without embedding an actor
//! transport. A carrier feeds inbound protobuf frames into one endpoint and
//! drains outbound frames from the same endpoint. `datum-net` uses this seam to
//! carry StreamRefs over a reliable, ordered QUIC bidirectional stream.

use std::{
    any::Any,
    collections::VecDeque,
    fmt,
    sync::{
        Arc, Condvar, Mutex, MutexGuard,
        atomic::{AtomicU64, Ordering},
    },
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use crate::stream::{
    BoxStream, Materializer, NotUsed, Sink, Source, StreamCompletion, StreamError, StreamResult,
    TerminalSinkConsumerDyn, TerminalSourceHookDyn, TerminalSourceStatus,
};
use futures::channel::oneshot;
use prost::Message as ProstMessage;
use tokio::sync::mpsc as tokio_mpsc;

use super::{SourceRef, StreamRefSettings};

static STREAM_REF_PROTO_ID: AtomicU64 = AtomicU64::new(1);

/// Element payload codec used by the protobuf StreamRefs transport seam.
///
/// The built-in impls use the same big-endian primitive encodings as Ractor's
/// `BytesConvertable`, but the trait is owned by `datum-core` so the protobuf
/// seam does not depend on the `cluster` feature.
pub trait StreamRefPayload: Send + 'static {
    fn encode_stream_ref_payload(self) -> Vec<u8>;

    fn encode_stream_ref_payload_into(self, bytes: &mut Vec<u8>)
    where
        Self: Sized,
    {
        bytes.extend(self.encode_stream_ref_payload());
    }

    fn decode_stream_ref_payload(bytes: Vec<u8>) -> StreamResult<Self>
    where
        Self: Sized;

    fn decode_stream_ref_payload_slice(bytes: &[u8]) -> StreamResult<Self>
    where
        Self: Sized,
    {
        Self::decode_stream_ref_payload(bytes.to_vec())
    }
}

macro_rules! impl_stream_ref_payload_numeric {
    ($($ty:ty),* $(,)?) => {
        $(
            impl StreamRefPayload for $ty {
                fn encode_stream_ref_payload(self) -> Vec<u8> {
                    self.to_be_bytes().to_vec()
                }

                fn encode_stream_ref_payload_into(self, bytes: &mut Vec<u8>) {
                    bytes.extend(self.to_be_bytes());
                }

                fn decode_stream_ref_payload(bytes: Vec<u8>) -> StreamResult<Self> {
                    let data: [u8; std::mem::size_of::<Self>()] =
                        bytes.as_slice().try_into().map_err(|_| {
                            StreamError::Failed(format!(
                                "invalid {} stream ref payload length: {}",
                                stringify!($ty),
                                bytes.len()
                            ))
                        })?;
                    Ok(Self::from_be_bytes(data))
                }

                fn decode_stream_ref_payload_slice(bytes: &[u8]) -> StreamResult<Self> {
                    let data: [u8; std::mem::size_of::<Self>()] =
                        bytes.try_into().map_err(|_| {
                            StreamError::Failed(format!(
                                "invalid {} stream ref payload length: {}",
                                stringify!($ty),
                                bytes.len()
                            ))
                        })?;
                    Ok(Self::from_be_bytes(data))
                }
            }
        )*
    };
}

impl_stream_ref_payload_numeric!(i8, i16, i32, i64, i128, u8, u16, u32, u64, u128, f32, f64);

impl StreamRefPayload for bool {
    fn encode_stream_ref_payload(self) -> Vec<u8> {
        vec![u8::from(self)]
    }

    fn encode_stream_ref_payload_into(self, bytes: &mut Vec<u8>) {
        bytes.push(u8::from(self));
    }

    fn decode_stream_ref_payload(bytes: Vec<u8>) -> StreamResult<Self> {
        match bytes.as_slice() {
            [0] => Ok(false),
            [1] => Ok(true),
            _ => Err(StreamError::Failed(
                "invalid bool stream ref payload".to_owned(),
            )),
        }
    }

    fn decode_stream_ref_payload_slice(bytes: &[u8]) -> StreamResult<Self> {
        match bytes {
            [0] => Ok(false),
            [1] => Ok(true),
            _ => Err(StreamError::Failed(
                "invalid bool stream ref payload".to_owned(),
            )),
        }
    }
}

impl StreamRefPayload for String {
    fn encode_stream_ref_payload(self) -> Vec<u8> {
        self.into_bytes()
    }

    fn encode_stream_ref_payload_into(self, bytes: &mut Vec<u8>) {
        bytes.extend(self.into_bytes());
    }

    fn decode_stream_ref_payload(bytes: Vec<u8>) -> StreamResult<Self> {
        String::from_utf8(bytes)
            .map_err(|error| StreamError::Failed(format!("invalid UTF-8 payload: {error}")))
    }

    fn decode_stream_ref_payload_slice(bytes: &[u8]) -> StreamResult<Self> {
        String::from_utf8(bytes.to_vec())
            .map_err(|error| StreamError::Failed(format!("invalid UTF-8 payload: {error}")))
    }
}

impl StreamRefPayload for Vec<u8> {
    fn encode_stream_ref_payload(self) -> Vec<u8> {
        self
    }

    fn encode_stream_ref_payload_into(self, bytes: &mut Vec<u8>) {
        bytes.extend(self);
    }

    fn decode_stream_ref_payload(bytes: Vec<u8>) -> StreamResult<Self> {
        Ok(bytes)
    }

    fn decode_stream_ref_payload_slice(bytes: &[u8]) -> StreamResult<Self> {
        Ok(bytes.to_vec())
    }
}

/// Stream-ref identifier scoped to one transport connection.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct StreamRefId(u128);

impl StreamRefId {
    /// Generates a process-local id suitable for a new connection-scoped ref.
    #[must_use]
    pub fn new() -> Self {
        let sequence = STREAM_REF_PROTO_ID.fetch_add(1, Ordering::Relaxed) as u128;
        let timestamp = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|duration| duration.as_nanos())
            .unwrap_or_default();
        let pid = std::process::id() as u128;
        Self(timestamp ^ (pid << 32) ^ sequence)
    }

    /// Constructs an id from a stable numeric value, primarily for tests and
    /// single-stream carriers that reserve a well-known id.
    #[must_use]
    pub const fn from_u128(value: u128) -> Self {
        Self(value)
    }

    #[must_use]
    pub const fn as_u128(self) -> u128 {
        self.0
    }

    #[must_use]
    pub fn to_bytes(self) -> [u8; 16] {
        self.0.to_be_bytes()
    }

    pub fn from_bytes(bytes: &[u8]) -> StreamResult<Self> {
        let value: [u8; 16] = bytes.try_into().map_err(|_| {
            StreamError::Failed("stream ref id must be exactly 16 bytes".to_owned())
        })?;
        Ok(Self(u128::from_be_bytes(value)))
    }
}

impl Default for StreamRefId {
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Display for StreamRefId {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{:032x}", self.0)
    }
}

/// Protobuf payload wrapper matching Akka's `Payload`, without Akka serializer
/// ids or manifests.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamRefPayloadBytes {
    pub bytes: Vec<u8>,
}

/// Transport-agnostic StreamRefs protocol messages.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StreamRefMessage {
    OnSubscribeHandshake,
    CumulativeDemand {
        seq_nr: u64,
    },
    SequencedOnNext {
        seq_nr: u64,
        payload: StreamRefPayloadBytes,
    },
    RemoteStreamCompleted {
        seq_nr: u64,
    },
    RemoteStreamFailure {
        cause: Vec<u8>,
    },
    Ack,
}

impl StreamRefMessage {
    #[must_use]
    pub fn failure_text(&self) -> Option<String> {
        match self {
            Self::RemoteStreamFailure { cause } => {
                Some(String::from_utf8_lossy(cause).into_owned())
            }
            _ => None,
        }
    }

    fn is_ack(&self) -> bool {
        matches!(self, Self::Ack)
    }
}

/// One protobuf frame tagged with a connection-scoped stream-ref id.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StreamRefFrame {
    pub stream_ref_id: StreamRefId,
    pub message: StreamRefMessage,
}

impl StreamRefFrame {
    #[must_use]
    pub fn new(stream_ref_id: StreamRefId, message: StreamRefMessage) -> Self {
        Self {
            stream_ref_id,
            message,
        }
    }

    #[must_use]
    pub fn encode_to_vec(&self) -> Vec<u8> {
        self.to_wire().encode_to_vec()
    }

    pub fn decode(bytes: &[u8]) -> StreamResult<Self> {
        Self::from_wire(WireStreamRefFrame::decode(bytes).map_err(|error| {
            StreamError::Failed(format!("invalid stream ref protobuf frame: {error}"))
        })?)
    }

    fn to_wire(&self) -> WireStreamRefFrame {
        WireStreamRefFrame {
            stream_ref_id: self.stream_ref_id.to_bytes().to_vec(),
            message: Some(match &self.message {
                StreamRefMessage::OnSubscribeHandshake => {
                    wire_stream_ref_frame::Message::OnSubscribeHandshake(
                        WireOnSubscribeHandshake {},
                    )
                }
                StreamRefMessage::CumulativeDemand { seq_nr } => {
                    wire_stream_ref_frame::Message::CumulativeDemand(WireCumulativeDemand {
                        seq_nr: *seq_nr,
                    })
                }
                StreamRefMessage::SequencedOnNext { seq_nr, payload } => {
                    wire_stream_ref_frame::Message::SequencedOnNext(WireSequencedOnNext {
                        seq_nr: *seq_nr,
                        payload: Some(WirePayload {
                            enclosed_message: payload.bytes.clone(),
                        }),
                    })
                }
                StreamRefMessage::RemoteStreamCompleted { seq_nr } => {
                    wire_stream_ref_frame::Message::RemoteStreamCompleted(
                        WireRemoteStreamCompleted { seq_nr: *seq_nr },
                    )
                }
                StreamRefMessage::RemoteStreamFailure { cause } => {
                    wire_stream_ref_frame::Message::RemoteStreamFailure(WireRemoteStreamFailure {
                        cause: cause.clone(),
                    })
                }
                StreamRefMessage::Ack => wire_stream_ref_frame::Message::Ack(WireAck {}),
            }),
        }
    }

    fn from_wire(wire: WireStreamRefFrame) -> StreamResult<Self> {
        let stream_ref_id = StreamRefId::from_bytes(&wire.stream_ref_id)?;
        let message = match wire.message.ok_or_else(|| {
            StreamError::Failed("stream ref protobuf frame has no message".to_owned())
        })? {
            wire_stream_ref_frame::Message::OnSubscribeHandshake(_) => {
                StreamRefMessage::OnSubscribeHandshake
            }
            wire_stream_ref_frame::Message::CumulativeDemand(message) => {
                StreamRefMessage::CumulativeDemand {
                    seq_nr: message.seq_nr,
                }
            }
            wire_stream_ref_frame::Message::SequencedOnNext(message) => {
                let payload = message.payload.ok_or_else(|| {
                    StreamError::Failed("SequencedOnNext missing payload".to_owned())
                })?;
                StreamRefMessage::SequencedOnNext {
                    seq_nr: message.seq_nr,
                    payload: StreamRefPayloadBytes {
                        bytes: payload.enclosed_message,
                    },
                }
            }
            wire_stream_ref_frame::Message::RemoteStreamCompleted(message) => {
                StreamRefMessage::RemoteStreamCompleted {
                    seq_nr: message.seq_nr,
                }
            }
            wire_stream_ref_frame::Message::RemoteStreamFailure(message) => {
                StreamRefMessage::RemoteStreamFailure {
                    cause: message.cause,
                }
            }
            wire_stream_ref_frame::Message::Ack(_) => StreamRefMessage::Ack,
        };
        Ok(Self {
            stream_ref_id,
            message,
        })
    }
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireStreamRefFrame {
    #[prost(bytes = "vec", tag = "1")]
    stream_ref_id: Vec<u8>,
    #[prost(oneof = "wire_stream_ref_frame::Message", tags = "2, 3, 4, 5, 6, 7")]
    message: Option<wire_stream_ref_frame::Message>,
}

mod wire_stream_ref_frame {
    #[derive(Clone, PartialEq, prost::Oneof)]
    pub enum Message {
        #[prost(message, tag = "2")]
        OnSubscribeHandshake(super::WireOnSubscribeHandshake),
        #[prost(message, tag = "3")]
        CumulativeDemand(super::WireCumulativeDemand),
        #[prost(message, tag = "4")]
        SequencedOnNext(super::WireSequencedOnNext),
        #[prost(message, tag = "5")]
        RemoteStreamCompleted(super::WireRemoteStreamCompleted),
        #[prost(message, tag = "6")]
        RemoteStreamFailure(super::WireRemoteStreamFailure),
        #[prost(message, tag = "7")]
        Ack(super::WireAck),
    }
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WirePayload {
    #[prost(bytes = "vec", tag = "1")]
    enclosed_message: Vec<u8>,
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireOnSubscribeHandshake {}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireCumulativeDemand {
    #[prost(uint64, tag = "1")]
    seq_nr: u64,
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireSequencedOnNext {
    #[prost(uint64, tag = "1")]
    seq_nr: u64,
    #[prost(message, optional, tag = "2")]
    payload: Option<WirePayload>,
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireRemoteStreamFailure {
    #[prost(bytes = "vec", tag = "1")]
    cause: Vec<u8>,
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireRemoteStreamCompleted {
    #[prost(uint64, tag = "1")]
    seq_nr: u64,
}

#[derive(Clone, PartialEq, ProstMessage)]
struct WireAck {}

#[derive(Debug, Clone)]
struct StreamRefPayloadSegment {
    offset: usize,
    len: usize,
}

/// A contiguous batch of already-encoded `SequencedOnNext` payloads.
#[derive(Debug, Clone)]
pub struct StreamRefPayloadBatch {
    stream_ref_id: StreamRefId,
    first_seq_nr: u64,
    payloads: Vec<u8>,
    segments: Vec<StreamRefPayloadSegment>,
}

impl StreamRefPayloadBatch {
    #[must_use]
    pub fn new(stream_ref_id: StreamRefId, first_seq_nr: u64) -> Self {
        Self {
            stream_ref_id,
            first_seq_nr,
            payloads: Vec::new(),
            segments: Vec::new(),
        }
    }

    #[must_use]
    pub fn stream_ref_id(&self) -> StreamRefId {
        self.stream_ref_id
    }

    #[must_use]
    pub fn first_seq_nr(&self) -> u64 {
        self.first_seq_nr
    }

    #[must_use]
    pub fn count(&self) -> usize {
        self.segments.len()
    }

    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.segments.is_empty()
    }

    #[must_use]
    pub fn payload_len(&self, index: usize) -> usize {
        self.segments[index].len
    }

    #[must_use]
    pub fn payload(&self, index: usize) -> &[u8] {
        let segment = &self.segments[index];
        &self.payloads[segment.offset..segment.offset + segment.len]
    }

    pub fn push_payload<T>(&mut self, item: T) -> StreamResult<()>
    where
        T: StreamRefPayload,
    {
        let offset = self.payloads.len();
        item.encode_stream_ref_payload_into(&mut self.payloads);
        let len = self.payloads.len().saturating_sub(offset);
        if len > u32::MAX as usize {
            return Err(StreamError::LimitExceeded {
                max: u32::MAX as u64,
            });
        }
        self.segments.push(StreamRefPayloadSegment { offset, len });
        Ok(())
    }

    fn into_single_frame(self) -> StreamResult<StreamRefFrame> {
        if self.count() != 1 {
            return Err(StreamError::Failed(
                "stream ref batch cannot be converted into a single frame".to_owned(),
            ));
        }
        Ok(StreamRefFrame::new(
            self.stream_ref_id,
            StreamRefMessage::SequencedOnNext {
                seq_nr: self.first_seq_nr,
                payload: StreamRefPayloadBytes {
                    bytes: self.payload(0).to_vec(),
                },
            },
        ))
    }
}

/// An outbound protocol item. Control remains a protobuf frame; steady data can
/// move as one compact batch without first inflating one frame per element.
#[derive(Debug, Clone)]
pub enum StreamRefOutbound {
    Frame(StreamRefFrame),
    SequencedBatch(StreamRefPayloadBatch),
}

impl StreamRefOutbound {
    fn into_single_frame(self) -> StreamResult<StreamRefFrame> {
        match self {
            Self::Frame(frame) => Ok(frame),
            Self::SequencedBatch(batch) => batch.into_single_frame(),
        }
    }
}

/// Common interface for external carriers that pump protobuf frames.
pub trait StreamRefProtoEndpoint: Clone + Send + Sync + 'static {
    fn stream_ref_id(&self) -> StreamRefId;
    fn next_frame(&self) -> Option<StreamResult<StreamRefFrame>>;
    fn next_outbound(
        &self,
        max_data_elements: usize,
        _max_data_bytes: usize,
    ) -> Option<StreamResult<StreamRefOutbound>> {
        let _ = max_data_elements;
        self.next_frame()
            .map(|frame| frame.map(StreamRefOutbound::Frame))
    }
    fn handle_frame(&self, frame: StreamRefFrame) -> StreamResult<()>;
    fn handle_sequenced_on_next_batch(
        &self,
        stream_ref_id: StreamRefId,
        first_seq_nr: u64,
        payloads: &[&[u8]],
    ) -> StreamResult<()> {
        for (index, payload) in payloads.iter().enumerate() {
            let seq_nr = first_seq_nr.checked_add(index as u64).ok_or_else(|| {
                StreamError::Failed("stream ref batch sequence overflow".to_owned())
            })?;
            self.handle_frame(StreamRefFrame::new(
                stream_ref_id,
                StreamRefMessage::SequencedOnNext {
                    seq_nr,
                    payload: StreamRefPayloadBytes {
                        bytes: payload.to_vec(),
                    },
                },
            ))?;
        }
        Ok(())
    }
    fn fail_connection(&self, error: StreamError);
}

/// Immediate outbound poll result for non-blocking carriers.
#[doc(hidden)]
pub enum StreamRefOutboundPoll {
    Ready(StreamResult<StreamRefOutbound>),
    Pending,
    Closed,
}

/// Internal bridge used by single-owner async carriers.
#[doc(hidden)]
pub trait StreamRefProtoEndpointWake: StreamRefProtoEndpoint {
    fn install_outbound_wake(&self, sender: tokio_mpsc::Sender<()>);
    fn clear_outbound_wake(&self);
    fn try_next_outbound(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> StreamRefOutboundPoll;
}

#[derive(Default)]
struct OutboundWake {
    sender: Mutex<Option<tokio_mpsc::Sender<()>>>,
}

impl OutboundWake {
    fn install(&self, sender: tokio_mpsc::Sender<()>) {
        *self
            .sender
            .lock()
            .unwrap_or_else(|poison| poison.into_inner()) = Some(sender);
    }

    fn clear(&self) {
        *self
            .sender
            .lock()
            .unwrap_or_else(|poison| poison.into_inner()) = None;
    }

    fn wake(&self) {
        let sender = self
            .sender
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
            .clone();
        if let Some(sender) = sender {
            let _ = sender.try_send(());
        }
    }
}

/// Producer-side endpoint for a local `SourceRef` or `Source`.
///
/// Feed inbound `OnSubscribeHandshake`, `CumulativeDemand`, cancellation, and
/// Ack frames with [`StreamRefProtoEndpoint::handle_frame`]. Drain outbound
/// `SequencedOnNext`, completion, failure, and Ack frames with
/// [`StreamRefProtoEndpoint::next_frame`].
pub struct StreamRefProtoProducer<T>
where
    T: StreamRefPayload,
{
    shared: Arc<ProducerShared<T>>,
}

impl<T> Clone for StreamRefProtoProducer<T>
where
    T: StreamRefPayload,
{
    fn clone(&self) -> Self {
        Self {
            shared: Arc::clone(&self.shared),
        }
    }
}

impl<T> StreamRefProtoProducer<T>
where
    T: StreamRefPayload,
{
    pub fn from_source_ref(
        source_ref: SourceRef<T>,
        stream_ref_id: StreamRefId,
        settings: StreamRefSettings,
    ) -> StreamResult<Self> {
        Self::from_source(
            super::stream_ref::proto_source(&source_ref),
            stream_ref_id,
            settings,
        )
    }

    pub fn from_source<Mat>(
        source: Source<T, Mat>,
        stream_ref_id: StreamRefId,
        settings: StreamRefSettings,
    ) -> StreamResult<Self>
    where
        Mat: Send + 'static,
    {
        let materializer = Materializer::new();
        let (input, materialized) = Arc::clone(&source.factory).create(&materializer)?;
        Ok(Self {
            shared: Arc::new(ProducerShared {
                stream_ref_id,
                settings,
                input: Mutex::new(Some(input)),
                state: Mutex::new(ProducerState {
                    partner_seen: false,
                    cumulative_demand: 0,
                    first_demand_deadline: None,
                    sent: 0,
                    terminal_sent: false,
                    waiting_for_ack: false,
                    ack_deadline: None,
                    stopped: None,
                    ack_queued: false,
                    pending_terminal: None,
                    done: false,
                    input_attached: true,
                    terminal_result: None,
                }),
                changed: Condvar::new(),
                outbound_wake: OutboundWake::default(),
                completion: Mutex::new(None),
                _materializer: materializer,
                _materialized: Mutex::new(Some(Box::new(materialized))),
            }),
        })
    }

    /// Creates a producer with no attached input, for the SinkRef sender side.
    ///
    /// The input stream is attached later by materializing the [`Sink`]
    /// returned from [`StreamRefProtoProducer::sink`]. Until then, the
    /// producer's `next_frame` waits on the condvar instead of spinning, so an
    /// idle lazy producer does not busy-loop while the remote consumer has not
    /// yet subscribed or attached demand.
    #[must_use]
    pub fn new_lazy(stream_ref_id: StreamRefId, settings: StreamRefSettings) -> Self {
        Self {
            shared: Arc::new(ProducerShared {
                stream_ref_id,
                settings,
                input: Mutex::new(None),
                state: Mutex::new(ProducerState {
                    partner_seen: false,
                    cumulative_demand: 0,
                    first_demand_deadline: None,
                    sent: 0,
                    terminal_sent: false,
                    waiting_for_ack: false,
                    ack_deadline: None,
                    stopped: None,
                    ack_queued: false,
                    pending_terminal: None,
                    done: false,
                    input_attached: false,
                    terminal_result: None,
                }),
                changed: Condvar::new(),
                outbound_wake: OutboundWake::default(),
                completion: Mutex::new(None),
                _materializer: Materializer::new(),
                _materialized: Mutex::new(None),
            }),
        }
    }

    /// Returns a [`Sink`] whose incoming elements are framed and sent as this
    /// producer's outbound `SequencedOnNext` frames.
    ///
    /// Materializing the sink attaches the input stream to the lazy producer
    /// and returns a [`StreamCompletion`] that resolves when the producer
    /// reaches its terminal state (all elements sent and acknowledged, the
    /// remote cancelled/failed, or the carrier failed).
    #[must_use]
    pub fn sink(&self) -> Sink<T, StreamCompletion<NotUsed>> {
        let shared = Arc::clone(&self.shared);
        Sink::from_runner(move |input, _materializer| {
            let (sender, receiver) = oneshot::channel();
            *shared
                .completion
                .lock()
                .unwrap_or_else(|poison| poison.into_inner()) = Some(sender);
            shared.attach_input(input);
            Ok(StreamCompletion::from_receiver(receiver, None))
        })
    }
}

impl<T> StreamRefProtoEndpoint for StreamRefProtoProducer<T>
where
    T: StreamRefPayload,
{
    fn stream_ref_id(&self) -> StreamRefId {
        self.shared.stream_ref_id
    }

    fn next_frame(&self) -> Option<StreamResult<StreamRefFrame>> {
        self.shared.next_frame()
    }

    fn next_outbound(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> Option<StreamResult<StreamRefOutbound>> {
        self.shared.next_outbound(max_data_elements, max_data_bytes)
    }

    fn handle_frame(&self, frame: StreamRefFrame) -> StreamResult<()> {
        self.shared.handle_frame(frame)
    }

    fn fail_connection(&self, error: StreamError) {
        self.shared.fail_connection(error);
    }
}

impl<T> StreamRefProtoEndpointWake for StreamRefProtoProducer<T>
where
    T: StreamRefPayload,
{
    fn install_outbound_wake(&self, sender: tokio_mpsc::Sender<()>) {
        self.shared.outbound_wake.install(sender);
    }

    fn clear_outbound_wake(&self) {
        self.shared.outbound_wake.clear();
    }

    fn try_next_outbound(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> StreamRefOutboundPoll {
        self.shared
            .try_next_outbound(max_data_elements, max_data_bytes)
    }
}

struct ProducerShared<T>
where
    T: StreamRefPayload,
{
    stream_ref_id: StreamRefId,
    settings: StreamRefSettings,
    input: Mutex<Option<BoxStream<T>>>,
    state: Mutex<ProducerState>,
    changed: Condvar,
    outbound_wake: OutboundWake,
    completion: Mutex<Option<oneshot::Sender<StreamResult<NotUsed>>>>,
    _materializer: Materializer,
    _materialized: Mutex<Option<Box<dyn Any + Send>>>,
}

struct ProducerState {
    partner_seen: bool,
    cumulative_demand: u64,
    first_demand_deadline: Option<Instant>,
    sent: u64,
    terminal_sent: bool,
    waiting_for_ack: bool,
    ack_deadline: Option<Instant>,
    stopped: Option<StreamError>,
    ack_queued: bool,
    pending_terminal: Option<StreamRefMessage>,
    done: bool,
    input_attached: bool,
    terminal_result: Option<StreamResult<NotUsed>>,
}

enum ProducerBatchPoll {
    Ready(StreamResult<StreamRefOutbound>),
    Pending,
    StateChanged,
}

enum InputItemPoll<T> {
    Ready(StreamResult<T>),
    Pending,
    TerminalQueued,
}

impl<T> ProducerShared<T>
where
    T: StreamRefPayload,
{
    fn lock_state(&self) -> MutexGuard<'_, ProducerState> {
        self.state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn lock_input(&self) -> MutexGuard<'_, Option<BoxStream<T>>> {
        self.input
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn notify_changed(&self) {
        self.changed.notify_all();
        self.outbound_wake.wake();
    }

    fn frame(&self, message: StreamRefMessage) -> StreamRefFrame {
        StreamRefFrame::new(self.stream_ref_id, message)
    }

    fn next_frame(&self) -> Option<StreamResult<StreamRefFrame>> {
        self.next_outbound(1, usize::MAX)
            .map(|outbound| outbound.and_then(StreamRefOutbound::into_single_frame))
    }

    fn next_outbound(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> Option<StreamResult<StreamRefOutbound>> {
        let subscription_deadline = deadline_from_now(self.settings.subscription_timeout());
        loop {
            let mut state = self.lock_state();
            if state.done {
                return None;
            }

            if state.ack_queued {
                state.ack_queued = false;
                state.done = true;
                state.terminal_result = Some(match state.stopped.clone() {
                    Some(error) => Err(error),
                    None => Ok(NotUsed),
                });
                self.notify_changed();
                drop(state);
                self.drop_input();
                self.settle();
                return Some(Ok(StreamRefOutbound::Frame(
                    self.frame(StreamRefMessage::Ack),
                )));
            }

            if let Some(message) = state.pending_terminal.take() {
                drop(state);
                return Some(Ok(StreamRefOutbound::Frame(self.frame(message))));
            }

            if state.waiting_for_ack {
                if state
                    .ack_deadline
                    .is_some_and(|deadline| Instant::now() >= deadline)
                {
                    let timeout_error =
                        subscription_timeout_error("stream ref producer terminal ack");
                    state.done = true;
                    state.terminal_result = Some(Err(timeout_error.clone()));
                    self.notify_changed();
                    drop(state);
                    self.drop_input();
                    self.settle();
                    return Some(Err(timeout_error));
                }
                if let Some(remaining) = state
                    .ack_deadline
                    .and_then(|deadline| deadline.checked_duration_since(Instant::now()))
                {
                    let (next, _) = wait_timeout_unpoison(&self.changed, state, remaining);
                    drop(next);
                } else {
                    drop(state);
                }
                continue;
            }

            if let Some(error) = state.stopped.clone() {
                state.done = true;
                state.terminal_result = Some(Err(error.clone()));
                self.notify_changed();
                drop(state);
                self.drop_input();
                self.settle();
                return Some(Err(error));
            }

            if state.cumulative_demand > 0 && state.sent < state.cumulative_demand {
                drop(state);
                if let Some(outbound) =
                    self.pull_next_outbound_batch(max_data_elements.max(1), max_data_bytes.max(1))
                {
                    return Some(outbound);
                }
                continue;
            }

            if state.cumulative_demand == 0 && Instant::now() >= subscription_deadline {
                let timeout_error = subscription_timeout_error("stream ref producer first demand");
                state.done = true;
                state.terminal_result = Some(Err(timeout_error.clone()));
                self.notify_changed();
                drop(state);
                self.drop_input();
                self.settle();
                return Some(Err(timeout_error));
            }

            if state.cumulative_demand == 0 {
                let remaining = subscription_deadline.saturating_duration_since(Instant::now());
                if remaining.is_zero() {
                    drop(state);
                    continue;
                }
                let (next, _) = wait_timeout_unpoison(&self.changed, state, remaining);
                drop(next);
            } else {
                let next = wait_unpoison(&self.changed, state);
                drop(next);
            }
        }
    }

    fn try_next_outbound(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> StreamRefOutboundPoll {
        loop {
            let mut state = self.lock_state();
            if state.done {
                return StreamRefOutboundPoll::Closed;
            }

            if state.ack_queued {
                state.ack_queued = false;
                state.done = true;
                state.terminal_result = Some(match state.stopped.clone() {
                    Some(error) => Err(error),
                    None => Ok(NotUsed),
                });
                self.notify_changed();
                drop(state);
                self.drop_input();
                self.settle();
                return StreamRefOutboundPoll::Ready(Ok(StreamRefOutbound::Frame(
                    self.frame(StreamRefMessage::Ack),
                )));
            }

            if let Some(message) = state.pending_terminal.take() {
                drop(state);
                return StreamRefOutboundPoll::Ready(Ok(StreamRefOutbound::Frame(
                    self.frame(message),
                )));
            }

            if state.waiting_for_ack {
                if state
                    .ack_deadline
                    .is_some_and(|deadline| Instant::now() >= deadline)
                {
                    let timeout_error =
                        subscription_timeout_error("stream ref producer terminal ack");
                    state.done = true;
                    state.terminal_result = Some(Err(timeout_error.clone()));
                    self.notify_changed();
                    drop(state);
                    self.drop_input();
                    self.settle();
                    return StreamRefOutboundPoll::Ready(Err(timeout_error));
                }
                return StreamRefOutboundPoll::Pending;
            }

            if let Some(error) = state.stopped.clone() {
                state.done = true;
                state.terminal_result = Some(Err(error.clone()));
                self.notify_changed();
                drop(state);
                self.drop_input();
                self.settle();
                return StreamRefOutboundPoll::Ready(Err(error));
            }

            if state.cumulative_demand > 0 && state.sent < state.cumulative_demand {
                drop(state);
                match self
                    .try_pull_next_outbound_batch(max_data_elements.max(1), max_data_bytes.max(1))
                {
                    ProducerBatchPoll::Ready(outbound) => {
                        return StreamRefOutboundPoll::Ready(outbound);
                    }
                    ProducerBatchPoll::StateChanged => continue,
                    ProducerBatchPoll::Pending => return StreamRefOutboundPoll::Pending,
                }
            }

            if state.cumulative_demand == 0 {
                let deadline = *state
                    .first_demand_deadline
                    .get_or_insert_with(|| deadline_from_now(self.settings.subscription_timeout()));
                if Instant::now() >= deadline {
                    let timeout_error =
                        subscription_timeout_error("stream ref producer first demand");
                    state.done = true;
                    state.terminal_result = Some(Err(timeout_error.clone()));
                    self.notify_changed();
                    drop(state);
                    self.drop_input();
                    self.settle();
                    return StreamRefOutboundPoll::Ready(Err(timeout_error));
                }
            }

            return StreamRefOutboundPoll::Pending;
        }
    }

    fn pull_next_outbound_batch(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> Option<StreamResult<StreamRefOutbound>> {
        let mut batch: Option<StreamRefPayloadBatch> = None;
        while batch
            .as_ref()
            .is_none_or(|batch| batch.count() < max_data_elements)
        {
            if batch
                .as_ref()
                .is_some_and(|batch| batch.payloads.len() >= max_data_bytes)
            {
                break;
            }

            let seq_nr = {
                let state = self.lock_state();
                if state.done
                    || state.stopped.is_some()
                    || state.waiting_for_ack
                    || state.sent >= state.cumulative_demand
                {
                    break;
                }
                state.sent
            };

            let item = match self.next_input_item() {
                Some(item) => item,
                None => break,
            };

            match item {
                Ok(item) => {
                    let mut state = self.lock_state();
                    if state.done
                        || state.stopped.is_some()
                        || state.waiting_for_ack
                        || state.sent != seq_nr
                    {
                        break;
                    }
                    state.sent = state.sent.saturating_add(1);
                    drop(state);

                    let batch = batch.get_or_insert_with(|| {
                        StreamRefPayloadBatch::new(self.stream_ref_id, seq_nr)
                    });
                    if let Err(error) = batch.push_payload(item) {
                        return Some(Err(error));
                    }
                }
                Err(error) => {
                    let terminal = StreamRefMessage::RemoteStreamFailure {
                        cause: failure_cause(&error),
                    };
                    self.note_terminal(Err(error), terminal.clone());
                    return match batch {
                        Some(batch) if !batch.is_empty() => {
                            Some(Ok(StreamRefOutbound::SequencedBatch(batch)))
                        }
                        _ => None,
                    };
                }
            }
        }

        batch
            .filter(|batch| !batch.is_empty())
            .map(|batch| Ok(StreamRefOutbound::SequencedBatch(batch)))
    }

    fn try_pull_next_outbound_batch(
        &self,
        max_data_elements: usize,
        max_data_bytes: usize,
    ) -> ProducerBatchPoll {
        let mut batch: Option<StreamRefPayloadBatch> = None;
        while batch
            .as_ref()
            .is_none_or(|batch| batch.count() < max_data_elements)
        {
            if batch
                .as_ref()
                .is_some_and(|batch| batch.payloads.len() >= max_data_bytes)
            {
                break;
            }

            let seq_nr = {
                let state = self.lock_state();
                if state.done
                    || state.stopped.is_some()
                    || state.waiting_for_ack
                    || state.sent >= state.cumulative_demand
                {
                    break;
                }
                if !state.input_attached {
                    return match batch {
                        Some(batch) if !batch.is_empty() => {
                            ProducerBatchPoll::Ready(Ok(StreamRefOutbound::SequencedBatch(batch)))
                        }
                        _ => ProducerBatchPoll::Pending,
                    };
                }
                state.sent
            };

            let item = match self.try_next_input_item() {
                InputItemPoll::Ready(item) => item,
                InputItemPoll::Pending => {
                    return match batch {
                        Some(batch) if !batch.is_empty() => {
                            ProducerBatchPoll::Ready(Ok(StreamRefOutbound::SequencedBatch(batch)))
                        }
                        _ => ProducerBatchPoll::Pending,
                    };
                }
                InputItemPoll::TerminalQueued => {
                    return match batch {
                        Some(batch) if !batch.is_empty() => {
                            ProducerBatchPoll::Ready(Ok(StreamRefOutbound::SequencedBatch(batch)))
                        }
                        _ => ProducerBatchPoll::StateChanged,
                    };
                }
            };

            match item {
                Ok(item) => {
                    let mut state = self.lock_state();
                    if state.done
                        || state.stopped.is_some()
                        || state.waiting_for_ack
                        || state.sent != seq_nr
                    {
                        break;
                    }
                    state.sent = state.sent.saturating_add(1);
                    drop(state);

                    let batch = batch.get_or_insert_with(|| {
                        StreamRefPayloadBatch::new(self.stream_ref_id, seq_nr)
                    });
                    if let Err(error) = batch.push_payload(item) {
                        return ProducerBatchPoll::Ready(Err(error));
                    }
                }
                Err(error) => {
                    let terminal = StreamRefMessage::RemoteStreamFailure {
                        cause: failure_cause(&error),
                    };
                    self.note_terminal(Err(error), terminal);
                    return match batch {
                        Some(batch) if !batch.is_empty() => {
                            ProducerBatchPoll::Ready(Ok(StreamRefOutbound::SequencedBatch(batch)))
                        }
                        _ => ProducerBatchPoll::StateChanged,
                    };
                }
            }
        }

        match batch {
            Some(batch) if !batch.is_empty() => {
                ProducerBatchPoll::Ready(Ok(StreamRefOutbound::SequencedBatch(batch)))
            }
            _ => ProducerBatchPoll::Pending,
        }
    }

    fn next_input_item(&self) -> Option<StreamResult<T>> {
        let mut input_guard = self.lock_input();
        if input_guard.is_none() {
            drop(input_guard);
            // Lazy producer: no input attached yet. Block on the condvar
            // (notify-driven, no fixed poll) until `attach_input` or a
            // terminal state wakes us, then let the caller re-check demand and
            // terminal predicates.
            let mut state = self.lock_state();
            while !state.input_attached
                && !state.done
                && state.stopped.is_none()
                && !state.terminal_sent
            {
                state = wait_unpoison(&self.changed, state);
            }
            drop(state);
            return None;
        }

        match input_guard.as_mut().expect("input attached").next() {
            Some(item) => Some(item),
            None => {
                drop(input_guard);
                let seq_nr = self.lock_state().sent;
                self.note_terminal(
                    Ok(NotUsed),
                    StreamRefMessage::RemoteStreamCompleted { seq_nr },
                );
                None
            }
        }
    }

    fn try_next_input_item(&self) -> InputItemPoll<T> {
        let mut input_guard = self.lock_input();
        let Some(input) = input_guard.as_mut() else {
            return InputItemPoll::Pending;
        };

        match input.next() {
            Some(item) => InputItemPoll::Ready(item),
            None => {
                drop(input_guard);
                let seq_nr = self.lock_state().sent;
                self.note_terminal(
                    Ok(NotUsed),
                    StreamRefMessage::RemoteStreamCompleted { seq_nr },
                );
                InputItemPoll::TerminalQueued
            }
        }
    }

    fn note_terminal(&self, result: StreamResult<NotUsed>, terminal_message: StreamRefMessage) {
        self.drop_input();
        let mut state = self.lock_state();
        if state.done || state.terminal_sent {
            return;
        }
        state.terminal_sent = true;
        state.waiting_for_ack = true;
        state.terminal_result = Some(result);
        state.ack_deadline = Some(deadline_from_now(self.settings.subscription_timeout()));
        state.pending_terminal = Some(terminal_message);
        self.notify_changed();
        drop(state);
    }

    fn handle_frame(&self, frame: StreamRefFrame) -> StreamResult<()> {
        self.validate_frame_id(frame.stream_ref_id)?;
        match frame.message {
            StreamRefMessage::OnSubscribeHandshake => {
                let mut state = self.lock_state();
                state.partner_seen = true;
                self.notify_changed();
                drop(state);
                Ok(())
            }
            StreamRefMessage::CumulativeDemand { seq_nr } => {
                if seq_nr == 0 {
                    return Err(StreamError::Failed(
                        "CumulativeDemand seq_nr must be positive".to_owned(),
                    ));
                }
                let mut state = self.lock_state();
                state.partner_seen = true;
                if seq_nr > state.cumulative_demand {
                    state.cumulative_demand = seq_nr;
                }
                self.notify_changed();
                drop(state);
                Ok(())
            }
            StreamRefMessage::RemoteStreamCompleted { .. } => {
                self.stop_from_consumer(StreamError::Cancelled);
                Ok(())
            }
            StreamRefMessage::RemoteStreamFailure { cause } => {
                self.stop_from_consumer(StreamError::Failed(
                    String::from_utf8_lossy(&cause).into_owned(),
                ));
                Ok(())
            }
            StreamRefMessage::Ack => {
                let mut state = self.lock_state();
                if state.waiting_for_ack {
                    state.waiting_for_ack = false;
                    state.done = true;
                    if state.terminal_result.is_none() {
                        state.terminal_result = Some(Ok(NotUsed));
                    }
                    self.notify_changed();
                    drop(state);
                    self.drop_input();
                    self.settle();
                } else {
                    drop(state);
                }
                Ok(())
            }
            StreamRefMessage::SequencedOnNext { .. } => Err(StreamError::Failed(
                "producer endpoint cannot receive SequencedOnNext".to_owned(),
            )),
        }
    }

    fn stop_from_consumer(&self, error: StreamError) {
        let mut state = self.lock_state();
        if !state.done {
            state.stopped = Some(error.clone());
            state.ack_queued = true;
            state.terminal_result = Some(Err(error));
        }
        self.notify_changed();
        drop(state);
        self.drop_input();
    }

    fn fail_connection(&self, error: StreamError) {
        let mut state = self.lock_state();
        if !state.done {
            state.stopped = Some(error.clone());
            state.done = true;
            state.terminal_result = Some(Err(error));
        }
        self.notify_changed();
        drop(state);
        self.drop_input();
        self.settle();
    }

    fn attach_input(&self, input: BoxStream<T>) {
        *self.lock_input() = Some(input);
        let mut state = self.lock_state();
        state.input_attached = true;
        self.notify_changed();
        drop(state);
    }

    fn settle(&self) {
        let result = self.lock_state().terminal_result.clone();
        let sender = self
            .completion
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
            .take();
        if let (Some(sender), Some(result)) = (sender, result) {
            let _ = sender.send(result);
        }
    }

    fn drop_input(&self) {
        let input = self.lock_input().take();
        drop(input);
    }

    fn validate_frame_id(&self, stream_ref_id: StreamRefId) -> StreamResult<()> {
        if stream_ref_id == self.stream_ref_id {
            Ok(())
        } else {
            Err(StreamError::Failed(format!(
                "stream ref id mismatch: expected {}, got {}",
                self.stream_ref_id, stream_ref_id
            )))
        }
    }
}

/// Consumer-side endpoint that exposes inbound remote elements as a local
/// [`Source`].
///
/// The source sends `OnSubscribeHandshake` and a cumulative demand ceiling when
/// it is materialized. It then refills demand at half the configured buffer.
pub struct StreamRefProtoConsumer<T>
where
    T: StreamRefPayload,
{
    shared: Arc<ConsumerShared<T>>,
}

impl<T> Clone for StreamRefProtoConsumer<T>
where
    T: StreamRefPayload,
{
    fn clone(&self) -> Self {
        Self {
            shared: Arc::clone(&self.shared),
        }
    }
}

impl<T> StreamRefProtoConsumer<T>
where
    T: StreamRefPayload,
{
    #[must_use]
    pub fn new(stream_ref_id: StreamRefId, settings: StreamRefSettings) -> Self {
        Self {
            shared: Arc::new(ConsumerShared {
                stream_ref_id,
                settings,
                state: Mutex::new(ConsumerState {
                    source_taken: false,
                    subscribed: false,
                    queue: VecDeque::new(),
                    direct_terminal: false,
                    direct_consumer: None,
                    direct_cancelled: None,
                    terminal: None,
                    expected_seq: 0,
                    delivered: 0,
                    cumulative_demand: 0,
                    outbound: VecDeque::new(),
                    finish_after_outbound_ack: false,
                    waiting_cancel_ack: false,
                    done: false,
                }),
                changed: Condvar::new(),
                outbound_wake: OutboundWake::default(),
            }),
        }
    }

    #[must_use]
    pub fn source(&self) -> Source<T, NotUsed> {
        let shared_for_stream = Arc::clone(&self.shared);
        let shared_for_terminal = Arc::clone(&self.shared);
        Source::from_terminal_direct_materialized_factory(
            move |_materializer| {
                shared_for_stream
                    .start_stream()
                    .map(|stream| (Box::new(stream) as BoxStream<T>, NotUsed))
            },
            move |_materializer| {
                Ok((
                    Arc::new(StreamRefProtoTerminalHook {
                        shared: Arc::clone(&shared_for_terminal),
                        stream: Mutex::new(None),
                    }) as Arc<dyn TerminalSourceHookDyn<T>>,
                    NotUsed,
                ))
            },
        )
    }
}

impl<T> StreamRefProtoEndpoint for StreamRefProtoConsumer<T>
where
    T: StreamRefPayload,
{
    fn stream_ref_id(&self) -> StreamRefId {
        self.shared.stream_ref_id
    }

    fn next_frame(&self) -> Option<StreamResult<StreamRefFrame>> {
        self.shared.next_frame()
    }

    fn handle_frame(&self, frame: StreamRefFrame) -> StreamResult<()> {
        self.shared.handle_frame(frame)
    }

    fn handle_sequenced_on_next_batch(
        &self,
        stream_ref_id: StreamRefId,
        first_seq_nr: u64,
        payloads: &[&[u8]],
    ) -> StreamResult<()> {
        self.shared
            .handle_sequenced_on_next_batch(stream_ref_id, first_seq_nr, payloads)
    }

    fn fail_connection(&self, error: StreamError) {
        self.shared.fail_connection(error);
    }
}

impl<T> StreamRefProtoEndpointWake for StreamRefProtoConsumer<T>
where
    T: StreamRefPayload,
{
    fn install_outbound_wake(&self, sender: tokio_mpsc::Sender<()>) {
        self.shared.outbound_wake.install(sender);
    }

    fn clear_outbound_wake(&self) {
        self.shared.outbound_wake.clear();
    }

    fn try_next_outbound(
        &self,
        _max_data_elements: usize,
        _max_data_bytes: usize,
    ) -> StreamRefOutboundPoll {
        self.shared.try_next_frame()
    }
}

struct ConsumerShared<T>
where
    T: StreamRefPayload,
{
    stream_ref_id: StreamRefId,
    settings: StreamRefSettings,
    state: Mutex<ConsumerState<T>>,
    changed: Condvar,
    outbound_wake: OutboundWake,
}

struct ConsumerState<T> {
    source_taken: bool,
    subscribed: bool,
    queue: VecDeque<T>,
    direct_terminal: bool,
    direct_consumer: Option<Box<dyn TerminalSinkConsumerDyn<T>>>,
    direct_cancelled: Option<Arc<std::sync::atomic::AtomicBool>>,
    terminal: Option<ConsumerTerminal>,
    expected_seq: u64,
    delivered: u64,
    cumulative_demand: u64,
    outbound: VecDeque<StreamRefMessage>,
    finish_after_outbound_ack: bool,
    waiting_cancel_ack: bool,
    done: bool,
}

#[derive(Clone)]
enum ConsumerTerminal {
    Complete,
    Error(StreamError),
}

struct StreamRefProtoTerminalHook<T>
where
    T: StreamRefPayload,
{
    shared: Arc<ConsumerShared<T>>,
    stream: Mutex<Option<ConsumerStream<T>>>,
}

impl<T> TerminalSourceHookDyn<T> for StreamRefProtoTerminalHook<T>
where
    T: StreamRefPayload,
{
    fn drain_terminal_batch(
        &self,
        materializer: &Materializer,
        cancelled: &Arc<std::sync::atomic::AtomicBool>,
        batch: &mut Vec<T>,
    ) -> StreamResult<TerminalSourceStatus> {
        batch.clear();
        if materializer.is_shutdown() {
            self.shared.cancel_from_downstream();
            return Err(StreamError::AbruptTermination);
        }
        if cancelled.load(Ordering::SeqCst) {
            self.shared.cancel_from_downstream();
            return Err(StreamError::Cancelled);
        }

        let mut stream = self
            .stream
            .lock()
            .unwrap_or_else(|poison| poison.into_inner());
        if stream.is_none() {
            *stream = Some(self.shared.start_stream()?);
        }
        let stream = stream.as_mut().expect("terminal stream present");
        for _ in 0..self.shared.settings.buffer_capacity().max(1) {
            match stream.next_item()? {
                Some(item) => batch.push(item),
                None => return Ok(TerminalSourceStatus::Completed),
            }
            if batch.len() >= 64 {
                break;
            }
        }
        Ok(TerminalSourceStatus::Active)
    }

    fn supports_direct_terminal(&self) -> bool {
        true
    }

    fn try_register_direct_terminal(
        &self,
        consumer: Box<dyn TerminalSinkConsumerDyn<T>>,
        cancelled: Arc<std::sync::atomic::AtomicBool>,
    ) -> Option<StreamResult<()>> {
        Some(self.shared.start_direct_terminal(consumer, cancelled))
    }

    fn cancel_terminal(&self) {
        self.shared.cancel_from_downstream();
    }
}

impl<T> ConsumerShared<T>
where
    T: StreamRefPayload,
{
    fn lock_state(&self) -> MutexGuard<'_, ConsumerState<T>> {
        self.state
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn notify_changed(&self) {
        self.changed.notify_all();
        self.outbound_wake.wake();
    }

    fn frame(&self, message: StreamRefMessage) -> StreamRefFrame {
        StreamRefFrame::new(self.stream_ref_id, message)
    }

    fn start_stream(self: &Arc<Self>) -> StreamResult<ConsumerStream<T>> {
        self.start_subscription()?;
        Ok(ConsumerStream {
            shared: Arc::clone(self),
            terminated: false,
        })
    }

    fn start_subscription(self: &Arc<Self>) -> StreamResult<()> {
        {
            let mut state = self.lock_state();
            if state.source_taken {
                return Err(StreamError::Failed(
                    "stream ref source has already been materialized".to_owned(),
                ));
            }
            state.source_taken = true;
            if !state.subscribed {
                state.subscribed = true;
                state
                    .outbound
                    .push_back(StreamRefMessage::OnSubscribeHandshake);
                if let Some(demand) = next_demand(&mut state, self.settings) {
                    state
                        .outbound
                        .push_back(StreamRefMessage::CumulativeDemand { seq_nr: demand });
                }
            }
            self.notify_changed();
        }
        Ok(())
    }

    fn start_direct_terminal(
        self: &Arc<Self>,
        consumer: Box<dyn TerminalSinkConsumerDyn<T>>,
        cancelled: Arc<std::sync::atomic::AtomicBool>,
    ) -> StreamResult<()> {
        let mut finish = None;
        {
            let mut state = self.lock_state();
            if state.source_taken {
                return Err(StreamError::Failed(
                    "stream ref source has already been materialized".to_owned(),
                ));
            }
            state.source_taken = true;
            state.direct_terminal = true;
            state.direct_cancelled = Some(cancelled);
            state.direct_consumer = Some(consumer);
            if let Some(terminal) = state.terminal.clone() {
                finish = state
                    .direct_consumer
                    .take()
                    .map(|consumer| (consumer, terminal_result(terminal)));
            } else if state.done {
                finish = state
                    .direct_consumer
                    .take()
                    .map(|consumer| (consumer, Err(StreamError::AbruptTermination)));
            } else if !state.subscribed {
                state.subscribed = true;
                state
                    .outbound
                    .push_back(StreamRefMessage::OnSubscribeHandshake);
                if let Some(demand) = next_demand(&mut state, self.settings) {
                    state
                        .outbound
                        .push_back(StreamRefMessage::CumulativeDemand { seq_nr: demand });
                }
            }
            self.notify_changed();
        }
        if let Some((consumer, result)) = finish {
            consumer.finish(result);
        }
        Ok(())
    }

    fn next_frame(&self) -> Option<StreamResult<StreamRefFrame>> {
        loop {
            let mut state = self.lock_state();
            if let Some(message) = state.outbound.pop_front() {
                let finish_after_ack = message.is_ack() && state.finish_after_outbound_ack;
                if finish_after_ack {
                    state.done = true;
                }
                drop(state);
                return Some(Ok(self.frame(message)));
            }
            if state.done {
                return None;
            }
            let next = wait_unpoison(&self.changed, state);
            drop(next);
        }
    }

    fn try_next_frame(&self) -> StreamRefOutboundPoll {
        let mut state = self.lock_state();
        if let Some(message) = state.outbound.pop_front() {
            let finish_after_ack = message.is_ack() && state.finish_after_outbound_ack;
            if finish_after_ack {
                state.done = true;
            }
            drop(state);
            StreamRefOutboundPoll::Ready(Ok(StreamRefOutbound::Frame(self.frame(message))))
        } else if state.done {
            StreamRefOutboundPoll::Closed
        } else {
            StreamRefOutboundPoll::Pending
        }
    }

    fn handle_frame(&self, frame: StreamRefFrame) -> StreamResult<()> {
        self.validate_frame_id(frame.stream_ref_id)?;
        match frame.message {
            StreamRefMessage::OnSubscribeHandshake => Ok(()),
            StreamRefMessage::SequencedOnNext { seq_nr, payload } => {
                let item = T::decode_stream_ref_payload(payload.bytes)?;
                let demand_ceiling = self.lock_state().cumulative_demand;
                self.handle_decoded_on_next(seq_nr, item, demand_ceiling)
            }
            StreamRefMessage::RemoteStreamCompleted { seq_nr } => {
                let finish = self.handle_remote_completed(seq_nr);
                if let Some((consumer, result)) = finish {
                    consumer.finish(result);
                }
                Ok(())
            }
            StreamRefMessage::RemoteStreamFailure { cause } => {
                let error = StreamError::Failed(String::from_utf8_lossy(&cause).into_owned());
                let finish = self.handle_remote_failure(error);
                if let Some((consumer, result)) = finish {
                    consumer.finish(result);
                }
                Ok(())
            }
            StreamRefMessage::Ack => {
                let mut state = self.lock_state();
                if state.waiting_cancel_ack {
                    state.waiting_cancel_ack = false;
                    state.done = true;
                }
                self.notify_changed();
                drop(state);
                Ok(())
            }
            StreamRefMessage::CumulativeDemand { .. } => Err(StreamError::Failed(
                "consumer endpoint cannot receive CumulativeDemand".to_owned(),
            )),
        }
    }

    fn handle_sequenced_on_next_batch(
        &self,
        stream_ref_id: StreamRefId,
        first_seq_nr: u64,
        payloads: &[&[u8]],
    ) -> StreamResult<()> {
        self.validate_frame_id(stream_ref_id)?;
        if self.lock_state().direct_terminal {
            return self.handle_direct_sequenced_on_next_batch(first_seq_nr, payloads);
        }
        let mut state = self.lock_state();
        for (index, payload) in payloads.iter().enumerate() {
            if state.terminal.is_some() || state.done {
                break;
            }
            let seq_nr = first_seq_nr.checked_add(index as u64).ok_or_else(|| {
                StreamError::Failed("stream ref batch sequence overflow".to_owned())
            })?;
            let item = T::decode_stream_ref_payload_slice(payload)?;
            self.handle_decoded_on_next_locked(&mut state, seq_nr, item);
        }
        self.notify_changed();
        drop(state);
        Ok(())
    }

    fn handle_direct_sequenced_on_next_batch(
        &self,
        first_seq_nr: u64,
        payloads: &[&[u8]],
    ) -> StreamResult<()> {
        if payloads.is_empty() {
            return Ok(());
        }
        let count = payloads.len() as u64;
        let last_seq_nr = first_seq_nr
            .checked_add(count - 1)
            .ok_or_else(|| StreamError::Failed("stream ref batch sequence overflow".to_owned()))?;

        let mut consumer = {
            let mut state = self.lock_state();
            if state.terminal.is_some() || state.done {
                return Ok(());
            }
            if state
                .direct_cancelled
                .as_ref()
                .is_some_and(|cancelled| cancelled.load(Ordering::SeqCst))
            {
                let error = StreamError::Cancelled;
                self.fail_consumer_locked(&mut state, error.clone());
                let consumer = state.direct_consumer.take();
                self.notify_changed();
                drop(state);
                if let Some(consumer) = consumer {
                    consumer.finish(Err(error));
                }
                return Ok(());
            }

            let error = if first_seq_nr != state.expected_seq {
                Some(invalid_sequence_error(
                    state.expected_seq,
                    first_seq_nr,
                    "stream ref element",
                ))
            } else if last_seq_nr >= state.cumulative_demand {
                Some(StreamError::Failed(
                    "stream ref receive buffer overflowed demand window".to_owned(),
                ))
            } else {
                None
            };
            if let Some(error) = error {
                self.fail_consumer_locked(&mut state, error.clone());
                let consumer = state.direct_consumer.take();
                self.notify_changed();
                drop(state);
                if let Some(consumer) = consumer {
                    consumer.finish(Err(error));
                }
                return Ok(());
            }

            state.expected_seq = state.expected_seq.saturating_add(count);
            match state.direct_consumer.take() {
                Some(consumer) => consumer,
                None => return Ok(()),
            }
        };

        let mut consumed = 0_u64;
        let mut consume_result = Ok(());
        for payload in payloads {
            match T::decode_stream_ref_payload_slice(payload)
                .and_then(|item| consumer.on_item(item))
            {
                Ok(()) => consumed = consumed.saturating_add(1),
                Err(error) => {
                    consume_result = Err(error);
                    break;
                }
            }
        }

        let finish = {
            let mut state = self.lock_state();
            if let Some(terminal) = state.terminal.clone() {
                Some((consumer, terminal_result(terminal)))
            } else if state.done {
                Some((consumer, Err(StreamError::AbruptTermination)))
            } else {
                match consume_result {
                    Ok(()) => {
                        debug_assert_eq!(consumed, count);
                        for _ in 0..count {
                            state.delivered = state.delivered.saturating_add(1);
                            if let Some(demand) = next_demand(&mut state, self.settings) {
                                state
                                    .outbound
                                    .push_back(StreamRefMessage::CumulativeDemand {
                                        seq_nr: demand,
                                    });
                            }
                        }
                        state.direct_consumer = Some(consumer);
                        None
                    }
                    Err(error) => {
                        self.fail_consumer_locked(&mut state, error.clone());
                        Some((consumer, Err(error)))
                    }
                }
            }
        };
        self.notify_changed();
        if let Some((consumer, result)) = finish {
            consumer.finish(result);
        }
        Ok(())
    }

    fn handle_decoded_on_next(
        &self,
        seq_nr: u64,
        item: T,
        demand_ceiling: u64,
    ) -> StreamResult<()> {
        if self.lock_state().direct_terminal {
            self.handle_direct_decoded_on_next(seq_nr, item, demand_ceiling);
        } else {
            let mut state = self.lock_state();
            self.handle_decoded_on_next_locked(&mut state, seq_nr, item);
            self.notify_changed();
            drop(state);
        }
        Ok(())
    }

    fn handle_decoded_on_next_locked(&self, state: &mut ConsumerState<T>, seq_nr: u64, item: T) {
        if state.terminal.is_some() || state.done {
            return;
        }
        if seq_nr != state.expected_seq {
            let error = invalid_sequence_error(state.expected_seq, seq_nr, "stream ref element");
            self.fail_consumer_locked(state, error);
        } else if state.queue.len() >= self.settings.buffer_capacity() {
            self.fail_consumer_locked(
                state,
                StreamError::Failed(
                    "stream ref receive buffer overflowed demand window".to_owned(),
                ),
            );
        } else {
            state.expected_seq = state.expected_seq.saturating_add(1);
            state.queue.push_back(item);
        }
    }

    fn handle_direct_decoded_on_next(&self, seq_nr: u64, item: T, demand_ceiling: u64) {
        let mut consumer = {
            let mut state = self.lock_state();
            if state.terminal.is_some() || state.done {
                return;
            }
            if state
                .direct_cancelled
                .as_ref()
                .is_some_and(|cancelled| cancelled.load(Ordering::SeqCst))
            {
                let error = StreamError::Cancelled;
                self.fail_consumer_locked(&mut state, error.clone());
                let consumer = state.direct_consumer.take();
                self.notify_changed();
                drop(state);
                if let Some(consumer) = consumer {
                    consumer.finish(Err(error));
                }
                return;
            }
            let error = if seq_nr != state.expected_seq {
                Some(invalid_sequence_error(
                    state.expected_seq,
                    seq_nr,
                    "stream ref element",
                ))
            } else if seq_nr >= demand_ceiling {
                Some(StreamError::Failed(
                    "stream ref receive buffer overflowed demand window".to_owned(),
                ))
            } else {
                None
            };
            if let Some(error) = error {
                self.fail_consumer_locked(&mut state, error.clone());
                let consumer = state.direct_consumer.take();
                self.notify_changed();
                drop(state);
                if let Some(consumer) = consumer {
                    consumer.finish(Err(error));
                }
                return;
            }
            state.expected_seq = state.expected_seq.saturating_add(1);
            match state.direct_consumer.take() {
                Some(consumer) => consumer,
                None => return,
            }
        };

        let consume_result = consumer.on_item(item);
        let finish = {
            let mut state = self.lock_state();
            if let Some(terminal) = state.terminal.clone() {
                Some((consumer, terminal_result(terminal)))
            } else if state.done {
                Some((consumer, Err(StreamError::AbruptTermination)))
            } else {
                match consume_result {
                    Ok(()) => {
                        state.delivered = state.delivered.saturating_add(1);
                        if let Some(demand) = next_demand(&mut state, self.settings) {
                            state
                                .outbound
                                .push_back(StreamRefMessage::CumulativeDemand { seq_nr: demand });
                        }
                        state.direct_consumer = Some(consumer);
                        None
                    }
                    Err(error) => {
                        self.fail_consumer_locked(&mut state, error.clone());
                        Some((consumer, Err(error)))
                    }
                }
            }
        };
        self.notify_changed();
        if let Some((consumer, result)) = finish {
            consumer.finish(result);
        }
    }

    fn handle_remote_completed(
        &self,
        seq_nr: u64,
    ) -> Option<(Box<dyn TerminalSinkConsumerDyn<T>>, StreamResult<()>)> {
        let mut finish = None;
        let mut state = self.lock_state();
        if state.terminal.is_none() && !state.done {
            if seq_nr != state.expected_seq {
                state.queue.clear();
                let error =
                    invalid_sequence_error(state.expected_seq, seq_nr, "stream ref completion");
                state.terminal = Some(ConsumerTerminal::Error(error.clone()));
                finish = state
                    .direct_consumer
                    .take()
                    .map(|consumer| (consumer, Err(error)));
            } else {
                state.terminal = Some(ConsumerTerminal::Complete);
                finish = state
                    .direct_consumer
                    .take()
                    .map(|consumer| (consumer, Ok(())));
            }
            state.outbound.push_back(StreamRefMessage::Ack);
            state.finish_after_outbound_ack = true;
        }
        self.notify_changed();
        drop(state);
        finish
    }

    fn handle_remote_failure(
        &self,
        error: StreamError,
    ) -> Option<(Box<dyn TerminalSinkConsumerDyn<T>>, StreamResult<()>)> {
        let mut state = self.lock_state();
        let mut finish = None;
        if state.terminal.is_none() && !state.done {
            state.queue.clear();
            state.terminal = Some(ConsumerTerminal::Error(error.clone()));
            finish = state
                .direct_consumer
                .take()
                .map(|consumer| (consumer, Err(error)));
            state.outbound.push_back(StreamRefMessage::Ack);
            state.finish_after_outbound_ack = true;
        }
        self.notify_changed();
        drop(state);
        finish
    }

    fn fail_consumer_locked(&self, state: &mut ConsumerState<T>, error: StreamError) {
        state.queue.clear();
        state.terminal = Some(ConsumerTerminal::Error(error.clone()));
        state
            .outbound
            .push_back(StreamRefMessage::RemoteStreamFailure {
                cause: failure_cause(&error),
            });
        state.waiting_cancel_ack = true;
    }

    fn cancel_from_downstream(&self) {
        let mut finish = None;
        let mut state = self.lock_state();
        if state.terminal.is_none() && !state.done {
            let seq_nr = state.expected_seq;
            let error = StreamError::Cancelled;
            state.terminal = Some(ConsumerTerminal::Error(error.clone()));
            finish = state
                .direct_consumer
                .take()
                .map(|consumer| (consumer, Err(error)));
            state
                .outbound
                .push_back(StreamRefMessage::RemoteStreamCompleted { seq_nr });
            state.waiting_cancel_ack = true;
        }
        self.notify_changed();
        drop(state);
        if let Some((consumer, result)) = finish {
            consumer.finish(result);
        }
    }

    fn fail_connection(&self, error: StreamError) {
        let mut finish = None;
        let mut state = self.lock_state();
        if state.terminal.is_none() {
            state.queue.clear();
            state.terminal = Some(ConsumerTerminal::Error(error.clone()));
            finish = state
                .direct_consumer
                .take()
                .map(|consumer| (consumer, Err(error.clone())));
        }
        state.done = true;
        self.notify_changed();
        drop(state);
        if let Some((consumer, result)) = finish {
            consumer.finish(result);
        }
    }

    fn validate_frame_id(&self, stream_ref_id: StreamRefId) -> StreamResult<()> {
        if stream_ref_id == self.stream_ref_id {
            Ok(())
        } else {
            Err(StreamError::Failed(format!(
                "stream ref id mismatch: expected {}, got {}",
                self.stream_ref_id, stream_ref_id
            )))
        }
    }
}

struct ConsumerStream<T>
where
    T: StreamRefPayload,
{
    shared: Arc<ConsumerShared<T>>,
    terminated: bool,
}

impl<T> ConsumerStream<T>
where
    T: StreamRefPayload,
{
    fn next_item(&mut self) -> StreamResult<Option<T>> {
        if self.terminated {
            return Ok(None);
        }
        loop {
            let mut state = self.shared.lock_state();
            if let Some(item) = state.queue.pop_front() {
                state.delivered = state.delivered.saturating_add(1);
                if let Some(demand) = next_demand(&mut state, self.shared.settings) {
                    state
                        .outbound
                        .push_back(StreamRefMessage::CumulativeDemand { seq_nr: demand });
                    self.shared.notify_changed();
                }
                return Ok(Some(item));
            }

            if let Some(terminal) = state.terminal.clone() {
                self.terminated = true;
                return match terminal {
                    ConsumerTerminal::Complete => Ok(None),
                    ConsumerTerminal::Error(error) => Err(error),
                };
            }

            let next = wait_unpoison(&self.shared.changed, state);
            drop(next);
        }
    }

    fn close(&mut self) {
        if !self.terminated {
            self.shared.cancel_from_downstream();
            self.terminated = true;
        }
    }
}

impl<T> Iterator for ConsumerStream<T>
where
    T: StreamRefPayload,
{
    type Item = StreamResult<T>;

    fn next(&mut self) -> Option<Self::Item> {
        match self.next_item() {
            Ok(Some(item)) => Some(Ok(item)),
            Ok(None) => None,
            Err(error) => Some(Err(error)),
        }
    }
}

impl<T> Drop for ConsumerStream<T>
where
    T: StreamRefPayload,
{
    fn drop(&mut self) {
        self.close();
    }
}

fn next_demand<T>(state: &mut ConsumerState<T>, settings: StreamRefSettings) -> Option<u64> {
    // Akka redelivers CumulativeDemand because some remoting carriers are
    // lossy. This seam is driven by reliable ordered carriers such as one QUIC
    // bidirectional stream, so a larger cumulative ceiling is sent once.
    if state.terminal.is_some() {
        return None;
    }
    let remaining_credit = state.cumulative_demand.saturating_sub(state.delivered);
    if state.cumulative_demand != 0 && remaining_credit > demand_replenish_threshold(settings) {
        return None;
    }
    let target = state
        .delivered
        .saturating_add(settings.buffer_capacity() as u64);
    if state.cumulative_demand >= target {
        return None;
    }
    state.cumulative_demand = target;
    Some(target)
}

fn demand_replenish_threshold(settings: StreamRefSettings) -> u64 {
    (settings.buffer_capacity() as u64) / 2
}

fn terminal_result(terminal: ConsumerTerminal) -> StreamResult<()> {
    match terminal {
        ConsumerTerminal::Complete => Ok(()),
        ConsumerTerminal::Error(error) => Err(error),
    }
}

fn failure_cause(error: &StreamError) -> Vec<u8> {
    match error {
        StreamError::Failed(message) => message.clone().into_bytes(),
        other => other.to_string().into_bytes(),
    }
}

fn subscription_timeout_error(side: &str) -> StreamError {
    StreamError::Failed(format!(
        "{side} remote side did not subscribe within subscription timeout"
    ))
}

fn invalid_sequence_error(expected: u64, got: u64, context: &str) -> StreamError {
    StreamError::Failed(format!(
        "{context} sequence gap: expected sequence {expected}, got {got}"
    ))
}

fn deadline_from_now(timeout: Duration) -> Instant {
    Instant::now()
        .checked_add(timeout)
        .unwrap_or_else(far_future)
}

fn far_future() -> Instant {
    Instant::now() + Duration::from_secs(60 * 60 * 24 * 365)
}

fn wait_timeout_unpoison<'a, T>(
    condvar: &Condvar,
    guard: MutexGuard<'a, T>,
    timeout: Duration,
) -> (MutexGuard<'a, T>, std::sync::WaitTimeoutResult) {
    condvar
        .wait_timeout(guard, timeout)
        .unwrap_or_else(|poison| poison.into_inner())
}

fn wait_unpoison<'a, T>(condvar: &Condvar, guard: MutexGuard<'a, T>) -> MutexGuard<'a, T> {
    condvar
        .wait(guard)
        .unwrap_or_else(|poison| poison.into_inner())
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;
    use crate::{Sink, Source, StreamRefs};

    fn short_settings() -> StreamRefSettings {
        StreamRefSettings::default()
            .with_buffer_capacity(4)
            .with_subscription_timeout(Duration::from_millis(50))
    }

    #[test]
    fn protobuf_frame_round_trip() {
        let frame = StreamRefFrame::new(
            StreamRefId::from_u128(42),
            StreamRefMessage::SequencedOnNext {
                seq_nr: 7,
                payload: StreamRefPayloadBytes {
                    bytes: 99_u64.encode_stream_ref_payload(),
                },
            },
        );

        let decoded = StreamRefFrame::decode(&frame.encode_to_vec()).unwrap();
        assert_eq!(decoded, frame);
    }

    #[test]
    fn producer_consumer_seam_streams_with_low_watermark_demand() {
        let id = StreamRefId::from_u128(1);
        let settings = short_settings();
        let source_ref = Source::from_iter(0_u64..10)
            .run_with(StreamRefs::source_ref_with_settings(settings))
            .unwrap();
        let producer = StreamRefProtoProducer::from_source_ref(source_ref, id, settings).unwrap();
        let consumer = StreamRefProtoConsumer::<u64>::new(id, settings);
        let consumer_source = consumer.source();

        let producer_thread = std::thread::spawn({
            let producer = producer.clone();
            let consumer = consumer.clone();
            move || {
                while let Some(frame) = producer.next_frame() {
                    consumer.handle_frame(frame?)?;
                }
                Ok::<_, StreamError>(())
            }
        });
        let consumer_thread = std::thread::spawn({
            let producer = producer.clone();
            let consumer = consumer.clone();
            move || {
                while let Some(frame) = consumer.next_frame() {
                    producer.handle_frame(frame?)?;
                }
                Ok::<_, StreamError>(())
            }
        });

        assert_eq!(
            consumer_source.run_collect().unwrap(),
            (0_u64..10).collect::<Vec<_>>()
        );
        producer_thread.join().unwrap().unwrap();
        consumer_thread.join().unwrap().unwrap();
    }

    #[test]
    fn strict_sequence_gap_fails_consumer_and_sends_failure() {
        let id = StreamRefId::from_u128(2);
        let consumer = StreamRefProtoConsumer::<u64>::new(id, short_settings());
        let source = consumer
            .source()
            .run_with(crate::testkit::TestSink::probe())
            .unwrap();
        source.request(1);
        consumer.next_frame().unwrap().unwrap();
        consumer.next_frame().unwrap().unwrap();

        consumer
            .handle_frame(StreamRefFrame::new(
                id,
                StreamRefMessage::SequencedOnNext {
                    seq_nr: 1,
                    payload: StreamRefPayloadBytes {
                        bytes: 1_u64.encode_stream_ref_payload(),
                    },
                },
            ))
            .unwrap();

        let outbound = consumer.next_frame().unwrap().unwrap();
        assert!(matches!(
            outbound.message,
            StreamRefMessage::RemoteStreamFailure { .. }
        ));
        assert!(matches!(source.expect_error(), StreamError::Failed(_)));
    }

    #[test]
    fn direct_terminal_sequence_gap_uses_shared_failure_path() {
        let id = StreamRefId::from_u128(23);
        let consumer = StreamRefProtoConsumer::<u64>::new(id, short_settings());
        let completion = consumer
            .source()
            .run_with(Sink::fold(0_u64, |acc, item| acc + item))
            .unwrap();
        consumer.next_frame().unwrap().unwrap();
        consumer.next_frame().unwrap().unwrap();

        consumer
            .handle_frame(StreamRefFrame::new(
                id,
                StreamRefMessage::SequencedOnNext {
                    seq_nr: 1,
                    payload: StreamRefPayloadBytes {
                        bytes: 1_u64.encode_stream_ref_payload(),
                    },
                },
            ))
            .unwrap();

        let outbound = consumer.next_frame().unwrap().unwrap();
        assert!(matches!(
            outbound.message,
            StreamRefMessage::RemoteStreamFailure { .. }
        ));
        assert!(
            matches!(completion.wait(), Err(StreamError::Failed(message)) if message.contains("sequence gap"))
        );
    }

    #[test]
    fn direct_terminal_batch_over_demand_ceiling_fails_consumer() {
        let id = StreamRefId::from_u128(24);
        let settings = short_settings();
        let consumer = StreamRefProtoConsumer::<u64>::new(id, settings);
        let completion = consumer
            .source()
            .run_with(Sink::fold(0_u64, |acc, item| acc + item))
            .unwrap();
        consumer.next_frame().unwrap().unwrap();
        consumer.next_frame().unwrap().unwrap();

        let payloads = (0_u64..=settings.buffer_capacity() as u64)
            .map(u64::encode_stream_ref_payload)
            .collect::<Vec<_>>();
        let payload_slices = payloads.iter().map(Vec::as_slice).collect::<Vec<&[u8]>>();
        consumer
            .handle_sequenced_on_next_batch(id, 0, &payload_slices)
            .unwrap();

        let mut saw_failure = false;
        for _ in 0..4 {
            let Some(outbound) = consumer.next_frame() else {
                break;
            };
            if matches!(
                outbound.unwrap().message,
                StreamRefMessage::RemoteStreamFailure { .. }
            ) {
                saw_failure = true;
                break;
            }
        }
        assert!(saw_failure);
        assert!(
            matches!(completion.wait(), Err(StreamError::Failed(message)) if message.contains("demand window"))
        );
    }

    #[test]
    fn producer_batches_ready_elements_and_preserves_completion_order() {
        let id = StreamRefId::from_u128(20);
        let settings = StreamRefSettings::default().with_buffer_capacity(8);
        let producer =
            StreamRefProtoProducer::from_source(Source::from_iter(0_u64..6), id, settings).unwrap();
        producer
            .handle_frame(StreamRefFrame::new(
                id,
                StreamRefMessage::CumulativeDemand { seq_nr: 8 },
            ))
            .unwrap();

        let first = producer.next_outbound(4, usize::MAX).unwrap().unwrap();
        let StreamRefOutbound::SequencedBatch(first) = first else {
            panic!("expected first data batch");
        };
        assert_eq!(first.first_seq_nr(), 0);
        assert_eq!(first.count(), 4);
        for index in 0..first.count() {
            assert_eq!(
                u64::decode_stream_ref_payload_slice(first.payload(index)).unwrap(),
                index as u64
            );
        }

        let second = producer.next_outbound(4, usize::MAX).unwrap().unwrap();
        let StreamRefOutbound::SequencedBatch(second) = second else {
            panic!("expected second data batch");
        };
        assert_eq!(second.first_seq_nr(), 4);
        assert_eq!(second.count(), 2);

        let completion = producer.next_outbound(4, usize::MAX).unwrap().unwrap();
        assert!(matches!(
            completion,
            StreamRefOutbound::Frame(StreamRefFrame {
                message: StreamRefMessage::RemoteStreamCompleted { seq_nr: 6 },
                ..
            })
        ));
    }

    #[test]
    fn consumer_batch_ingress_preserves_order() {
        let id = StreamRefId::from_u128(21);
        let consumer = StreamRefProtoConsumer::<u64>::new(id, short_settings());
        let probe = consumer
            .source()
            .run_with(crate::testkit::TestSink::probe())
            .unwrap();
        probe.request(3);
        consumer.next_frame().unwrap().unwrap();
        consumer.next_frame().unwrap().unwrap();

        let payloads = [10_u64, 11, 12]
            .into_iter()
            .map(u64::encode_stream_ref_payload)
            .collect::<Vec<_>>();
        let payload_slices = payloads.iter().map(Vec::as_slice).collect::<Vec<&[u8]>>();
        consumer
            .handle_sequenced_on_next_batch(id, 0, &payload_slices)
            .unwrap();

        assert_eq!(probe.expect_next(), 10);
        assert_eq!(probe.expect_next(), 11);
        assert_eq!(probe.expect_next(), 12);
    }

    #[test]
    fn consumer_batch_sequence_gap_uses_shared_failure_path() {
        let id = StreamRefId::from_u128(22);
        let consumer = StreamRefProtoConsumer::<u64>::new(id, short_settings());
        let source = consumer
            .source()
            .run_with(crate::testkit::TestSink::probe())
            .unwrap();
        source.request(1);
        consumer.next_frame().unwrap().unwrap();
        consumer.next_frame().unwrap().unwrap();

        let payload = 1_u64.encode_stream_ref_payload();
        consumer
            .handle_sequenced_on_next_batch(id, 1, &[payload.as_slice()])
            .unwrap();

        let outbound = consumer.next_frame().unwrap().unwrap();
        assert!(matches!(
            outbound.message,
            StreamRefMessage::RemoteStreamFailure { .. }
        ));
        assert!(matches!(source.expect_error(), StreamError::Failed(_)));
    }

    #[test]
    fn producer_times_out_without_first_demand() {
        let producer = StreamRefProtoProducer::from_source(
            Source::repeat(1_u64),
            StreamRefId::from_u128(3),
            short_settings(),
        )
        .unwrap();

        let error = producer.next_frame().unwrap().unwrap_err();
        assert!(matches!(error, StreamError::Failed(message) if message.contains("first demand")));
    }

    #[test]
    fn demand_redelivery_is_not_required_by_reliable_carriers() {
        // Akka redelivers CumulativeDemand because Artery/Aeron may lose messages.
        // The protobuf seam is intended for reliable ordered carriers such as a
        // single QUIC bidirectional stream, so each cumulative ceiling is sent
        // once and remains valid until a larger ceiling replaces it.
        assert_eq!(
            StreamRefSettings::default().demand_redelivery_interval(),
            Duration::from_secs(1)
        );
    }
}