liminal-server 0.28.2

use super::fixture::{Fixture, toml_path};
use super::{TestError, TestResult};
use liminal::protocol::{Frame, FrameHeader, ProtocolVersion, decode, encode, encoded_len};
use std::fmt::Write as _;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex, mpsc};
use std::thread::JoinHandle;
use std::time::Duration;

const TEST: &str = "a_refused_boot_pass_names_itself_and_the_server_starts";
const CONFIG: &str = "NET012_BOOT_CONFIG";
const OBSERVATION: Duration = Duration::from_secs(30);
const EXPORTED_CHANNEL: &str = "[[channels]]\nname = 'orders'\ndurable = true\n";

/// What the server tells the test while it runs: a line of its log, the
/// peer socket's accepted dial, a refused accept, or the end of its output.
/// The test waits on these; a clock never stands in for one of them.
enum Event {
    Line(String),
    Dialled,
    PeerAccept(std::io::Error),
    Ended,
}

/// The health door answers 200 while the pass is refused.
pub fn health_answers(health: SocketAddr) -> TestResult {
    let mut socket = TcpStream::connect(health)?;
    socket.set_read_timeout(Some(OBSERVATION))?;
    socket.write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")?;
    let mut response = String::new();
    socket.read_to_string(&mut response)?;
    assert!(response.starts_with("HTTP/1.1 200 "), "health refused");
    Ok(())
}

/// The boot refusal is logged once, at warn, by Lys's name, saying the next
/// open asks again.
fn boot_refusal_named_once(output: &str) {
    let refusals: Vec<_> = output
        .lines()
        .filter(|line| line.contains("phase=\"boot\""))
        .collect();
    assert_eq!(
        refusals.len(),
        1,
        "boot refusal was not logged once: {output}"
    );
    assert!(
        refusals[0].contains("WARN")
            && refusals[0].contains("machine pass refused")
            && refusals[0].contains("asked again at the next link open")
            && refusals[0].contains("agent_dial_refused"),
        "boot refusal was not named at warn: {}",
        refusals[0]
    );
}

/// In the child a test re-ran itself as: the production runtime over the
/// fixture's configuration. The server owns process signals, so it lives in
/// a child; `None` in the test itself.
pub fn serve_child() -> Option<TestResult> {
    let config = std::env::var_os(CONFIG)?;
    Some(
        tracing_subscriber::fmt()
            .without_time()
            .with_ansi(false)
            .with_writer(std::io::stderr)
            .try_init()
            .and_then(|()| {
                liminal_server::server::runtime::run(Path::new(&config)).map_err(Into::into)
            }),
    )
}

