liminal-server 0.24.9

//! 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`).
#[derive(Default)]
struct Steps(Mutex<Vec<(String, Option<u64>)>>);
struct Capture(Arc<Steps>);

#[derive(Default)]
struct Step {
    boot_line: bool,
    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.boot_line = value == "boot_step",
            "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);
        if let (true, Some(name), Ok(mut steps)) = (step.boot_line, step.name, 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(())
}