liminal-sdk 0.19.6

//! Link-level canonical-frame exchange machinery for [`WsConnection`].
//!
//! Every function here operates on one established [`WsLink`]: it feeds
//! socket facts through the transport-neutral driver, executes emitted
//! commands, and reports failures as typed [`ExchangeError`]s. The owning
//! connection decides what a loss means (routing the typed terminal into the
//! client unit); nothing here owns reconnect, retry, or replay policy.
//!
//! [`WsConnection`]: super::WsConnection

mod pump;
pub(super) use pump::try_receive_participant_frame;

use alloc::boxed::Box;
use alloc::collections::BTreeSet;
use alloc::format;
use alloc::string::{String, ToString};
use alloc::vec::Vec;
use core::time::Duration;

use std::time::Instant;

use liminal::protocol::{Frame, FrameType, PUBLISH_ACK_POSITION_VERSION, ProtocolVersion, decode};

use crate::SdkError;
use crate::remote::participant::ParticipantPumpDiagnostics;

use super::super::core::{
    DriverOutput, FrameCorrelation, FrameViolation, ResponseExpectation, SocketCommand,
    SocketEvent, SocketFailure, TransportTerminal, WebSocketFrameDriver,
};
use super::super::encode_frame;
use super::super::std_socket::{IO_TIMEOUT, SocketRead, WsSocket};
use crate::remote::framing::OPEN_DEADLINE;

/// Minimum protocol version this client advertises during the handshake.
const CLIENT_MIN_VERSION: ProtocolVersion = ProtocolVersion::new(1, 0);
/// Maximum protocol version this client advertises during the handshake; the
/// server selects the highest version both support, so 1.0 stays offered.
const CLIENT_MAX_VERSION: ProtocolVersion = PUBLISH_ACK_POSITION_VERSION;
/// Brief window used to detect an error reply for an otherwise-silent
/// conversation send, mirroring the TCP transport's drain contract: the server
/// replies synchronously on the connection thread, so on success it stays
/// silent and this single bounded read times out cleanly.
const CONVERSATION_DRAIN_TIMEOUT: Duration = Duration::from_millis(250);
/// Application stream id used for conversation frames.
pub(super) const APPLICATION_STREAM_ID: u32 = 1;

/// One established WebSocket link: socket, driver, and per-connection
/// conversation-open state (server conversation state is connection-scoped,
/// so the set resets with every reconnect).
pub(super) struct WsLink {
    pub(super) socket: WsSocket,
    pub(super) driver: WebSocketFrameDriver,
    pub(super) open_conversations: BTreeSet<u64>,
    pub(super) pump_fences_pending: usize,
    /// What this link's quiet drains have done.
    pub(super) pump_diagnostics: ParticipantPumpDiagnostics,
    /// The version this link's `ConnectAck` selected; `None` until it arrives.
    pub(super) negotiated_version: Option<ProtocolVersion>,
    /// How many socket windows closed under reads that owed a reply (SDK-024 R3).
    #[cfg(test)]
    pub(super) windows_closed: usize,
}

/// A failure inside one link operation.
pub(super) enum ExchangeError {
    /// The link is dead: the typed terminal must reach the client unit and
    /// the link must be dropped.
    Loss(Box<LinkLoss>),
    /// A local, non-fatal failure; the link remains usable.
    Local(SdkError),
}

/// The typed terminal and diagnostic detail of one link loss.
pub(super) struct LinkLoss {
    pub(super) terminal: TransportTerminal,
    pub(super) description: String,
}

impl ExchangeError {
    fn loss(terminal: TransportTerminal, description: String) -> Self {
        Self::Loss(Box::new(LinkLoss {
            terminal,
            description,
        }))
    }
}

