liminal-sdk 0.19.0

//! Client-side subscription stream: the receive half of the delivery pump.
//!
//! Where [`PushClient`](super::push_client::PushClient) consumes server-initiated
//! *pushes*, a [`SubscriptionStream`] consumes server-initiated *deliveries*: the
//! server writes a [`Frame::Deliver`] on the subscription's stream every time a
//! message is published to the subscribed channel. This client owns a dedicated
//! connection whose socket is drained by a background reader thread that routes
//! each `Deliver` into a bounded mailbox the caller pulls with
//! [`SubscriptionStream::recv_timeout`]; a full mailbox stops the reader
//! reading the socket rather than dropping or holding without limit.
//!
//! # v1 shape
//!
//! One subscription per dedicated connection. Multiplexing several subscriptions
//! over one connection arrives with the v2 credit mode (which also adds explicit
//! per-delivery acks); until then a `SubscriptionStream` is a single channel
//! subscription bound to its own socket, mirroring the one-connection-per-role
//! shape the `PushClient` already uses.

use alloc::format;
use alloc::string::ToString;
use alloc::sync::Arc;
use alloc::vec::Vec;
use core::time::Duration;

use std::net::{Shutdown, TcpStream};
use std::thread::JoinHandle;

use liminal::protocol::{
    DeliveredId, Frame, MessageId, ProtocolError, SchemaId, WorkerRegistration,
};

use crate::SdkError;
use crate::remote::SETUP_TIMEOUT;
mod subscription_reader;
mod subscription_setup;
use subscription_reader::run_reader;
mod subscription_signals;
mod subscription_wire;
use crate::remote::mailbox::{Mailbox, MailboxItem, MailboxLimits, MailboxStats, MailboxWords};
use subscription_setup::{connect_socket, handshake, subscribe};
use subscription_wire::write_frame;

/// The subscription's words in the errors its mailbox hands the consumer.
const SUBSCRIPTION_WORDS: MailboxWords = MailboxWords {
    receive: "subscription receive failed",
    item: "delivery",
    reader: "subscription reader",
};

/// Bound on a single socket write.
const WRITE_TIMEOUT: Duration = Duration::from_secs(5);
/// Read chunk size used when draining the socket into the frame buffer.
const READ_CHUNK_BYTES: usize = 4096;
/// Upper bound on a single buffered frame, guarding against runaway buffering.
const MAX_FRAME_BYTES: usize = 64 * 1024 * 1024;
/// The single application stream this subscription's deliveries ride on. One
/// subscription per connection in v1, so a fixed stream id is sufficient.
const SUBSCRIPTION_STREAM_ID: u32 = 1;
/// A message the server delivered on this subscription.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct DeliveredMessage {
    delivery_seq: u64,
    schema_id: SchemaId,
    payload: Vec<u8>,
    /// The server's id for the message and its boot, when a 1.2 server sent it.
    message: Option<DeliveredId>,
}

impl MailboxItem for DeliveredMessage {
    fn payload_len(&self) -> usize {
        self.payload.len()
    }
}

impl DeliveredMessage {
    /// The per-subscription monotonic delivery sequence (starts at 1). The anchor
    /// the future ack/resume protocol will acknowledge against.
    #[must_use]
    pub const fn delivery_seq(&self) -> u64 {
        self.delivery_seq
    }

    /// The schema id the server selected for this subscription's stream.
    #[must_use]
    pub const fn schema_id(&self) -> SchemaId {
        self.schema_id
    }

    /// The delivered payload bytes.
    #[must_use]
    pub fn payload(&self) -> &[u8] {
        &self.payload
    }

    /// Consumes the message, returning the owned payload bytes.
    #[must_use]
    pub fn into_payload(self) -> Vec<u8> {
        self.payload
    }

    /// The server's id for this message and the boot it was drawn in, when the
    /// server sent it (protocol 1.2, Fable 4c). `None` from an older server,
    /// and for a durable message replayed from the store.
    #[must_use]
    pub const fn message(&self) -> Option<DeliveredId> {
        self.message
    }

    /// The causal parent a reply to this message names, `<boot>:<id>`, or
    /// `None` when the message carried no id: never an invented one.
    #[must_use]
    pub fn reply_parent(&self) -> Option<MessageId> {
        self.message.map(DeliveredId::reply_parent)
    }
}

