liminal-sdk 0.14.1

//! The push mailbox bound: a stalled consumer holds at most the bound in
//! frames and bytes, the reader stops reading instead of dropping, and a push
//! that could never fit is refused by name.

use super::*;
use crate::remote::mailbox::MailboxOverflow;
use std::net::TcpListener;
use std::sync::mpsc;

/// The client's half of a test server: it tells the server its handshake is
/// complete and the reader owns the socket. The handshake reads into a buffer
/// of its own, so bytes the server sends before this signal could be read
/// with the `ConnectAck` and never reach the reader.
type Gate = mpsc::Receiver<()>;

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

const PAYLOAD: usize = 1024;

/// Accepts one client, answers its handshake, pushes `count` frames of
/// `payload` bytes, then reads to the client's teardown.
fn serve_pushes(
    listener: TcpListener,
    gate: Gate,
    count: u64,
    payload: usize,
) -> std::thread::JoinHandle<Result<(), SdkError>> {
    serve(listener, gate, move |socket| {
        for correlation_id in 0..count {
            write_frame(
                socket,
                &Frame::Push {
                    flags: 0,
                    stream_id: APPLICATION_STREAM_ID,
                    correlation_id,
                    payload: vec![0xA5; payload],
                },
            )?;
        }
        Ok(())
    })
}

/// Accepts one client, answers its handshake, waits for `gate`, runs
/// `after_ack` on the socket, then reads to the client's teardown.
fn serve(
    listener: TcpListener,
    gate: Gate,
    after_ack: impl FnOnce(&mut TcpStream) -> Result<(), SdkError> + Send + 'static,
) -> std::thread::JoinHandle<Result<(), SdkError>> {
    std::thread::spawn(move || -> Result<(), SdkError> {
        let (mut socket, _peer) = listener.accept().map_err(|source| SdkError::Connection {
            description: format!("fake push server accept failed: {source}"),
        })?;
        let mut buffer = Vec::new();
        loop {
            match decode(&buffer) {
                Ok((Frame::Connect { .. }, consumed)) => {
                    buffer.drain(..consumed);
                    break;
                }
                Ok((other, _)) => {
                    return Err(SdkError::Protocol {
                        description: format!("expected Connect, read {:?}", other.frame_type()),
                    });
                }
                Err(_) => {
                    let mut chunk = [0_u8; 512];
                    let read = socket
                        .read(&mut chunk)
                        .map_err(|source| SdkError::Connection {
                            description: format!("fake push server read failed: {source}"),
                        })?;
                    if read == 0 {
                        return Err(SdkError::Connection {
                            description: "client closed before Connect".to_string(),
                        });
                    }
                    buffer.extend_from_slice(chunk.get(..read).unwrap_or(&[]));
                }
            }
        }
        write_frame(
            &mut socket,
            &Frame::ConnectAck {
                flags: 0,
                selected_version: CLIENT_MAX_VERSION,
                capabilities: 0,
            },
        )?;
        gate.recv_timeout(GUARD)
            .map_err(|source| SdkError::Connection {
                description: format!("fake push server gate: {source}"),
            })?;
        after_ack(&mut socket)?;
        let mut scratch = [0_u8; 512];
        loop {
            match socket.read(&mut scratch) {
                Ok(0) => return Ok(()),
                Ok(_) => {}
                Err(source) => {
                    return Err(SdkError::Connection {
                        description: format!("fake push server drain failed: {source}"),
                    });
                }
            }
        }
    })
}

fn client(
    limits: MailboxLimits,
    count: u64,
    payload: usize,
) -> Result<(PushClient, std::thread::JoinHandle<Result<(), SdkError>>), SdkError> {
    client_of(limits, |listener, gate| {
        serve_pushes(listener, gate, count, payload)
    })
}

fn client_of(
    limits: MailboxLimits,
    server: impl FnOnce(TcpListener, Gate) -> std::thread::JoinHandle<Result<(), SdkError>>,
) -> Result<(PushClient, std::thread::JoinHandle<Result<(), SdkError>>), SdkError> {
    let listener = TcpListener::bind("127.0.0.1:0").map_err(|source| SdkError::Connection {
        description: format!("failed to bind fake push server: {source}"),
    })?;
    let address = listener
        .local_addr()
        .map_err(|source| SdkError::Connection {
            description: format!("failed to read fake push server address: {source}"),
        })?
        .to_string();
    let (go, gate) = mpsc::channel();
    let server = server(listener, gate);
    let mut stream = connect_socket(&address, GUARD)?;
    handshake(&mut stream, &[], GUARD)?;
    let client = PushClient::start_reader_with(stream, limits)?;
    go.send(()).map_err(|source| SdkError::Connection {
        description: format!("fake push server gone before its gate: {source}"),
    })?;
    Ok((client, server))
}