/// Drives the liminal handshake (`Connect` -> `ConnectAck`) over a fresh link,
/// returning the protocol version the server selected.
pub(super) fn handshake(
    link: &mut WsLink,
    auth_token: &[u8],
) -> Result<ProtocolVersion, ExchangeError> {
    let connect = Frame::Connect {
        flags: 0,
        min_version: CLIENT_MIN_VERSION,
        max_version: CLIENT_MAX_VERSION,
        auth_token: auth_token.to_vec(),
    };
    let bytes = encode_frame(&connect).map_err(ExchangeError::Local)?;
    dispatch_send(link, bytes, ResponseExpectation::Correlated)?;
    // SDK-024 R4: the open is the one read with a bound (`OPEN_DEADLINE`).
    match receive_correlated(link, Some(OPEN_DEADLINE))? {
        Frame::ConnectAck {
            selected_version, ..
        } => Ok(selected_version),
        Frame::ConnectError {
            reason_code,
            message,
            ..
        } => Err(ExchangeError::Local(SdkError::Connection {
            description: format!(
                "server rejected connection (reason {reason_code}): {}",
                message.unwrap_or_else(|| "no detail".to_string())
            ),
        })),
        other => Err(ExchangeError::Local(unexpected_response(
            "ConnectAck",
            &other,
        ))),
    }
}

/// Sends one fire-and-forget frame through the driver.
pub(super) fn send_frame(link: &mut WsLink, frame: &Frame) -> Result<(), ExchangeError> {
    let bytes = encode_frame(frame).map_err(ExchangeError::Local)?;
    dispatch_send(link, bytes, ResponseExpectation::None)
}

/// Sends one correlated request and blocks for its response frame.
pub(super) fn exchange_correlated(
    link: &mut WsLink,
    bytes: Vec<u8>,
) -> Result<Frame, ExchangeError> {
    dispatch_send(link, bytes, ResponseExpectation::Correlated)?;
    unbounded(link, |link| receive_correlated(link, None))
}

/// Runs a read that owes a reply with the socket's read timeout cleared, so
/// it ends on a frame, the peer's close or a transport error (the kernel's
/// keepalive among them) and never on a window (SDK-024 R3). The steady-state
/// window is restored on every exit; the read's own error wins over the
/// restore's.
fn unbounded<T>(
    link: &mut WsLink,
    read: impl FnOnce(&mut WsLink) -> Result<T, ExchangeError>,
) -> Result<T, ExchangeError> {
    link.socket
        .set_read_timeout(None)
        .map_err(ExchangeError::Local)?;
    let result = read(link);
    let restored = link
        .socket
        .set_read_timeout(Some(IO_TIMEOUT))
        .map_err(ExchangeError::Local);
    let value = result?;
    restored?;
    Ok(value)
}

/// Emits one send command from the driver and executes it on the socket.
fn dispatch_send(
    link: &mut WsLink,
    bytes: Vec<u8>,
    expectation: ResponseExpectation,
) -> Result<(), ExchangeError> {
    let command = link
        .driver
        .command_send(bytes, expectation)
        .map_err(|refusal| {
            ExchangeError::Local(SdkError::Protocol {
                description: format!("websocket driver refused the send: {refusal:?}"),
            })
        })?;
    let SocketCommand::SendBinary(payload) = command else {
        return Err(ExchangeError::Local(SdkError::Protocol {
            description: "websocket driver emitted a non-send command for a send".to_string(),
        }));
    };
    if let Err(failure) = link.socket.send_binary(payload) {
        let step = link.driver.handle_event(SocketEvent::Failed(failure));
        if step.command == Some(SocketCommand::Close) {
            link.socket.execute_close();
        }
        let detail = link
            .socket
            .last_failure_detail()
            .unwrap_or("websocket send failed")
            .to_string();
        return Err(ExchangeError::loss(
            TransportTerminal::SocketFailed(failure),
            detail,
        ));
    }
    Ok(())
}