pub fn refused_boot() -> TestResult {
    if let Some(served) = serve_child() {
        return served;
    }
    let fixture = Fixture::start(Some("agent_dial_refused"))?;
    let peer = TcpListener::bind("127.0.0.1:0")?;
    peer.set_nonblocking(true)?;
    let wire = TcpListener::bind("127.0.0.1:0")?;
    let health = TcpListener::bind("127.0.0.1:0")?;
    let config = configuration(
        &fixture,
        peer.local_addr()?,
        wire.local_addr()?,
        health.local_addr()?,
    )?;
    drop(wire);
    drop(health);
    let (server, wire, health) = BootServer::start(&config)?;
    // Boot told the test its refusal and its promise before it started.
    let boot = server.line(|line| line.contains("phase=\"boot\""))?;
    assert!(
        boot.contains("asked again at the next link open"),
        "the boot refusal does not promise a re-ask: {boot}"
    );
    health_answers(health)?;
    let mut socket = TcpStream::connect(wire)?;
    socket.set_read_timeout(Some(OBSERVATION))?;
    send(
        &mut socket,
        &Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: b"fixture".to_vec(),
        },
    )?;
    assert!(matches!(receive(&mut socket)?, Frame::ConnectAck { .. }));
    send(
        &mut socket,
        &Frame::Subscribe {
            flags: 0,
            stream_id: 1,
            channel: "console.topology".into(),
            accepted_schemas: Vec::new(),
            max_in_flight: 1,
        },
    )?;
    assert!(matches!(receive(&mut socket)?, Frame::SubscribeAck { .. }));
    let Frame::Deliver { envelope, .. } = receive(&mut socket)? else {
        return Err("topology was not delivered".into());
    };
    let topology: serde_json::Value = serde_json::from_slice(&envelope.payload)?;
    // The host is installed without a pass (ACCESS-009 R1), so the
    // configured link may already be listed by its refusal; none is up.
    assert!(
        !topology["links"]
            .as_array()
            .ok_or("topology links absent")?
            .iter()
            .any(|link| link["state"] == "up"),
        "a link came up without a pass: {topology}"
    );
    assert!(
        matches!(peer.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock),
        "refused boot pass dialed an outbound link"
    );
    assert!(fixture.posts.load(Ordering::Acquire) >= 1);
    // ACCESS-009 R2: Lys begins issuing; the configured link retry owner's
    // next attempt asks, takes the pass and dials the peer, with no restart
    // and no operator act. The pass is asked before any wire connect, so
    // the accepted connection is the proof of the issued pass.
    fixture.issue();
    peer.set_nonblocking(false)?;
    server.dial(peer);
    // The dial is the proof of the pass taken at the re-ask; a refusal whose
    // reason is no longer the pass's names what went wrong after it.
    loop {
        match server.event()? {
            Event::Dialled => break,
            Event::Line(line)
                if line.contains("dial will be retried")
                    && !line.contains("agent_dial_refused") =>
            {
                return Err(format!("the dial after Lys issued was refused: {line}").into());
            }
            Event::Line(_) => {}
            Event::PeerAccept(error) => return Err(format!("peer accept refused: {error}").into()),
            Event::Ended => return Err("server output ended before the dial".into()),
        }
    }
    assert!(
        fixture.posts.load(Ordering::Acquire) >= 2,
        "Lys was not asked again after it began issuing"
    );
    drop(socket);
    let output = server.stop()?;
    boot_refusal_named_once(&output);
    assert!(
        !output.contains(&fixture.first_pass) && !output.contains(&fixture.peer_pass),
        "pass leaked"
    );
    fixture.finish()
}

pub fn configuration(
    fixture: &Fixture,
    peer: SocketAddr,
    wire: SocketAddr,
    health: SocketAddr,
) -> Result<std::path::PathBuf, TestError> {
    let auth = fixture.auth("")?;
    let lys = auth.lys.as_ref().ok_or("Lys config absent")?;
    let machine = lys.machine.as_ref().ok_or("machine config absent")?;
    let store = fixture.root.path().join("store");
    std::fs::create_dir(&store)?;
    let config = fixture.root.path().join("boot.toml");
    let store = toml_path(&store)?;
    let keys = toml_path(&lys.keys_path)?;
    let identity = toml_path(&machine.identity_path)?;
    let peer_key = lys_core::keys::Ed25519Identity::from_seed(&zeroize::Zeroizing::new([2; 32]));
    std::fs::write(
        &config,
        format!(
            "listen_address = '{wire}'\nhealth_listen_address = '{health}'\n\
         drain_timeout_ms = 0\nrouting_rules = []\npersistence_path = {store}\n\
         {EXPORTED_CHANNEL}\
         {PARTICIPANT}\n[auth]\ntoken = 'fixture'\n\
         [auth.lys]\nissuer = {:?}\nnetwork_id = 'network'\nkeys_path = {keys}\napi_url = {:?}\n\
         [auth.lys.machine]\nidentity_path = {identity}\nagent_id = '{}'\naudience = 'liminal'\n\
         [auth.lys.machine.machines]\nimporter = {:?}\n\
         [[auth.lys.machine.links]]\nlink_id = 'orders-link'\npeer_holder = 'importer'\n\
         peer_address = '{peer}'\nexports = ['orders']\n",
            lys.issuer,
            lys.api_url.as_ref().ok_or("API URL absent")?,
            super::fixture::agent_issuer::AGENT,
            hex::encode(peer_key.public_key_bytes()),
        ),
    )?;
    Ok(config)
}