fn wait_for_backpressure(client: &PushClient) -> Result<(), SdkError> {
    if client
        .mailbox
        .wait_until(GUARD, |stats, _| stats.backpressure_waits >= 1)?
    {
        Ok(())
    } else {
        Err(SdkError::Connection {
            description: "the reader never stopped on a full mailbox".to_string(),
        })
    }
}

fn finish(
    client: PushClient,
    server: std::thread::JoinHandle<Result<(), SdkError>>,
) -> Result<(), SdkError> {
    drop(client);
    server.join().map_err(|_| SdkError::Connection {
        description: "fake push server panicked".to_string(),
    })?
}

#[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 (client, server) = client(limits, 8, PAYLOAD)?;

    wait_for_backpressure(&client)?;
    let stalled = client.mailbox_stats()?;
    assert_eq!(stalled.held_frames, 4);
    assert_eq!(stalled.held_bytes, 4 * PAYLOAD);
    assert_eq!(stalled.backpressure_waits, 1);

    for expected in 0..8 {
        let frame = client.recv_timeout(GUARD)?;
        assert_eq!(
            frame.correlation_id(),
            expected,
            "every push arrives, in order"
        );
    }
    let drained = client.mailbox_stats()?;
    assert_eq!(drained.peak_frames, 4);
    assert_eq!(drained.peak_bytes, 4 * PAYLOAD);
    assert_eq!(drained.held_frames, 0);
    finish(client, server)
}

#[test]
fn a_stalled_consumer_holds_at_most_the_byte_bound() -> Result<(), SdkError> {
    let limits = MailboxLimits {
        frames: 1000,
        bytes: 3 * PAYLOAD + PAYLOAD / 2,
    };
    let (client, server) = client(limits, 8, PAYLOAD)?;

    wait_for_backpressure(&client)?;
    let stalled = client.mailbox_stats()?;
    assert_eq!(stalled.held_frames, 3);
    assert_eq!(stalled.held_bytes, 3 * PAYLOAD);
    assert!(stalled.held_bytes <= limits.bytes);

    for expected in 0..8 {
        assert_eq!(client.recv_timeout(GUARD)?.correlation_id(), expected);
    }
    assert_eq!(client.mailbox_stats()?.peak_bytes, 3 * PAYLOAD);
    finish(client, server)
}

#[test]
fn a_push_larger_than_the_mailbox_is_refused_by_name() -> Result<(), SdkError> {
    let limits = MailboxLimits {
        frames: 4,
        bytes: PAYLOAD / 2,
    };
    let (client, server) = client(limits, 1, PAYLOAD)?;

    let refusal = client.recv_timeout(GUARD);

    let expected = MailboxOverflow {
        payload_bytes: PAYLOAD,
        max_bytes: PAYLOAD / 2,
    };
    assert_eq!(refusal, Err(SdkError::from(expected)));
    let stats = client.mailbox_stats()?;
    assert_eq!(stats.refused_frames, 1);
    assert_eq!(stats.peak_bytes, 0);
    finish(client, server)
}

#[test]
fn a_malformed_frame_reaches_the_consumer_as_its_decode_error() -> Result<(), SdkError> {
    // A `Push` header declaring a three-byte body, too short for its fields.
    let mut malformed = vec![0x15_u8, 0, 0, 0, 0, 1];
    malformed.extend_from_slice(&3_u32.to_be_bytes());
    malformed.extend_from_slice(&[0xEE; 3]);
    let (client, server) = client_of(MailboxLimits::PUSH, move |listener, gate| {
        serve(listener, gate, move |socket| {
            socket
                .write_all(&malformed)
                .map_err(|source| SdkError::Connection {
                    description: format!("fake push server write failed: {source}"),
                })
        })
    })?;

    let received = client.recv_timeout(GUARD);

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

#[test]
fn a_stray_frame_is_counted_and_the_push_behind_it_still_arrives() -> Result<(), SdkError> {
    let (client, server) = client_of(MailboxLimits::PUSH, |listener, gate| {
        serve(listener, gate, |socket| {
            write_frame(socket, &Frame::Pong { flags: 0 })?;
            write_frame(
                socket,
                &Frame::Push {
                    flags: 0,
                    stream_id: APPLICATION_STREAM_ID,
                    correlation_id: 41,
                    payload: vec![1, 2, 3],
                },
            )
        })
    })?;

    let pushed = client.recv_timeout(GUARD)?;

    assert_eq!(pushed.correlation_id(), 41);
    assert_eq!(client.mailbox_stats()?.ignored_frames, 1);
    finish(client, server)
}