liminal-sdk 0.19.5

//! The windowless subscription socket reader and its retained terminal.

use super::subscription_wire::next_frame;
use super::{DeliveredMessage, Frame, Mailbox, SdkError, TcpStream, ToString, Vec, format};
use crate::remote::mailbox::{ReaderEnd, offer_or_end};

/// Background loop: drains the socket, surfacing each `Deliver` frame's message in
/// `mailbox`, and returns the end the consumer receives after them.
///
/// The socket carries no read window here, so the loop blocks until the server
/// sends or the connection ends: nothing wakes it on a timer and nothing sweeps.
/// A full mailbox holds the loop inside its offer, so the socket is not read
/// until the consumer makes room. It returns (ending the thread) when the
/// connection closes — including the `shutdown` teardown performs — when a
/// `Disconnect` or `SubscribeError` arrives, when a delivery could never fit
/// the mailbox, or on a fatal decode/IO error, carrying that reason.
///
/// `buffer` is seeded with the setup residue (see [`SubscriptionStream::open`]): any
/// `Deliver` bytes the synchronous subscribe read past the `SubscribeAck` are
/// already here, so the loop decodes them first — before its next socket read —
/// instead of losing them and starting mid-stream.
pub(super) fn run_reader(
    mut stream: TcpStream,
    mut buffer: Vec<u8>,
    mailbox: &Mailbox<DeliveredMessage>,
) -> ReaderEnd {
    loop {
        // Connection closed or a fatal read/decode error: end the thread,
        // carrying the error itself to the consumer.
        let frame = match next_frame(&mut stream, &mut buffer) {
            Ok(frame) => frame,
            Err(error) => return ReaderEnd::Failed(error),
        };
        match frame {
            Frame::Deliver {
                delivery_seq,
                envelope,
                message,
                ..
            } => {
                let message = DeliveredMessage {
                    delivery_seq,
                    schema_id: envelope.schema_id,
                    payload: envelope.payload,
                    message,
                };
                if let Some(end) = offer_or_end(mailbox, message) {
                    return end;
                }
            }
            // A server `Disconnect` ends the subscription cleanly, and the
            // consumer is told so by name.
            Frame::Disconnect { .. } => {
                return ReaderEnd::Failed(SdkError::Connection {
                    description: "subscription receive failed: the server disconnected the \
                                  subscription"
                        .to_string(),
                });
            }
            // A `SubscribeError` arriving AFTER setup is the server ending this
            // subscription -- the overflow shed sends exactly this and then
            // releases the subscription at the channel actor, so no further
            // delivery can ever arrive. It is surfaced to the consumer and ends
            // the reader (P0 #55).
            //
            // This is the one exception to the stray-frame rule below, and the
            // distinction is deliveries: ignoring a stray frame protects the
            // deliveries still to come, and here there are none. Dropping this
            // frame is what left a shed subscriber unable to tell "the server
            // dropped me" from "nothing was published".
            Frame::SubscribeError {
                reason_code,
                message,
                ..
            } => {
                // A consumer already gone is counted by the mailbox's close.
                return ReaderEnd::Failed(subscription_ended(reason_code, message));
            }
            // Any other frame on a subscription connection is unexpected; it is
            // set aside and counted rather than tearing the reader down, so a
            // stray frame cannot drop the deliveries behind it.
            _ => {
                if let Err(error) = mailbox.record_ignored_frame() {
                    return ReaderEnd::Failed(error);
                }
            }
        }
    }
}

/// Builds the typed terminal for a `SubscribeError` the server sent mid-stream.
///
/// The server's own detail is carried VERBATIM: it is the only text that says
/// which limit ended the subscription, and a client that paraphrased it would
/// leave an operator correlating a client log against a server log by guesswork.
fn subscription_ended(reason_code: u16, message: Option<alloc::string::String>) -> SdkError {
    SdkError::Protocol {
        description: format!(
            "server ended the subscription (reason {reason_code}): {}",
            message.unwrap_or_else(|| "no detail".to_string())
        ),
    }
}