#[test]
fn boot_export_configuration_passes_validation() -> TestResult {
    let root = tempfile::tempdir()?;
    let peer = lys_core::keys::Ed25519Identity::from_seed(&zeroize::Zeroizing::new([2; 32]));
    let mut config: liminal_server::config::ServerConfig = toml::from_str(&format!(
        "listen_address = '127.0.0.1:8080'\nhealth_listen_address = '127.0.0.1:8081'\n\
         drain_timeout_ms = 0\nrouting_rules = []\npersistence_path = {}\n\
         {EXPORTED_CHANNEL}\n[auth]\ntoken = 'fixture'\n\
         [auth.lys]\nissuer = 'https://issuer.example'\nnetwork_id = 'network'\nkeys_path = 'keys.json'\n\
         [auth.lys.machine]\nidentity_path = 'identity.key'\npass_path = 'machine.pass'\n\
         [auth.lys.machine.machines]\nimporter = {:?}\n\
         [[auth.lys.machine.links]]\nlink_id = 'orders-link'\npeer_holder = 'importer'\nexports = ['orders']\n",
        toml_path(root.path())?,
        hex::encode(peer.public_key_bytes()),
    ))?;
    liminal_server::config::validate(&mut config, None)?;
    Ok(())
}

pub struct BootServer {
    child: Child,
    reader: Option<JoinHandle<TestResult>>,
    logs: Arc<Mutex<String>>,
    events: mpsc::Receiver<Event>,
    teller: mpsc::Sender<Event>,
}

impl BootServer {
    fn start(config: &Path) -> Result<(Self, SocketAddr, SocketAddr), TestError> {
        Self::start_as(TEST, config)
    }

    /// Runs the test named `test` as the server's child over `config`.
    pub fn start_as(
        test: &str,
        config: &Path,
    ) -> Result<(Self, SocketAddr, SocketAddr), TestError> {
        let mut command = Command::new(std::env::current_exe()?);
        command
            .args(["--exact", test, "--nocapture"])
            .env(CONFIG, config)
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped());
        // Deployment overrides must not replace this fixture's configuration.
        for (key, _) in
            std::env::vars_os().filter(|(key, _)| key.to_string_lossy().starts_with("LIMINAL_"))
        {
            command.env_remove(key);
        }
        let (teller, events) = mpsc::channel();
        let mut server = Self {
            child: command.spawn()?,
            reader: None,
            logs: Arc::new(Mutex::new(String::new())),
            events,
            teller,
        };
        let stderr = server.child.stderr.take().ok_or("server stderr absent")?;
        let logs = Arc::clone(&server.logs);
        let (started, startup) = mpsc::channel();
        let teller = server.teller.clone();
        server.reader = Some(std::thread::spawn(move || {
            let mut started = Some(started);
            for line in BufReader::new(stderr).lines() {
                let line = line?;
                writeln!(logs.lock().map_err(|error| error.to_string())?, "{line}")?;
                if line.contains("liminal server started")
                    && let Some(started) = started.take()
                {
                    started.send(line.clone())?;
                }
                teller
                    .send(Event::Line(line))
                    .map_err(|_| "the test stopped listening to the server")?;
            }
            teller
                .send(Event::Ended)
                .map_err(|_| "the test stopped listening to the server")?;
            Ok(())
        }));
        let line = startup.recv_timeout(OBSERVATION).map_err(|error| {
            format!(
                "server startup refused: {error}; {}",
                server.output().unwrap_or_else(|error| error.to_string())
            )
        })?;
        let address = |field| -> Result<SocketAddr, TestError> {
            line.split_whitespace()
                .find_map(|word| word.strip_prefix(field))
                .ok_or("startup address absent")?
                .parse()
                .map_err(Into::into)
        };
        let wire = address("listen_address=")?;
        let health = address("health_listen_address=")?;
        Ok((server, wire, health))
    }

    /// The next thing the server told the test. The observation watchdog is
    /// the one clock here: it names a server that stopped telling anything,
    /// neither a line nor a dial, and never stands in for an event.
    fn event(&self) -> Result<Event, TestError> {
        self.events.recv_timeout(OBSERVATION).map_err(|error| {
            format!(
                "the server told the test nothing within the observation watchdog: {error}; {}",
                self.output().unwrap_or_else(|error| error.to_string())
            )
            .into()
        })
    }

    /// The next log line the server told that `matches`.
    pub fn line(&self, matches: impl Fn(&str) -> bool) -> Result<String, TestError> {
        loop {
            match self.event()? {
                Event::Line(line) if matches(&line) => return Ok(line),
                Event::Line(_) | Event::Dialled => {}
                Event::PeerAccept(error) => {
                    return Err(format!("peer accept refused: {error}").into());
                }
                Event::Ended => return Err("server output ended before the line".into()),
            }
        }
    }

    /// Accepts the server's dial on `peer` on its own thread and tells it.
    fn dial(&self, peer: TcpListener) {
        let teller = self.teller.clone();
        std::thread::spawn(move || {
            let event = match peer.accept() {
                Ok((stream, _)) => {
                    drop(stream);
                    Event::Dialled
                }
                Err(error) => Event::PeerAccept(error),
            };
            // Nobody listens once the test has ended; nothing to tell.
            if teller.send(event).is_err() {
                eprintln!("the dial arrived after the test ended");
            }
        });
    }

    pub fn output(&self) -> Result<String, TestError> {
        Ok(self.logs.lock().map_err(|error| error.to_string())?.clone())
    }

    pub fn stop(mut self) -> Result<String, TestError> {
        let status = Command::new("/bin/kill")
            .args(["-TERM", &self.child.id().to_string()])
            .status()?;
        assert!(
            status.success(),
            "owned server termination refused: {status}"
        );
        let status = self.child.wait()?;
        self.join_reader()?;
        assert!(
            status.success(),
            "server shutdown refused: {status}; {}",
            self.output()?
        );
        self.output()
    }

    fn join_reader(&mut self) -> TestResult {
        if let Some(reader) = self.reader.take() {
            reader.join().map_err(|_| "server log reader panicked")??;
        }
        Ok(())
    }
}

