liminal-cli 0.3.0

//! The shipped binary must actually SAY something.
//!
//! `liminal-server` emits 120 `tracing` events (58 `warn`, 17 `error` — connection
//! failures, drain timeouts, durable-flush failures). Without a subscriber
//! installed at startup every one of them is dropped on the floor: the process
//! runs from boot to shutdown printing nothing at all, so an operator watching
//! the logs cannot tell a healthy server from a wedged one.
//!
//! This test spawns the REAL binary against a minimal valid config and waits to
//! be told: it reads the child's stderr until the `runtime.rs` startup event —
//! `"liminal server started"` at `info` — arrives, then proves the server it
//! names really booted (its wire port accepts a TCP connection). `RUST_LOG` is
//! explicitly removed from the child environment so this exercises the DEFAULT
//! level, not an operator override.
//!
//! Nothing here polls or times the start. The only bound is a hang guard far
//! beyond any start, and when it fires the one connect attempt made then tells
//! "the server never started" from "the server started but said nothing".

#[path = "../../liminal-server/src/test_persistence.rs"]
mod test_persistence;

use std::error::Error;
use std::io::{BufRead, BufReader};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::process::{Child, Command, Stdio};
use std::sync::mpsc::{self, Receiver, RecvTimeoutError};
use std::thread;
use std::time::{Duration, Instant};

/// The startup event `crate::server::runtime::run` emits once the listeners are
/// bound. Its arrival on stderr is the user-visible proof a subscriber is live.
const STARTUP_EVENT: &str = "liminal server started";

/// How long the test waits for the startup line before it calls the run hung.
/// It is not a bound on how long a start may take: a start costs milliseconds,
/// and this only keeps a wedged child from holding the battery forever.
const HANG_GUARD: Duration = Duration::from_secs(300);

/// How long the one connect made after the startup line, or when the hang
/// guard fires, may take: the address is loopback.
const CONNECT_GUARD: Duration = Duration::from_secs(5);

/// Kills and reaps the spawned server however the test exits, so a failing
/// assertion never leaks a listening process onto the box.
struct ServerProcess {
    child: Child,
}

impl Drop for ServerProcess {
    fn drop(&mut self) {
        if let Err(error) = self.child.kill() {
            match self.child.try_wait() {
                Ok(Some(status)) => eprintln!(
                    "server cleanup kill failed because the child already exited ({status}): {error}"
                ),
                Ok(None) => {
                    eprintln!("server cleanup kill failed with the child still running: {error}");
                }
                Err(status_error) => eprintln!(
                    "server cleanup kill failed: {error}; child status failed: {status_error}"
                ),
            }
        }
        if let Err(error) = self.child.wait() {
            eprintln!("server cleanup wait failed: {error}");
        }
    }
}

/// Reserves an ephemeral loopback port and releases it, returning the address.
///
/// Config validation rejects port `0` (`listen_address: port must be non-zero`),
/// so the binary cannot be handed an ephemeral bind directly; the kernel picks
/// the number here and the server re-binds it a moment later. This is the same
/// idiom the existing server e2e tests use for their health ports.
fn reserved_address() -> Result<SocketAddr, Box<dyn Error>> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let address = listener.local_addr()?;
    drop(listener);
    Ok(address)
}

/// Whether `address` accepts a TCP connection now. One attempt, never a loop:
/// it is asked only once the server has said it started, or once the hang
/// guard has fired.
fn accepts_connection(address: SocketAddr) -> bool {
    TcpStream::connect_timeout(&address, CONNECT_GUARD).is_ok()
}

/// Drains the child's stderr on a helper thread, forwarding whole lines.
///
/// The reader owns the pipe and ends at EOF, which the [`ServerProcess`] kill
/// guarantees, so the thread is left detached rather than joined.
fn stderr_lines(child: &mut Child) -> Result<Receiver<String>, Box<dyn Error>> {
    let stderr = child
        .stderr
        .take()
        .ok_or("spawned server had no piped stderr")?;
    let (sender, receiver) = mpsc::channel();
    thread::spawn(move || {
        for line in BufReader::new(stderr).lines() {
            let Ok(line) = line else { return };
            if sender.send(line).is_err() {
                return;
            }
        }
    });
    Ok(receiver)
}

/// Why the startup line never arrived.
enum Unheard {
    /// The child closed its stderr, so it exited, without writing the line.
    Exited,
    /// [`HANG_GUARD`] passed with the child still running and silent.
    Hung,
}