/// Blocks for the frame that resolves the outstanding correlated exchange,
/// draining unsolicited deliveries exactly like the TCP transport does.
///
/// With no `deadline` the socket carries no window ([`unbounded`]). With one
/// (the open's [`OPEN_DEADLINE`] alone), a closed `IO_TIMEOUT` window is a
/// wait, and only the deadline ends it, terminating the link typed.
fn receive_correlated(
    link: &mut WsLink,
    deadline: Option<Duration>,
) -> Result<Frame, ExchangeError> {
    let started = Instant::now();
    loop {
        match read_step(link, None)? {
            LinkRead::PumpFence => {}
            LinkRead::Frame { bytes, correlation } => match correlation {
                FrameCorrelation::CorrelatedResponse => return decode_response(link, &bytes),
                FrameCorrelation::UnsolicitedDelivery => {}
                FrameCorrelation::UnsolicitedFrame => {
                    return Err(ExchangeError::Local(SdkError::Protocol {
                        description:
                            "websocket driver classified a response as unsolicited while an \
                             exchange was outstanding"
                                .to_string(),
                    }));
                }
            },
            LinkRead::TimedOut => {
                #[cfg(test)]
                {
                    link.windows_closed += 1;
                }
                let elapsed = started.elapsed();
                let Some(deadline) = deadline.filter(|deadline| elapsed >= *deadline) else {
                    continue;
                };
                // The deadline is spent: the correlation window is now genuinely
                // broken, the positional contract cannot recover, and the link
                // terminates typed.
                let step = link
                    .driver
                    .handle_event(SocketEvent::Failed(SocketFailure::Transport));
                if step.command == Some(SocketCommand::Close) {
                    link.socket.execute_close();
                }
                return Err(ExchangeError::loss(
                    TransportTerminal::SocketFailed(SocketFailure::Transport),
                    deadline_detail("the correlated websocket response", elapsed, deadline),
                ));
            }
        }
    }
}

/// Blocks for the next non-delivery frame with no exchange outstanding: the
/// participant receive path. Unsolicited deliveries are drained exactly like
/// the TCP transport's receive loop; the wait carries no window ([`unbounded`]).
pub(super) fn receive_participant_frame(link: &mut WsLink) -> Result<Frame, ExchangeError> {
    unbounded(link, receive_participant_frame_unbounded)
}

fn receive_participant_frame_unbounded(link: &mut WsLink) -> Result<Frame, ExchangeError> {
    loop {
        match read_step(link, None)? {
            LinkRead::PumpFence => {}
            // No window is armed, so a timeout is only a platform's spurious
            // wake; the wait goes on.
            LinkRead::TimedOut => {
                #[cfg(test)]
                {
                    link.windows_closed += 1;
                }
            }
            LinkRead::Frame { bytes, correlation } => match correlation {
                FrameCorrelation::UnsolicitedDelivery => {}
                FrameCorrelation::CorrelatedResponse | FrameCorrelation::UnsolicitedFrame => {
                    return decode_response(link, &bytes);
                }
            },
        }
    }
}

/// Reads the next non-delivery frame if one arrives within `budget`, reporting
/// a quiet link as `Ok(None)`.
///
/// The WebSocket twin of the TCP transport's pump door, and the parity matters
/// for the same reason #61's did: a consumer draining an idle connection must
/// end its drain on the same silence whichever transport carries it, or the
/// same boot gate blows on one mount and not the other.
///
/// `budget` is spent across as many armed windows as it takes, with
/// [`IO_TIMEOUT`] as the polling grain — a remaining budget shorter than one
/// grain arms exactly that much rather than overshooting the caller's bound. A
/// spent budget returns before arming anything, so `Duration::ZERO` never
/// reaches the socket's `setsockopt`, which refuses a zero deadline.
///
/// Nothing here terminates the link: a quiet window is not a failure, so the
/// driver never sees a `Failed` event and the socket is retained for the next
/// call — unlike [`receive_correlated`], where a spent deadline genuinely does
/// break the positional correlation contract.
pub(super) fn receive_participant_frame_within(
    link: &mut WsLink,
    budget: Duration,
) -> Result<Option<Frame>, ExchangeError> {
    let started = Instant::now();
    loop {
        let Some(remaining) = budget.checked_sub(started.elapsed()) else {
            return Ok(None);
        };
        if remaining.is_zero() {
            return Ok(None);
        }
        match read_step(link, Some(remaining.min(IO_TIMEOUT)))? {
            // A closed window is a wait here too; the budget check above is the
            // only thing that ends this loop with silence.
            LinkRead::TimedOut | LinkRead::PumpFence => {}
            LinkRead::Frame { bytes, correlation } => match correlation {
                FrameCorrelation::UnsolicitedDelivery => {}
                FrameCorrelation::CorrelatedResponse | FrameCorrelation::UnsolicitedFrame => {
                    return decode_response(link, &bytes).map(Some);
                }
            },
        }
    }
}

