liminal-server 0.25.4

//! SRV-041 R1: a start names each fixed step once, with its milliseconds.

use super::TARGET;
use crate::config::types::{ChannelDef, LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::{ConnectionSupervisor, LiminalConnectionServices};
use crate::server::shutdown::ShutdownHandle;
use std::sync::{Arc, Mutex};
use tracing_subscriber::prelude::*;

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

/// The steps a start over a store with no unfinished connection names, each
/// exactly once, in no promised order.
const CLEAN_START: [&str; 6] = [
    "store_open",
    "channel_construction",
    "incarnation_startup",
    "restart_repair",
    "connection_scheduler_start",
    "reclaim_start",
];

/// Each boot step line as (name, `elapsed_ms`); a shutdown step's name is
/// prefixed `stop:`.
#[derive(Default)]
struct Steps(Mutex<Vec<(String, Option<u64>)>>);
struct Capture(Arc<Steps>);

#[derive(Default)]
struct Step {
    kind: Option<String>,
    name: Option<String>,
    elapsed_ms: Option<u64>,
}

impl tracing::field::Visit for Step {
    fn record_debug(&mut self, _: &tracing::field::Field, _: &dyn std::fmt::Debug) {}
    fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
        match field.name() {
            "step" => self.kind = Some(value.to_owned()),
            "name" => self.name = Some(value.to_owned()),
            _ => {}
        }
    }
    fn record_u64(&mut self, field: &tracing::field::Field, value: u64) {
        if field.name() == "elapsed_ms" {
            self.elapsed_ms = Some(value);
        }
    }
}

impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Capture {
    fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) {
        if event.metadata().target() != TARGET {
            return;
        }
        let mut step = Step::default();
        event.record(&mut step);
        let name = match (step.kind.as_deref(), step.name) {
            (Some("boot_step"), Some(name)) => name,
            (Some("shutdown_step"), Some(name)) => format!("stop:{name}"),
            _ => return,
        };
        if let Ok(mut steps) = self.0.0.lock() {
            steps.push((name, step.elapsed_ms));
        }
    }
}

fn config(home: &std::path::Path) -> TestResult<ServerConfig> {
    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![ChannelDef {
            name: "orders".to_owned(),
            schema_ref: None,
            durable: true,
            loaded_schema: None,
        }],
        routing_rules: Vec::new(),
        persistence_path: Some(home.to_path_buf()),
        cluster: None,
        auth: None,
        services: ServicesConfig::default(),
        limits: LimitsConfig::default(),
        participant: Some(crate::server::participant::test_participant_config()),
        websocket: None,
    })
}

/// The boot step lines as (name, `elapsed_ms`).
type Logged = Vec<(String, Option<u64>)>;

/// Starts services and a supervisor over `home` the way an embedder does
/// (Cambium: `LiminalConnectionServices::from_config`, then the supervisor,
/// then the incarnation start that follows the bind), stops them, and answers
/// the boot steps logged by the build and by the whole start.
fn built_then_started(home: &std::path::Path) -> TestResult<(Logged, Logged)> {
    let steps = Arc::new(Steps::default());
    let subscriber = tracing_subscriber::registry().with(Capture(Arc::clone(&steps)));
    let config = config(home)?;
    let (supervisor, services, built) = tracing::subscriber::with_default(subscriber, || {
        let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
        let shared: Arc<dyn crate::server::connection::ConnectionServices> = services.clone();
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            shared,
            None,
            None,
            config.limits,
            ShutdownHandle::new(),
        )?;
        let built = steps.0.lock().map(|logged| logged.clone()).ok();
        supervisor.current_server_incarnation()?;
        Ok::<_, crate::ServerError>((supervisor, services, built))
    })?;
    let built = built.ok_or("captured steps poisoned")?;
    drop(supervisor);
    drop(services);
    let logged = steps
        .0
        .lock()
        .map_err(|_| "captured steps poisoned")?
        .clone();
    Ok((built, logged))
}

fn started(home: &std::path::Path) -> TestResult<Logged> {
    Ok(built_then_started(home)?.1)
}

fn assert_clean_start(logged: &[(String, Option<u64>)]) {
    for name in CLEAN_START {
        let lines: Vec<_> = logged.iter().filter(|(step, _)| step == name).collect();
        assert_eq!(lines.len(), 1, "{name} logged once: {logged:?}");
        assert!(
            lines[0].1.is_some(),
            "{name} carries elapsed_ms: {logged:?}"
        );
    }
    assert!(
        logged
            .iter()
            .all(|(step, _)| step != "incarnation_recovery"),
        "no recovery after a clean stop: {logged:?}"
    );
}