/// A connected subscription whose background reader surfaces delivered messages.
///
/// Construct with [`SubscriptionStream::open`]; the background reader starts
/// immediately and runs until the stream is dropped. Pull delivered messages with
/// [`SubscriptionStream::recv_timeout`].
#[derive(Debug)]
pub struct SubscriptionStream {
    /// Write half, used only by setup and the best-effort teardown on drop.
    writer: TcpStream,
    /// Server-assigned subscription id, echoed on `Unsubscribe` at teardown.
    subscription_id: u64,
    /// Delivered messages surfaced by the background reader, bounded in count
    /// and bytes, then the one typed end the reader stopped with. A
    /// `SubscribeError` arriving mid-stream is the ONLY explanation the
    /// consumer will ever get for deliveries stopping, so it is that end,
    /// received after every delivery before it (P0 #55).
    mailbox: Arc<Mailbox<DeliveredMessage>>,
    /// Background reader handle, joined on drop.
    reader: Option<JoinHandle<()>>,
}

impl SubscriptionStream {
    /// Connects to `address`, performs the handshake, subscribes to `channel`, and
    /// starts the background reader that drains delivered messages.
    ///
    /// `accepted_schemas` is the client's schema-compatibility list; pass an empty
    /// vector to let the server select the channel's configured schema (the
    /// server's negotiation contract).
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when the TCP connection or socket
    /// configuration fails, and [`SdkError::Protocol`] when the handshake or
    /// subscribe is rejected, or the socket cannot be cloned for the reader thread.
    pub fn open(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
    ) -> Result<Self, SdkError> {
        Self::open_with_auth(address, channel, accepted_schemas, &[])
    }

    /// Connects, handshakes carrying `auth_token`, subscribes to `channel`, and
    /// starts the background reader.
    ///
    /// A subscription owns a dedicated connection (the v1 shape), so it presents
    /// its own credential in its own `Connect` frame; the token a
    /// request/response transport was built with lives on that transport's
    /// socket and cannot travel here. Additive to [`open`]: an empty token is
    /// exactly the open-access handshake `open` performs, so an ungated server
    /// sees byte-identical bytes either way.
    ///
    /// The server compares the token during the handshake and answers a
    /// mismatch with `ConnectError` before closing, which surfaces here as
    /// [`SdkError::Connection`].
    ///
    /// `accepted_schemas` is the client's schema-compatibility list; pass an
    /// empty vector to let the server select the channel's configured schema.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when the TCP connection or socket
    /// configuration fails or the token is rejected, and [`SdkError::Protocol`]
    /// when the subscribe is rejected, or the socket cannot be cloned for the
    /// reader thread.
    ///
    /// [`open`]: Self::open
    pub fn open_with_auth(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        auth_token: &[u8],
    ) -> Result<Self, SdkError> {
        Self::open_with_limits(
            address,
            channel,
            accepted_schemas,
            auth_token,
            MailboxLimits::SUBSCRIPTION,
        )
    }

    /// [`Self::open_with_auth`] with the mailbox bounds supplied.
    fn open_with_limits(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        auth_token: &[u8],
        limits: MailboxLimits,
    ) -> Result<Self, SdkError> {
        Self::open_configured(address, channel, accepted_schemas, auth_token, None, limits)
    }

    /// Connects, handshakes, registers the connection as the worker
    /// `registration` names, then subscribes to `channel`, so the server can
    /// tell which worker the subscription belongs to (a subscription owns its
    /// own connection, so the worker's push connection cannot say it).
    ///
    /// # Errors
    ///
    /// As [`open`], and [`SdkError::Protocol`] when the server rejects the
    /// registration (the reason is carried in the error).
    ///
    /// [`open`]: Self::open
    pub fn open_registered(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        registration: WorkerRegistration,
    ) -> Result<Self, SdkError> {
        Self::open_configured(
            address,
            channel,
            accepted_schemas,
            &[],
            Some(registration),
            MailboxLimits::SUBSCRIPTION,
        )
    }

    fn open_configured(
        address: &str,
        channel: &str,
        accepted_schemas: Vec<SchemaId>,
        auth_token: &[u8],
        registration: Option<WorkerRegistration>,
        limits: MailboxLimits,
    ) -> Result<Self, SdkError> {
        let mut stream = connect_socket(address)?;
        // A single buffer threads through the whole synchronous setup so any bytes
        // the setup reads past the control-frame reply are preserved. The server
        // may coalesce a `SubscribeAck` with the first `Deliver` frames into one TCP
        // segment (the delivery pump runs in the same slice that acks the
        // subscribe), and a socket read pulls up to `READ_CHUNK_BYTES` at once — so
        // this buffer can hold whole (or partial) `Deliver` frames after the ack.
        // Handing that residue to the reader thread is what keeps those deliveries
        // from being dropped and, worse, from desyncing a reader that would
        // otherwise start mid-frame on a fresh empty buffer.
        let mut buffer = Vec::new();
        handshake(&mut stream, &mut buffer, auth_token)?;
        if let Some(registration) = registration {
            super::push_client::register(&mut stream, registration, SETUP_TIMEOUT)?;
        }
        let subscription_id = subscribe(&mut stream, &mut buffer, channel, accepted_schemas)?;

        // The control exchange is over, so its deadline comes off: the reader
        // blocks on socket input with no read window at all. Teardown shuts the
        // socket down, which surfaces as a typed terminal — the socket signals,
        // nothing sweeps. A window left armed here would be a wake cadence in
        // steady state, which is the defect this retires, whatever period it
        // carried.
        stream
            .set_read_timeout(None)
            .map_err(|source| SdkError::Connection {
                description: format!("failed to clear the subscription read deadline: {source}"),
            })?;
        let read_stream = stream.try_clone().map_err(|source| SdkError::Protocol {
            description: format!("failed to clone subscription socket for reader thread: {source}"),
        })?;
        Self::start(stream, read_stream, buffer, subscription_id, limits)
    }

