liminal-sdk 0.19.2

//! The TCP subscription's mailbox: a stalled consumer holds at most the bound
//! and nothing is dropped; the reader's decode error, a server-ended
//! subscription no consumer was left for, and stray frames are each surfaced
//! or counted, never silent.

use super::*;
use liminal::protocol::{CausalContext, MessageEnvelope};
use std::net::TcpListener;
use std::sync::mpsc;

/// A guard on every wait: it fires only when the signal never comes.
const GUARD: Duration = Duration::from_secs(60);

const PAYLOAD: usize = 1024;

fn server_error(detail: &impl core::fmt::Display) -> SdkError {
    SdkError::Connection {
        description: format!("fake subscription server: {detail}"),
    }
}

fn encoded(frame: &Frame) -> Result<Vec<u8>, SdkError> {
    let len = encoded_len(frame).map_err(|error| protocol_error(&error))?;
    let mut bytes = vec![0_u8; len];
    let written = encode(frame, &mut bytes).map_err(|error| protocol_error(&error))?;
    bytes.truncate(written);
    Ok(bytes)
}

fn deliver(seq: u64, payload: usize) -> Result<Vec<u8>, SdkError> {
    let frame = Frame::new_deliver(
        SUBSCRIPTION_STREAM_ID,
        seq,
        MessageEnvelope::new(
            SchemaId::new([7; SchemaId::WIRE_LEN]),
            CausalContext::independent(),
            vec![0xA5; payload],
        ),
    )
    .map_err(|error| protocol_error(&error))?;
    encoded(&frame)
}

/// Reads one complete client frame, discarding it.
fn consume_one(socket: &mut TcpStream, buffer: &mut Vec<u8>) -> Result<(), SdkError> {
    loop {
        if let Ok((_, consumed)) = decode(buffer) {
            buffer.drain(..consumed);
            return Ok(());
        }
        let mut chunk = [0_u8; 512];
        let read = socket
            .read(&mut chunk)
            .map_err(|error| server_error(&error))?;
        if read == 0 {
            return Err(server_error(&"client closed before a full frame"));
        }
        buffer.extend_from_slice(chunk.get(..read).unwrap_or(&[]));
    }
}

/// Accepts one client, answers Connect and Subscribe, waits for `gate` when
/// one is given, writes each of `after`, then reads to the client's teardown.
fn serve(
    listener: TcpListener,
    gate: Option<mpsc::Receiver<()>>,
    after: Vec<Vec<u8>>,
) -> std::thread::JoinHandle<Result<(), SdkError>> {
    std::thread::spawn(move || -> Result<(), SdkError> {
        let (mut socket, _peer) = listener.accept().map_err(|error| server_error(&error))?;
        let mut buffer = Vec::new();
        consume_one(&mut socket, &mut buffer)?;
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: CLIENT_MAX_VERSION,
                capabilities: 0,
            },
        )?;
        consume_one(&mut socket, &mut buffer)?;
        write_frame(
            &mut socket,
            &Frame::SubscribeAck {
                flags: 0,
                stream_id: SUBSCRIPTION_STREAM_ID,
                subscription_id: 42,
                selected_schema: SchemaId::new([7; SchemaId::WIRE_LEN]),
            },
        )?;
        if let Some(gate) = gate {
            gate.recv_timeout(GUARD)
                .map_err(|error| server_error(&error))?;
        }
        for bytes in after {
            socket
                .write_all(&bytes)
                .map_err(|error| server_error(&error))?;
        }
        let mut scratch = [0_u8; 512];
        loop {
            match socket.read(&mut scratch) {
                Ok(0) => return Ok(()),
                Ok(_) => {}
                Err(error) if error.kind() == std::io::ErrorKind::ConnectionReset => {
                    return Ok(());
                }
                Err(error) => return Err(server_error(&error)),
            }
        }
    })
}

type Server = std::thread::JoinHandle<Result<(), SdkError>>;

