liminal-server 0.25.4

//! SRV-042 R1: a connection accepted while the incarnation start runs is held
//! off the accept thread, so stopping the listener never waits on the start;
//! the start's outcome admits or refuses what it holds, and the supervisor's
//! shutdown closes it.

use std::io::{Read, Write};
use std::net::{SocketAddr, TcpStream};
use std::sync::mpsc::{RecvTimeoutError, channel};
use std::time::Duration;

use liminal::protocol::{Frame, ProtocolError, decode, encode, encoded_len};

use crate::ServerError;
use crate::config::types::{LimitsConfig, ServerConfig, ServicesConfig, WebSocketConfig};
use crate::server::connection::WebSocketListener;
use crate::server::connection::participant_only_services::{HeldStart, held_start_supervisor};
use crate::server::listener::ServerListener;

type TestResult = Result<(), Box<dyn std::error::Error>>;

/// How long a test waits on a signal before it reports the wait as the defect.
const WITHIN: Duration = Duration::from_secs(5);

fn config() -> Result<ServerConfig, Box<dyn std::error::Error>> {
    Ok(ServerConfig {
        listen_address: "127.0.0.1:0".parse()?,
        extra_listen_addresses: Vec::new(),
        health_listen_address: "127.0.0.1:0".parse()?,
        drain_timeout_ms: 30_000,
        channels: Vec::new(),
        routing_rules: Vec::new(),
        persistence_path: None,
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: None,
        websocket: None,
    })
}

/// A listener on a supervisor whose start has begun and is held.
fn held_listener() -> Result<(HeldStart, ServerListener), Box<dyn std::error::Error>> {
    let held = held_start_supervisor()?;
    let listener = ServerListener::bind(&config()?, held.supervisor.clone())?;
    held.entered.recv_timeout(WITHIN)?;
    Ok((held, listener))
}

/// Runs `stop` on its own thread and answers whether it returned within
/// [`WITHIN`]; a stop still running is handed back to be joined after the
/// start is released, so the test never leaves a thread waiting.
fn stops_within<T: Send + 'static>(
    stop: impl FnOnce() -> Result<T, ServerError> + Send + 'static,
) -> (bool, std::thread::JoinHandle<Result<T, ServerError>>) {
    let (done, stopped) = channel();
    let worker = std::thread::spawn(move || {
        let result = stop();
        let _sent = done.send(());
        result
    });
    let returned = !matches!(stopped.recv_timeout(WITHIN), Err(RecvTimeoutError::Timeout));
    (returned, worker)
}

fn write_ping(stream: &mut TcpStream) -> TestResult {
    let frame = Frame::Ping { flags: 0 };
    let mut bytes = vec![0_u8; encoded_len(&frame)?];
    let written = encode(&frame, &mut bytes)?;
    stream.write_all(
        bytes
            .get(..written)
            .ok_or("encoded frame length was invalid")?,
    )?;
    Ok(())
}

/// The next frame, or `None` when the server closed the socket.
fn read_frame(stream: &mut TcpStream) -> Result<Option<Frame>, Box<dyn std::error::Error>> {
    let mut buffer = Vec::new();
    loop {
        match decode(&buffer) {
            Ok((frame, _)) => return Ok(Some(frame)),
            Err(
                ProtocolError::IncompleteHeader { .. } | ProtocolError::TruncatedPayload { .. },
            ) => {
                let mut chunk = [0_u8; 1024];
                let read = stream.read(&mut chunk)?;
                if read == 0 {
                    return if buffer.is_empty() {
                        Ok(None)
                    } else {
                        Err("the server closed inside a frame".into())
                    };
                }
                buffer.extend_from_slice(chunk.get(..read).ok_or("invalid read length")?);
            }
            Err(error) => return Err(error.into()),
        }
    }
}

fn client(address: SocketAddr) -> Result<TcpStream, Box<dyn std::error::Error>> {
    let stream = TcpStream::connect(address)?;
    stream.set_read_timeout(Some(WITHIN))?;
    Ok(stream)
}

/// Waits until the supervisor holds `count` clients (SRV-046 R1): a client the
/// accept worker has not yet taken from the backlog closes with the listener.
fn holds(held: &HeldStart, count: usize) -> TestResult {
    if held.supervisor.inner.held.wait_held(count, WITHIN) {
        Ok(())
    } else {
        Err(format!("the supervisor did not hold {count} client(s)").into())
    }
}

fn served(stream: &mut TcpStream) -> Result<bool, Box<dyn std::error::Error>> {
    write_ping(stream)?;
    Ok(matches!(read_frame(stream)?, Some(Frame::Pong { .. })))
}

