liminal-server 0.28.2

//! The wait a person sees when a page opens a WebSocket, split by the server's
//! own log lines (work-list row "the server's start path", Hermes 7 Oct 05:00).
//!
//! Buttercup's live paint harness measured the page's new WebSocket to the
//! server's "websocket connection admitted" line at 70 to 1,005 ms, and the
//! first three connections after a start at 418, 681 and 191 ms against 14 to
//! 20 ms after them. Main logged neither the accept nor the first frame sent,
//! so that wait could not be split. One connection now logs four lines, each
//! timed from its own accept on the server's monotonic clock:
//!
//! 1. `websocket connection accepted`, when the listener's accept returns;
//! 2. `websocket upgrade written` with `upgrade_ms`, when the 101 is flushed;
//! 3. `websocket connection admitted` with `admit_ms`, when the spawn of its
//!    connection process returns;
//! 4. `websocket first frame sent` with `built_ms` (when its connection process
//!    was built) and `first_frame_ms`, when its first outbound frame is flushed.
//!
//! The pin opens one WebSocket subscription (whose first frame from the server
//! is the `ConnectAck`) and reads the four lines back, once each. Only causal
//! orders are asserted: the accept before the upgrade before the admission
//! (one thread hands to the next), and the upgrade before the process is built
//! before its first frame. The admission line and the first frame race: the
//! client's `Connect` can be answered before the spawn returns to log, so
//! `built_ms` is the admission time the first frame is measured against.

#[path = "support/persistence.rs"]
mod persistence;

use persistence::Persistence;

use std::error::Error;
use std::io::Write;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex, OnceLock};

use liminal_sdk::remote::websocket::WebSocketSubscriptionStream;
use liminal_server::config::{
    ChannelDef, LimitsConfig, ServerConfig, ServicesConfig, WebSocketConfig,
};
use liminal_server::server::connection::{ConnectionSupervisor, WebSocketListener};

const CHANNEL: &str = "events";
const PATH: &str = "/liminal";

/// The four lines, in the order one connection logs them.
const ACCEPTED: &str = "websocket connection accepted";
const UPGRADED: &str = "websocket upgrade written";
const ADMITTED: &str = "websocket connection admitted";
const FIRST_FRAME: &str = "websocket first frame sent";

type CapturedLog = Arc<Mutex<Vec<u8>>>;

static CAPTURED: OnceLock<CapturedLog> = OnceLock::new();

#[derive(Clone)]
struct CaptureWriter(CapturedLog);

impl Write for CaptureWriter {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
        let mut guard = self
            .0
            .lock()
            .map_err(|_| std::io::Error::other("the captured log's lock was poisoned"))?;
        guard.extend_from_slice(buf);
        drop(guard);
        Ok(buf.len())
    }

    fn flush(&mut self) -> std::io::Result<()> {
        Ok(())
    }
}

/// Installs the capturing subscriber for this test process, at debug for the
/// server's websocket door, and returns its buffer.
fn captured_log() -> Result<CapturedLog, Box<dyn Error>> {
    let mut installed = Ok(());
    let buffer = CAPTURED.get_or_init(|| {
        let buffer: CapturedLog = Arc::new(Mutex::new(Vec::new()));
        let writer = CaptureWriter(Arc::clone(&buffer));
        installed = tracing_subscriber::fmt()
            .with_writer(move || writer.clone())
            .with_env_filter("liminal_server::server::connection::websocket=debug")
            .with_ansi(false)
            .try_init()
            .map_err(|error| error.to_string());
        buffer
    });
    installed?;
    Ok(Arc::clone(buffer))
}

fn captured_text(buffer: &CapturedLog) -> Result<String, Box<dyn Error>> {
    let guard = buffer
        .lock()
        .map_err(|_| "the captured log's lock was poisoned")?;
    let text = String::from_utf8_lossy(&guard).into_owned();
    drop(guard);
    Ok(text)
}

/// A server with only its WebSocket door, on ephemeral ports.
struct RunningServer {
    _ws: WebSocketListener,
    supervisor: ConnectionSupervisor,
    ws_addr: SocketAddr,
    persistence: Persistence,
}