fn open(
    limits: MailboxLimits,
    gate: Option<mpsc::Receiver<()>>,
    after: Vec<Vec<u8>>,
) -> Result<(SubscriptionStream, Server), SdkError> {
    let listener = TcpListener::bind("127.0.0.1:0").map_err(|error| server_error(&error))?;
    let address = listener
        .local_addr()
        .map_err(|error| server_error(&error))?
        .to_string();
    let server = serve(listener, gate, after);
    let stream = SubscriptionStream::open_with_limits(&address, "orders", Vec::new(), &[], limits)?;
    Ok((stream, server))
}

fn finish(stream: SubscriptionStream, server: Server) -> Result<(), SdkError> {
    drop(stream);
    server
        .join()
        .map_err(|_| server_error(&"server thread panicked"))?
}

#[test]
fn a_stalled_consumer_holds_at_most_the_frame_bound_and_nothing_is_dropped() -> Result<(), SdkError>
{
    let limits = MailboxLimits {
        frames: 4,
        bytes: 1024 * 1024,
    };
    let deliveries = (1..=8)
        .map(|seq| deliver(seq, PAYLOAD))
        .collect::<Result<Vec<_>, _>>()?;
    let (stream, server) = open(limits, None, deliveries)?;

    let stalled = stream
        .mailbox
        .wait_until(GUARD, |stats, _| stats.backpressure_waits >= 1)?;
    assert!(stalled, "the reader never stopped on a full mailbox");
    let held = stream.stats()?;
    assert_eq!(held.held_frames, 4);
    assert_eq!(held.held_bytes, 4 * PAYLOAD);

    for expected in 1..=8 {
        assert_eq!(stream.recv_timeout(GUARD)?.delivery_seq(), expected);
    }
    assert_eq!(stream.stats()?.peak_frames, 4);
    finish(stream, server)
}

#[test]
fn a_malformed_delivery_reaches_the_consumer_as_its_decode_error() -> Result<(), SdkError> {
    // A `Deliver` header declaring a three-byte body, too short for its fields.
    let mut malformed = vec![0x19_u8, 0, 0, 0, 0, 1];
    malformed.extend_from_slice(&3_u32.to_be_bytes());
    malformed.extend_from_slice(&[0xEE; 3]);
    let (stream, server) = open(MailboxLimits::SUBSCRIPTION, None, vec![malformed])?;

    let received = stream.recv_timeout(GUARD);

    assert!(
        matches!(
            &received,
            Err(SdkError::Protocol { description })
                if description.starts_with("subscription wire codec error")
        ),
        "expected the decode error itself, received {received:?}"
    );
    finish(stream, server)
}

#[test]
fn a_stray_frame_is_counted_and_the_delivery_behind_it_still_arrives() -> Result<(), SdkError> {
    let after = vec![encoded(&Frame::Pong { flags: 0 })?, deliver(1, 3)?];
    let (stream, server) = open(MailboxLimits::SUBSCRIPTION, None, after)?;

    assert_eq!(stream.recv_timeout(GUARD)?.delivery_seq(), 1);
    assert_eq!(stream.stats()?.ignored_frames, 1);
    finish(stream, server)
}

#[test]
fn a_subscribe_error_no_consumer_was_left_for_is_counted() -> Result<(), SdkError> {
    let (go, gate) = mpsc::channel();
    let subscribe_error = encoded(&Frame::SubscribeError {
        flags: 0,
        stream_id: SUBSCRIPTION_STREAM_ID,
        reason_code: 2,
        message: Some("shed".to_string()),
    })?;
    let (mut stream, server) = open(
        MailboxLimits::SUBSCRIPTION,
        Some(gate),
        vec![subscribe_error],
    )?;
    // The consumer goes away while the stream, and its reader, stay up.
    stream.mailbox.release();
    go.send(()).map_err(|error| server_error(&error))?;
    let reader = stream
        .reader
        .take()
        .ok_or_else(|| server_error(&"the stream has no reader"))?;
    reader
        .join()
        .map_err(|_| server_error(&"reader thread panicked"))?;

    assert_eq!(stream.stats()?.undelivered_errors, 1);
    finish(stream, server)
}