/// Collects stderr lines until one contains `needle`, the child closes its
/// stderr, or [`HANG_GUARD`] passes, returning everything seen so a failure
/// names what DID arrive.
fn await_line(
    receiver: &Receiver<String>,
    needle: &str,
) -> Result<Vec<String>, (Unheard, Vec<String>)> {
    let guard = Instant::now() + HANG_GUARD;
    let mut seen = Vec::new();
    loop {
        let remaining = guard.saturating_duration_since(Instant::now());
        match receiver.recv_timeout(remaining) {
            Ok(line) => {
                let matched = line.contains(needle);
                seen.push(line);
                if matched {
                    return Ok(seen);
                }
            }
            Err(RecvTimeoutError::Disconnected) => return Err((Unheard::Exited, seen)),
            Err(RecvTimeoutError::Timeout) => return Err((Unheard::Hung, seen)),
        }
    }
}

#[test]
fn spawned_server_reports_startup_on_stderr() -> Result<(), Box<dyn Error>> {
    let directory = test_persistence::directory()?;
    let listen_address = reserved_address()?;
    let health_listen_address = reserved_address()?;
    let config_path = directory.path().join("server.toml");
    std::fs::write(
        &config_path,
        format!(
            "listen_address = \"{listen_address}\"\n\
             health_listen_address = \"{health_listen_address}\"\n\
             drain_timeout_ms = 5000\n\
             channels = []\n\
             routing_rules = []\n"
        ),
    )?;

    let mut server = ServerProcess {
        child: Command::new(env!("CARGO_BIN_EXE_liminal"))
            .arg("--config")
            .arg(&config_path)
            .env("LIMINAL_PERSISTENCE_PATH", directory.path())
            // The DEFAULT level is what is under test: an operator who sets
            // nothing must still see the server's own info events.
            .env_remove("RUST_LOG")
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()?,
    };
    let stderr = stderr_lines(&mut server.child)?;

    match await_line(&stderr, STARTUP_EVENT) {
        // Proves the boot path the line reports really succeeded: the event is
        // emitted once the listeners are bound, so the wire port accepts now.
        Ok(_) if accepts_connection(listen_address) => Ok(()),
        Ok(seen) => Err(format!(
            "server wrote {STARTUP_EVENT:?} but {listen_address} refused a connection; \
             stderr carried {} line(s): {seen:#?}",
            seen.len()
        )
        .into()),
        Err((Unheard::Exited, seen)) => {
            let status = server.child.wait()?;
            Err(format!(
                "server exited ({status}) without writing {STARTUP_EVENT:?}; \
                 stderr carried {} line(s): {seen:#?}",
                seen.len()
            )
            .into())
        }
        // Only a server that is up and silent is the failure under test; one
        // that never bound is a start that hung, and says so.
        Err((Unheard::Hung, seen)) => {
            let bound = accepts_connection(listen_address);
            Err(format!(
                "after {HANG_GUARD:?} the server {} and never wrote {STARTUP_EVENT:?}; \
                 stderr carried {} line(s): {seen:#?}",
                if bound {
                    format!("bound {listen_address}")
                } else {
                    format!("had not bound {listen_address}")
                },
                seen.len()
            )
            .into())
        }
    }
}

#[test]
fn spawned_server_storage_refusal_is_named_and_exits_nonzero() -> Result<(), Box<dyn Error>> {
    spawned_server_refusal(true)
}

#[test]
fn spawned_server_invalid_persistence_path_is_named_and_exits_nonzero() -> Result<(), Box<dyn Error>>
{
    spawned_server_refusal(false)
}

fn spawned_server_refusal(store_failure: bool) -> Result<(), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let persistence = directory.path().join("not-a-directory");
    let blocked = if store_failure {
        std::fs::create_dir(&persistence)?;
        persistence.join("durability")
    } else {
        persistence.clone()
    };
    std::fs::write(&blocked, b"occupied")?;
    let encoded_path = toml::Value::String(
        persistence
            .to_str()
            .ok_or("persistence path is not UTF-8")?
            .to_owned(),
    );
    let listen_address = reserved_address()?;
    let health_listen_address = reserved_address()?;
    let config = directory.path().join("refused.toml");
    std::fs::write(
        &config,
        format!(
            "listen_address = \"{listen_address}\"\nhealth_listen_address = \"{health_listen_address}\"\ndrain_timeout_ms = 5000\npersistence_path = {encoded_path}\nchannels = []\nrouting_rules = []\n"
        ),
    )?;
    let output = Command::new(env!("CARGO_BIN_EXE_liminal"))
        .arg("--config")
        .arg(config)
        .env_remove("RUST_LOG")
        .stdin(Stdio::null())
        .output()?;
    let stderr = String::from_utf8(output.stderr)?;
    assert_eq!(output.status.code(), Some(1), "{stderr}");
    let named_error = if store_failure {
        "failed to open durable store at"
    } else {
        "path must be an existing directory"
    };
    assert!(stderr.contains(named_error), "{stderr}");
    assert!(
        stderr.contains(blocked.to_str().ok_or("refused path is not UTF-8")?),
        "{stderr}"
    );
    assert_eq!(std::fs::read(blocked)?, b"occupied");
    Ok(())
}