    /// Starts the background reader over a subscribed stream, its setup
    /// residue, and a mailbox with `limits`.
    fn start(
        stream: TcpStream,
        read_stream: TcpStream,
        buffer: Vec<u8>,
        subscription_id: u64,
        limits: MailboxLimits,
    ) -> Result<Self, SdkError> {
        let mailbox = Arc::new(Mailbox::new(limits, SUBSCRIPTION_WORDS));
        let reader_mailbox = Arc::clone(&mailbox);
        let reader = std::thread::Builder::new()
            .name("liminal-subscription-reader".to_string())
            .spawn(move || {
                let end = run_reader(read_stream, buffer, &reader_mailbox);
                reader_mailbox.close(end);
            })
            .map_err(|source| SdkError::Protocol {
                description: format!("failed to start subscription reader thread: {source}"),
            })?;

        Ok(Self {
            writer: stream,
            subscription_id,
            mailbox,
            reader: Some(reader),
        })
    }

    /// Blocks up to `timeout` for the next delivered message from the server.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when no message arrives within
    /// `timeout`. Once the reader has stopped and every held delivery has been
    /// taken, returns the error it stopped with: the server's mid-stream
    /// `SubscribeError`, its `Disconnect`, the read or decode failure itself,
    /// or [`SdkError::Backpressure`] naming a delivery too large for the
    /// mailbox.
    pub fn recv_timeout(&self, timeout: Duration) -> Result<DeliveredMessage, SdkError> {
        self.mailbox.take(timeout)
    }

    /// The mailbox counters: deliveries and bytes held now and at peak, how
    /// often the reader stopped reading for a consumer that had not made
    /// room, refused and stray frames, and errors no consumer was left to
    /// receive.
    ///
    /// # Errors
    ///
    /// Returns [`SdkError::Connection`] when the mailbox lock is poisoned.
    pub fn stats(&self) -> Result<MailboxStats, SdkError> {
        self.mailbox.stats()
    }

    /// The server-assigned id for this subscription.
    #[must_use]
    pub const fn subscription_id(&self) -> u64 {
        self.subscription_id
    }
}

impl Drop for SubscriptionStream {
    fn drop(&mut self) {
        // Best-effort clean teardown: tell the server to drop the subscription and
        // close the connection. Failures are ignored — the connection close alone
        // frees the server-side subscription when its subscriber process exits.
        let unsubscribe = Frame::Unsubscribe {
            flags: 0,
            stream_id: SUBSCRIPTION_STREAM_ID,
            subscription_id: self.subscription_id,
        };
        let _ = write_frame(&mut self.writer, &unsubscribe);
        let _ = write_frame(&mut self.writer, &Frame::Disconnect { flags: 0 });
        // Then TELL the reader. It blocks on socket input with no read window, so
        // nothing but the socket can end its wait — a stop flag it never wakes to
        // sample would be a lie about how it stops. Shutting the socket down
        // surfaces a typed terminal to the blocked reader, exactly as the
        // WebSocket sibling does, and the shutdown of the write half flushes the
        // frames just written before its FIN. The join is therefore bounded by the
        // shutdown, not by a peer's goodwill.
        let _ = self.writer.shutdown(Shutdown::Both);
        // Nothing receives from here on: release a reader waiting on a full
        // mailbox so the join cannot wait on a consumer that has gone.
        self.mailbox.release();
        if let Some(reader) = self.reader.take() {
            reader.join().ok();
        }
    }
}

#[cfg(test)]
use subscription_setup::CLIENT_MAX_VERSION;

#[cfg(test)]
use alloc::vec;
#[cfg(test)]
use liminal::protocol::{decode, encode, encoded_len};
#[cfg(test)]
use std::io::{Read, Write};
#[cfg(test)]
use subscription_wire::protocol_error;

#[cfg(test)]
#[path = "subscription_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "subscription_mailbox_tests.rs"]
mod mailbox_tests;