impl RunningServer {
    fn start() -> Result<Self, Box<dyn Error>> {
        let persistence = Persistence::copied()?;
        let health = std::net::TcpListener::bind("127.0.0.1:0")?;
        let health_listen_address = health.local_addr()?;
        drop(health);
        let config = ServerConfig {
            listen_address: "127.0.0.1:0".parse()?,
            extra_listen_addresses: Vec::new(),
            health_listen_address,
            drain_timeout_ms: 30_000,
            channels: vec![ChannelDef {
                name: CHANNEL.to_owned(),
                schema_ref: None,
                durable: false,
                loaded_schema: None,
            }],
            routing_rules: Vec::new(),
            persistence_path: Some(persistence.path().to_path_buf()),
            cluster: None,
            auth: None,
            services: ServicesConfig::default(),
            limits: LimitsConfig::default(),
            websocket: None,
            open: None,
            templates: liminal_server::services::lifecycle::templates::TemplateSources::default(),
            agents: None,
            participant: None,
        };
        let supervisor = ConnectionSupervisor::from_config(&config)?;
        let ws_config = WebSocketConfig {
            listen_address: "127.0.0.1:0".parse()?,
            path: PATH.to_owned(),
            allowed_origins: Vec::new(),
            allowed_origin_ports: Vec::new(),
            ping_interval_ms: None,
        };
        let ws = WebSocketListener::bind(&ws_config, supervisor.clone())?;
        let ws_addr = ws.local_addr();
        Ok(Self {
            _ws: ws,
            supervisor,
            ws_addr,
            persistence,
        })
    }
}

impl Drop for RunningServer {
    fn drop(&mut self) {
        self.supervisor.shutdown();
        assert!(
            self.persistence.path().is_dir(),
            "persistence fixture removed before shutdown"
        );
    }
}

/// The one line carrying `message`, which must appear exactly once.
fn the_line<'log>(log: &'log str, message: &str) -> Result<(usize, &'log str), Box<dyn Error>> {
    let mut found = log
        .lines()
        .enumerate()
        .filter(|(_, line)| line.contains(message));
    let first = found
        .next()
        .ok_or_else(|| format!("no `{message}` line in:\n{log}"))?;
    if found.next().is_some() {
        return Err(format!("`{message}` logged more than once in:\n{log}").into());
    }
    Ok(first)
}

/// The value of the `field=<ms>` the line carries.
fn millis(line: &str, field: &str) -> Result<u64, Box<dyn Error>> {
    let marker = format!("{field}=");
    let start = line
        .find(&marker)
        .ok_or_else(|| format!("no {field} on: {line}"))?
        + marker.len();
    let digits: String = line
        .get(start..)
        .unwrap_or_default()
        .chars()
        .take_while(char::is_ascii_digit)
        .collect();
    Ok(digits
        .parse()
        .map_err(|_| format!("{field} is not a count of milliseconds on: {line}"))?)
}

#[test]
fn one_websocket_logs_its_accept_upgrade_admission_and_first_frame_in_order()
-> Result<(), Box<dyn Error>> {
    let buffer = captured_log()?;
    let server = RunningServer::start()?;
    let url = format!("ws://{}{PATH}", server.ws_addr);
    let subscription = WebSocketSubscriptionStream::open(&url, CHANNEL, Vec::new())?;
    drop(subscription);
    drop(server);
    let log = captured_text(&buffer)?;

    let (accepted, _) = the_line(&log, ACCEPTED)?;
    let (upgraded, upgraded_line) = the_line(&log, UPGRADED)?;
    let (admitted, admitted_line) = the_line(&log, ADMITTED)?;
    let (first_frame, first_frame_line) = the_line(&log, FIRST_FRAME)?;
    assert!(
        accepted < upgraded && upgraded < admitted && upgraded < first_frame,
        "the lines out of causal order:\n{log}"
    );
    let upgrade_ms = millis(upgraded_line, "upgrade_ms")?;
    let admit_ms = millis(admitted_line, "admit_ms")?;
    let built_ms = millis(first_frame_line, "built_ms")?;
    let first_frame_ms = millis(first_frame_line, "first_frame_ms")?;
    assert!(
        upgrade_ms <= admit_ms && upgrade_ms <= built_ms && built_ms <= first_frame_ms,
        "upgrade {upgrade_ms} ms, admission {admit_ms} ms, built {built_ms} ms, \
         first frame {first_frame_ms} ms"
    );
    Ok(())
}