/// Describes a spent deadline in seconds waited and seconds budgeted, naming
/// the timeout rather than the window that reported it.
fn deadline_detail(awaited: &str, elapsed: Duration, deadline: Duration) -> String {
    format!(
        "timed out after {:.3}s waiting for {awaited} (deadline {:.3}s): the peer accepted \
         the socket and said nothing, which a liminal server does not do",
        elapsed.as_secs_f64(),
        deadline.as_secs_f64()
    )
}

/// One validated read outcome surfaced to the exchange loops.
enum LinkRead {
    /// A validated canonical frame with its correlation class.
    Frame {
        /// Exact canonical frame bytes.
        bytes: Vec<u8>,
        /// Correlation class assigned by the driver.
        correlation: FrameCorrelation,
    },
    /// The armed read window elapsed (only when a window was armed).
    TimedOut,
    /// A validated, owned transport Pong; never a participant response.
    PumpFence,
}

/// Reads one socket fact, feeds it through the driver, executes any emitted
/// command, and maps terminals to typed losses.
fn read_step(link: &mut WsLink, window: Option<Duration>) -> Result<LinkRead, ExchangeError> {
    if let Some(window) = window {
        link.socket
            .set_read_timeout(Some(window))
            .map_err(ExchangeError::Local)?;
    }
    let read = link.socket.read_event();
    if window.is_some() {
        link.socket
            .set_read_timeout(Some(IO_TIMEOUT))
            .map_err(ExchangeError::Local)?;
    }
    let event = match read {
        SocketRead::TimedOut => return Ok(LinkRead::TimedOut),
        SocketRead::Event(event) => event,
    };
    // Retire owned transport fences before the driver's positional correlation
    // can mistake a delayed Pong for a later application's response.
    if let SocketEvent::Binary(bytes) = &event {
        if link.pump_fences_pending > 0
            && matches!(decode(bytes), Ok((Frame::Pong { .. }, consumed)) if consumed == bytes.len())
        {
            link.pump_fences_pending -= 1;
            return Ok(LinkRead::PumpFence);
        }
    }
    let step = link.driver.handle_event(event);
    if step.command == Some(SocketCommand::Close) {
        link.socket.execute_close();
    }
    match step.output {
        DriverOutput::Frame { bytes, correlation } => Ok(LinkRead::Frame { bytes, correlation }),
        DriverOutput::Terminal(terminal) => {
            let detail = link
                .socket
                .last_failure_detail()
                .map_or_else(|| terminal_description(&terminal), ToString::to_string);
            Err(ExchangeError::loss(terminal, detail))
        }
        DriverOutput::Opened | DriverOutput::PostTerminalIgnored(_) | DriverOutput::Refused(_) => {
            Err(ExchangeError::Local(SdkError::Protocol {
                description: format!(
                    "websocket driver produced an unexpected read output: {:?}",
                    step.output
                ),
            }))
        }
    }
}

/// Decodes one driver-validated canonical frame, requiring the decoder to
/// consume exactly the message bytes (one message, one frame).
pub(super) fn decode_response(link: &mut WsLink, bytes: &[u8]) -> Result<Frame, ExchangeError> {
    match decode(bytes) {
        Ok((frame, consumed)) if consumed == bytes.len() => Ok(frame),
        Ok((_, consumed)) => Err(undecodable_body(
            link,
            &format!(
                "canonical decode consumed {consumed} of {} websocket message bytes",
                bytes.len()
            ),
        )),
        Err(error) => Err(undecodable_body(
            link,
            &format!("canonical decode failed: {error}"),
        )),
    }
}

/// Terminates the link for a semantically undecodable body: the framing was
/// valid but the canonical codec refused the bytes, which is malformed input
/// and closes the connection like every other malformed message.
fn undecodable_body(link: &mut WsLink, detail: &str) -> ExchangeError {
    if link.driver.command_close().is_ok() {
        link.socket.execute_close();
    }
    ExchangeError::loss(
        TransportTerminal::ProtocolViolation(FrameViolation::UndecodableBody),
        format!("websocket message failed canonical decode: {detail}"),
    )
}