#[test]
fn a_first_start_names_every_fixed_step_once_with_its_milliseconds() -> TestResult {
    let home = tempfile::tempdir()?;
    assert_clean_start(&started(home.path())?);
    Ok(())
}

#[test]
fn a_start_after_a_clean_stop_names_the_same_steps_and_no_recovery() -> TestResult {
    let home = tempfile::tempdir()?;
    started(home.path())?;
    assert_clean_start(&started(home.path())?);
    Ok(())
}

/// SRV-041 R2: building the supervisor reads nothing; the incarnation start
/// and its restart repair are logged only once the start runs.
#[test]
fn the_supervisor_build_names_no_incarnation_step() -> TestResult {
    let home = tempfile::tempdir()?;
    let (built, started) = built_then_started(home.path())?;
    for name in [
        "incarnation_startup",
        "incarnation_recovery",
        "restart_repair",
    ] {
        assert!(
            built.iter().all(|(step, _)| step != name),
            "{name} ran inside the supervisor build: {built:?}"
        );
    }
    for name in [
        "store_open",
        "channel_construction",
        "connection_scheduler_start",
        "reclaim_start",
    ] {
        assert!(
            built.iter().any(|(step, _)| step == name),
            "{name} in the build: {built:?}"
        );
    }
    assert_clean_start(&started);
    Ok(())
}

/// SRV-042 R3: a stop names the accept stop and the incarnation start's join,
/// each once with its milliseconds.
#[test]
fn a_stop_names_the_accept_stop_and_the_incarnation_join() -> TestResult {
    let home = tempfile::tempdir()?;
    let steps = Arc::new(Steps::default());
    let subscriber = tracing_subscriber::registry().with(Capture(Arc::clone(&steps)));
    let config = config(home.path())?;
    tracing::subscriber::with_default(subscriber, || {
        let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
        let shared: Arc<dyn crate::server::connection::ConnectionServices> = services.clone();
        let supervisor = ConnectionSupervisor::with_fatal_shutdown(
            shared,
            None,
            None,
            config.limits,
            ShutdownHandle::new(),
        )?;
        let listener = crate::server::listener::ServerListener::bind(&config, supervisor.clone())?;
        supervisor.current_server_incarnation()?;
        listener.shutdown()?;
        supervisor.shutdown();
        drop(supervisor);
        drop(services);
        Ok::<_, crate::ServerError>(())
    })?;
    let logged = steps
        .0
        .lock()
        .map_err(|_| "captured steps poisoned")?
        .clone();
    for name in ["stop:accept_stop", "stop:incarnation_join"] {
        let lines: Vec<_> = logged.iter().filter(|(step, _)| step == name).collect();
        assert_eq!(lines.len(), 1, "{name} logged once: {logged:?}");
        assert!(
            lines[0].1.is_some(),
            "{name} carries elapsed_ms: {logged:?}"
        );
    }
    Ok(())
}

/// The boot steps of one services build, in the order logged.
fn store_parts(built: &Logged) -> Vec<&str> {
    built
        .iter()
        .map(|(step, _)| step.as_str())
        .filter(|step| {
            matches!(
                *step,
                "ephemeral_root_open"
                    | "database_create"
                    | "database_open"
                    | "event_store_wrap"
                    | "store_open"
            )
        })
        .collect()
}

/// The `elapsed_ms` of the one line named `name`.
fn elapsed(built: &Logged, name: &str) -> TestResult<u64> {
    built
        .iter()
        .find(|(step, _)| step == name)
        .and_then(|(_, ms)| *ms)
        .ok_or_else(|| format!("{name} carries no elapsed_ms: {built:?}").into())
}

/// SRV-045 R1: `store_open` names its three parts, the database step by the
/// arm that ran, and their milliseconds sum to no more than its own.
#[test]
fn store_open_names_its_parts_create_then_open() -> TestResult {
    let home = tempfile::tempdir()?;
    for arm in ["database_create", "database_open"] {
        let (built, _) = built_then_started(home.path())?;
        assert_eq!(
            store_parts(&built),
            ["ephemeral_root_open", arm, "event_store_wrap", "store_open"],
            "the {arm} build's store steps: {built:?}"
        );
        let parts = elapsed(&built, "ephemeral_root_open")?
            + elapsed(&built, arm)?
            + elapsed(&built, "event_store_wrap")?;
        let whole = elapsed(&built, "store_open")?;
        assert!(
            parts <= whole,
            "the parts ({parts} ms) exceed store_open ({whole} ms): {built:?}"
        );
    }
    Ok(())
}