#[test]
fn stopping_the_listener_does_not_wait_on_the_start_and_the_held_client_is_served() -> TestResult {
    let (held, mut listener) = held_listener()?;
    let mut first = client(listener.local_addr())?;
    holds(&held, 1)?;
    let (returned, stopping) = stops_within(move || {
        listener.stop_accepting()?;
        Ok(listener)
    });
    held.release.send(true)?;
    let _listener = stopping.join().map_err(|_| "the stop panicked")??;
    assert!(
        returned,
        "stop_accepting waited on the incarnation start behind a client it accepted"
    );
    assert!(
        served(&mut first)?,
        "the held client is served once the start finishes"
    );
    held.supervisor.shutdown();
    Ok(())
}

#[test]
fn held_clients_are_spawned_in_the_order_they_arrived() -> TestResult {
    let (held, mut listener) = held_listener()?;
    let mut first = client(listener.local_addr())?;
    let mut second = client(listener.local_addr())?;
    holds(&held, 2)?;
    let (returned, stopping) = stops_within(move || {
        listener.stop_accepting()?;
        Ok(listener)
    });
    held.release.send(true)?;
    let _listener = stopping.join().map_err(|_| "the stop panicked")??;
    assert!(returned, "stop_accepting waited on the incarnation start");
    assert!(served(&mut first)? && served(&mut second)?);
    let mut spawned = held.supervisor.inner.runtime.active_connections();
    spawned.sort_by_key(|connection| connection.pid);
    let peers: Vec<_> = spawned
        .iter()
        .map(|connection| connection.peer_addr)
        .collect();
    assert_eq!(
        peers,
        vec![Some(first.local_addr()?), Some(second.local_addr()?)],
        "held connections are spawned in arrival order"
    );
    held.supervisor.shutdown();
    Ok(())
}

#[test]
fn a_failed_start_refuses_the_held_client_with_its_error() -> TestResult {
    let (held, mut listener) = held_listener()?;
    let mut first = client(listener.local_addr())?;
    holds(&held, 1)?;
    let (returned, stopping) = stops_within(move || {
        listener.stop_accepting()?;
        Ok(listener)
    });
    held.release.send(false)?;
    let _listener = stopping.join().map_err(|_| "the stop panicked")??;
    assert!(returned, "stop_accepting waited on the incarnation start");
    let refusal = read_frame(&mut first)?;
    assert!(
        matches!(refusal, Some(Frame::ConnectError { .. })),
        "the held client reads the start's refusal: {refusal:?}"
    );
    assert!(read_frame(&mut first)?.is_none(), "and the socket closes");
    held.supervisor.shutdown();
    Ok(())
}

#[test]
fn the_supervisor_shutdown_closes_a_held_client_without_spawning_it() -> TestResult {
    let (held, mut listener) = held_listener()?;
    let mut first = client(listener.local_addr())?;
    holds(&held, 1)?;
    let supervisor = held.supervisor.clone();
    let (returned, stopping) = stops_within(move || {
        listener.stop_accepting()?;
        supervisor.shutdown();
        Ok(listener)
    });
    let closed = if returned {
        read_frame(&mut first).map(|frame| frame.is_none())
    } else {
        Ok(false)
    };
    held.release.send(true)?;
    let _listener = stopping.join().map_err(|_| "the stop panicked")??;
    assert!(returned, "the stop waited on the incarnation start");
    assert!(
        closed?,
        "the held client is closed by the shutdown, before the start finishes"
    );
    assert_eq!(held.supervisor.active_connection_count(), 0);
    Ok(())
}

#[test]
fn stopping_the_websocket_listener_does_not_wait_on_the_start() -> TestResult {
    let held = held_start_supervisor()?;
    let mut listener = WebSocketListener::bind(
        &WebSocketConfig {
            listen_address: "127.0.0.1:0".parse()?,
            path: "/liminal".to_owned(),
            allowed_origins: Vec::new(),
            allowed_origin_ports: Vec::new(),
            ping_interval_ms: None,
        },
        held.supervisor.clone(),
    )?;
    held.supervisor.listening()?;
    held.entered.recv_timeout(WITHIN)?;
    let address = listener.local_addr();
    let stream = TcpStream::connect(address)?;
    stream.set_read_timeout(Some(WITHIN))?;
    let (mut socket, _) = tungstenite::client::client(format!("ws://{address}/liminal"), stream)?;
    let (returned, stopping) = stops_within(move || {
        listener.stop_accepting()?;
        Ok(listener)
    });
    held.release.send(true)?;
    let _listener = stopping.join().map_err(|_| "the stop panicked")??;
    assert!(
        returned,
        "the websocket stop waited on the incarnation start behind an upgraded client"
    );
    let mut bytes = vec![0_u8; encoded_len(&Frame::Ping { flags: 0 })?];
    let written = encode(&Frame::Ping { flags: 0 }, &mut bytes)?;
    bytes.truncate(written);
    socket.send(tungstenite::Message::Binary(bytes.into()))?;
    let tungstenite::Message::Binary(reply) = socket.read()? else {
        return Err("the held websocket client got no binary reply".into());
    };
    assert!(matches!(decode(&reply)?.0, Frame::Pong { .. }));
    held.supervisor.shutdown();
    Ok(())
}