impl Drop for BootServer {
    fn drop(&mut self) {
        match self.child.try_wait() {
            Ok(Some(_)) => {}
            result => {
                if let Err(error) = result {
                    eprintln!("server cleanup status: {error}");
                }
                if let Err(error) = self.child.kill() {
                    eprintln!("owned server cleanup: {error}");
                }
                if let Err(error) = self.child.wait() {
                    eprintln!("owned server reap: {error}");
                }
            }
        }
        if let Err(error) = self.join_reader() {
            eprintln!("server log cleanup: {error}");
        }
    }
}

fn send(socket: &mut TcpStream, frame: &Frame) -> TestResult {
    let mut bytes = vec![0; encoded_len(frame)?];
    let size = encode(frame, &mut bytes)?;
    socket.write_all(&bytes[..size])?;
    Ok(())
}

fn receive(socket: &mut TcpStream) -> Result<Frame, TestError> {
    let mut header = [0; FrameHeader::WIRE_LEN];
    socket.read_exact(&mut header)?;
    let length = usize::try_from(u32::from_be_bytes(header[6..10].try_into()?))?;
    if length > 1_048_576 {
        return Err("fixture frame exceeds bound".into());
    }
    let mut bytes = header.to_vec();
    bytes.resize(header.len() + length, 0);
    socket.read_exact(&mut bytes[header.len()..])?;
    Ok(decode(&bytes)?.0)
}

const PARTICIPANT: &str = "[participant]
wire_frame_limit = 65536
attach_receipt_ttl_ms = 60000
receipt_provenance_ttl_ms = 600000
live_receipt_server_report_threshold = 1024
max_live_attach_receipts_per_participant = 8
receipt_provenance_server_report_threshold = 4096
receipt_provenance_per_conversation_report_threshold = 256
max_receipt_provenance_per_participant = 64
max_retired_identity_slots_server = 1024
identity_slots = 4
observer_recovery_max_entries = 64
max_semantic_conversations_per_connection = 32
max_ordinary_record_entries = 1
max_ordinary_record_bytes = 131072
max_generated_marker_entries = 1
max_generated_marker_bytes = 4096
mandatory_transaction_bound_entries = 4
mandatory_transaction_bound_bytes = 16384
full_recovery_claim_entries = 4
full_recovery_claim_bytes = 16384
retained_capacity_entries = 2048
retained_capacity_bytes = 16777216
max_retained_record_rows = 1024
closure_episode_churn_limit = 1024
";