/// Opens the conversation on first use, surfacing any open failure before
/// recording the conversation as open.
pub(super) fn ensure_conversation_open(
    link: &mut WsLink,
    conversation_id: u64,
    subject: &str,
) -> Result<(), ExchangeError> {
    if link.open_conversations.contains(&conversation_id) {
        return Ok(());
    }
    let open = Frame::ConversationOpen {
        flags: 0,
        stream_id: APPLICATION_STREAM_ID,
        conversation_id,
        subject: subject.to_string(),
    };
    send_frame(link, &open)?;
    drain_conversation_error(link, conversation_id)?;
    link.open_conversations.insert(conversation_id);
    Ok(())
}

/// Reads a single pending reply under the brief drain window. A
/// `ConversationError` is surfaced typed; silence (timeout) is success.
pub(super) fn drain_conversation_error(
    link: &mut WsLink,
    conversation_id: u64,
) -> Result<(), ExchangeError> {
    loop {
        match read_step(link, Some(CONVERSATION_DRAIN_TIMEOUT))? {
            LinkRead::PumpFence => {}
            LinkRead::TimedOut => return Ok(()),
            LinkRead::Frame { bytes, correlation } => match correlation {
                FrameCorrelation::UnsolicitedDelivery => {}
                FrameCorrelation::CorrelatedResponse | FrameCorrelation::UnsolicitedFrame => {
                    let frame = decode_response(link, &bytes)?;
                    match frame {
                        Frame::ConversationError {
                            conversation_id: replied,
                            reason_code,
                            message,
                            ..
                        } => {
                            return Err(ExchangeError::Local(SdkError::Conversation {
                                conversation_id: replied.to_string(),
                                description: format!(
                                    "server rejected conversation {conversation_id} \
                                     (reason {reason_code}): {}",
                                    message.unwrap_or_else(|| "no detail".to_string())
                                ),
                            }));
                        }
                        other => {
                            return Err(ExchangeError::Local(unexpected_response(
                                "ConversationError or no reply",
                                &other,
                            )));
                        }
                    }
                }
            },
        }
    }
}

/// Builds a protocol error describing an unexpected response frame.
pub(in crate::remote::websocket) fn unexpected_response(
    expected: &str,
    actual: &Frame,
) -> SdkError {
    SdkError::Protocol {
        description: format!(
            "expected {expected} frame, received {:?}",
            FrameType::from(u8::from(actual.frame_type()))
        ),
    }
}

/// Human-readable description of a typed terminal (diagnostics only).
fn terminal_description(terminal: &TransportTerminal) -> String {
    match terminal {
        TransportTerminal::PeerClosed => "server closed the websocket connection".to_string(),
        TransportTerminal::CloseCompleted => "websocket close completed".to_string(),
        TransportTerminal::SocketFailed(failure) => {
            format!("websocket socket failed: {failure:?}")
        }
        TransportTerminal::ProtocolViolation(violation) => {
            format!("websocket message violated the frame contract: {violation:?}")
        }
    }
}

/// Maps a typed terminal to the SDK error taxonomy: protocol violations and
/// bound/content failures are protocol errors; socket losses are connection
/// errors. The mapping affects only the reported error class — the aggregate
/// transition was already decided by the client unit.
pub(super) fn loss_error(terminal: &TransportTerminal, description: &str) -> SdkError {
    match terminal {
        TransportTerminal::ProtocolViolation(_)
        | TransportTerminal::SocketFailed(
            SocketFailure::UnsupportedTextMessage | SocketFailure::MessageBeyondBound,
        ) => SdkError::Protocol {
            description: description.to_string(),
        },
        TransportTerminal::PeerClosed
        | TransportTerminal::CloseCompleted
        | TransportTerminal::SocketFailed(SocketFailure::Transport) => SdkError::Connection {
            description: description.to_string(),
        },
    }
}

#[cfg(test)]
#[path = "../reply_wait_tests.rs"]
mod reply_wait_tests;