liminal-cli 0.3.1

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

use std::error::Error;
use std::fs::File;
use std::io::{BufRead, BufReader, ErrorKind, Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::sync::mpsc;
use std::thread::{self, JoinHandle};

use liminal::protocol::{
    CausalContext, Frame, MessageEnvelope, MessageId, ProtocolVersion, SchemaId, decode, encode,
    encoded_len,
};
use serde_json::Value;

const PARTICIPANT: &str = r"
[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
";

struct RunningServer {
    child: Child,
    reader: Option<JoinHandle<Result<(), String>>>,
    log: PathBuf,
}

impl RunningServer {
    fn start(root: &Path) -> Result<(Self, SocketAddr), Box<dyn Error>> {
        let wire_reservation = TcpListener::bind("127.0.0.1:0")?;
        let health_reservation = TcpListener::bind("127.0.0.1:0")?;
        let wire = wire_reservation.local_addr()?;
        let health = health_reservation.local_addr()?;
        let config = root.join("server.toml");
        std::fs::write(
            &config,
            format!(
                "listen_address = \"{wire}\"\nhealth_listen_address = \"{health}\"\n\
                 drain_timeout_ms = 5000\nrouting_rules = []\n\
                 [[channels]]\nname = \"ordinary\"\ndurable = true\n{PARTICIPANT}"
            ),
        )?;
        let log = root.join("server-stderr.log");
        let mut log_writer = File::create(&log)?;
        drop(wire_reservation);
        drop(health_reservation);
        let child = Command::new(env!("CARGO_BIN_EXE_liminal"))
            .arg("--config")
            .arg(config)
            .env("LIMINAL_PERSISTENCE_PATH", root)
            .env_remove("RUST_LOG")
            .stdin(Stdio::null())
            .stdout(Stdio::null())
            .stderr(Stdio::piped())
            .spawn()?;
        let mut server = Self {
            child,
            reader: None,
            log,
        };
        let stderr = server.child.stderr.take().ok_or("server stderr absent")?;
        let (ready, started) = mpsc::sync_channel(1);
        server.reader = Some(thread::spawn(move || {
            let mut ready = Some(ready);
            for line in BufReader::new(stderr).lines() {
                let line = line.map_err(|error| format!("server stderr read failed: {error}"))?;
                writeln!(log_writer, "{line}")
                    .map_err(|error| format!("server stderr log failed: {error}"))?;
                if line.contains("liminal server started") {
                    if let Some(ready) = ready.take() {
                        ready
                            .send(())
                            .map_err(|error| format!("startup receiver closed: {error}"))?;
                    }
                }
            }
            if ready.is_some() {
                return Err("server exited before its startup signal".to_owned());
            }
            Ok(())
        }));
        started.recv().map_err(|error| {
            format!(
                "ordinary server startup failed: {error}; stderr: {}",
                std::fs::read_to_string(&server.log)
                    .unwrap_or_else(|read_error| format!("stderr log read failed: {read_error}"))
            )
        })?;
        Ok((server, wire))
    }

    fn join_reader(&mut self) -> Result<(), Box<dyn Error>> {
        if let Some(reader) = self.reader.take() {
            reader
                .join()
                .map_err(|_| "server stderr reader panicked")??;
        }
        Ok(())
    }

    fn stop(mut self) -> Result<(), Box<dyn Error>> {
        let signalled = Command::new("/bin/kill")
            .arg("-TERM")
            .arg(self.child.id().to_string())
            .status()?;
        if !signalled.success() {
            return Err(format!("owned server termination refused: {signalled}").into());
        }
        let status = self.child.wait()?;
        self.join_reader()?;
        if !status.success() {
            return Err(format!(
                "ordinary server shutdown failed: {status}; stderr: {}",
                std::fs::read_to_string(&self.log)?
            )
            .into());
        }
        Ok(())
    }
}

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

struct Peer {
    stream: TcpStream,
    buffer: Vec<u8>,
}

impl Peer {
    fn connect(address: SocketAddr) -> Result<Self, Box<dyn Error>> {
        let mut peer = Self {
            stream: TcpStream::connect(address)?,
            buffer: Vec::new(),
        };
        peer.send(&Frame::Connect {
            flags: 0,
            min_version: ProtocolVersion::new(1, 0),
            max_version: ProtocolVersion::new(1, 0),
            auth_token: Vec::new(),
        })?;
        assert!(matches!(peer.read()?, Frame::ConnectAck { .. }));
        Ok(peer)
    }

    fn send(&mut self, frame: &Frame) -> Result<(), Box<dyn Error>> {
        let mut bytes = vec![0; encoded_len(frame)?];
        let written = encode(frame, &mut bytes)?;
        self.stream
            .write_all(bytes.get(..written).ok_or("invalid encoded frame length")?)?;
        Ok(())
    }

    fn read(&mut self) -> Result<Frame, Box<dyn Error>> {
        loop {
            match decode(&self.buffer) {
                Ok((frame, consumed)) => {
                    self.buffer.drain(..consumed);
                    return Ok(frame);
                }
                Err(
                    liminal::protocol::ProtocolError::IncompleteHeader { .. }
                    | liminal::protocol::ProtocolError::TruncatedPayload { .. },
                ) => {
                    let mut chunk = [0; 8192];
                    let count = self.stream.read(&mut chunk)?;
                    if count == 0 {
                        return Err("ordinary wire closed before its frame arrived".into());
                    }
                    if self.buffer.len() + count > 65_536 {
                        return Err("ordinary wire proof frame exceeds its bound".into());
                    }
                    self.buffer.extend_from_slice(&chunk[..count]);
                }
                Err(error) => return Err(error.into()),
            }
        }
    }

    fn subscribe(&mut self, stream_id: u32, channel: &str) -> Result<u64, Box<dyn Error>> {
        self.send(&Frame::Subscribe {
            flags: 0,
            stream_id,
            channel: channel.to_owned(),
            accepted_schemas: Vec::new(),
            max_in_flight: 1,
        })?;
        match self.read()? {
            Frame::SubscribeAck {
                stream_id: actual,
                subscription_id,
                ..
            } if actual == stream_id => Ok(subscription_id),
            frame => Err(format!("unexpected subscription acknowledgement: {frame:?}").into()),
        }
    }

    fn first_cost_window(&mut self, version: &str) -> Result<(Value, String, u64), Box<dyn Error>> {
        self.subscribe(1, "liminal.cost")?;
        let subscription_id = self.subscribe(3, "liminal.events")?;
        // The earlier connection event may predate its subscription. This
        // later connection produces a native identity event after SubscribeAck.
        drop(Self::connect(self.stream.peer_addr()?)?);
        self.send(&Frame::new_publish(
            2,
            "ordinary",
            MessageEnvelope::new(
                SchemaId::from_schema_bytes(b"{}"),
                CausalContext::default(),
                b"{}".to_vec(),
            ),
        )?)?;
        let mut published = false;
        let mut tick: Option<(u64, Value)> = None;
        let mut commit = None;
        let mut events_closed = false;
        // The publication acknowledgement, native identity event and cost tick
        // may arrive in any order on this connection.
        while !published || tick.is_none() || !events_closed {
            match self.read()? {
                Frame::PublishAck { stream_id: 2, .. } if !published => published = true,
                Frame::Deliver {
                    stream_id: 3,
                    envelope,
                    ..
                } => {
                    let observed = server_identity(&envelope.payload, version)?;
                    if let Some(expected) = &commit {
                        assert_eq!(&observed, expected);
                    } else {
                        commit = Some(observed);
                        self.send(&Frame::Unsubscribe {
                            flags: 0,
                            stream_id: 3,
                            subscription_id,
                        })?;
                        self.send(&Frame::new_ping(0)?)?;
                    }
                }
                Frame::Pong { .. } if commit.is_some() && !events_closed => {
                    events_closed = true;
                }
                Frame::Deliver {
                    stream_id: 1,
                    delivery_seq,
                    envelope,
                    ..
                } if tick.is_none() => {
                    tick = Some((delivery_seq, serde_json::from_slice(&envelope.payload)?));
                }
                frame => {
                    return Err(format!("unexpected first cost-window frame: {frame:?}").into());
                }
            }
        }
        let (sequence, cost) =
            tick.ok_or("production owner did not deliver its native cost tick")?;
        Ok((
            cost,
            commit.ok_or("server native build identity absent")?,
            sequence,
        ))
    }

    fn acknowledge_and_check_idle(&mut self, sequence: u64) -> Result<(), Box<dyn Error>> {
        self.send(&Frame::Accept {
            flags: 0,
            stream_id: 1,
            referenced_message_id: MessageId::new(sequence.to_string()),
        })?;
        self.send(&Frame::new_ping(0)?)?;
        assert!(matches!(self.read()?, Frame::Pong { .. }));
        assert!(
            self.buffer.is_empty(),
            "cost acknowledgement queued an extra frame"
        );
        self.stream.set_nonblocking(true)?;
        let mut pending = [0];
        match self.stream.peek(&mut pending) {
            Err(error) if error.kind() == ErrorKind::WouldBlock => Ok(()),
            result => Err(format!("ordinary peer did not return to idle: {result:?}").into()),
        }
    }
}

fn ordinary_tick(address: SocketAddr, server_version: &str) -> Result<(), Box<dyn Error>> {
    let mut peer = Peer::connect(address)?;
    let (cost, build_commit, delivery_seq) = peer.first_cost_window(server_version)?;
    assert_eq!(cost["v"], 1);
    assert_eq!(cost["publisher"], "liminal.local");
    assert_eq!(cost["device"], "local");
    assert_eq!(cost["crate_version"], server_version);
    assert_eq!(cost["build_commit"], build_commit);
    assert!(
        cost["server_incarnation"]
            .as_u64()
            .ok_or("missing incarnation")?
            > 0
    );
    assert!(
        cost["server_time_ms"]
            .as_u64()
            .ok_or("missing server time")?
            > 0
    );
    // `StoreCounts::named` lists 22 counters: the 17 store counters this pin
    // first carried, `read_stream_next_seq_calls`, and the four routed ones
    // (`apply_routed_batch_calls`, `apply_routed_batch_items`,
    // `get_routed_calls`, `get_routed_rows`).
    assert_eq!(
        cost["counts"].as_object().ok_or("missing counters")?.len(),
        22
    );
    let previous = cost["previous_snapshot"]
        .as_object()
        .ok_or("missing baseline")?;
    assert_eq!(previous.len(), 22);
    assert!(cost["counts"]["read_stream_next_seq_calls"].is_u64());
    assert_eq!(previous["read_stream_next_seq_calls"], 0);
    for counter in ["truncate_calls", "truncate_rows"] {
        assert_eq!(cost["counts"][counter], 0);
        assert_eq!(previous.get(counter), Some(&Value::from(0)));
    }
    assert!(previous.values().all(|value| value == 0));
    assert!(
        cost["counts"]["append_rows"]
            .as_u64()
            .ok_or("missing appended rows")?
            > cost["self_rows"].as_u64().ok_or("missing own row count")?
    );
    peer.acknowledge_and_check_idle(delivery_seq)
}

fn server_version() -> Result<String, Box<dyn Error>> {
    let manifest: toml::Value = toml::from_str(include_str!("../../liminal-server/Cargo.toml"))?;
    manifest
        .get("package")
        .and_then(|package| package.get("version"))
        .and_then(toml::Value::as_str)
        .map(str::to_owned)
        .ok_or_else(|| "server manifest version absent".into())
}

fn server_identity(payload: &[u8], server_version: &str) -> Result<String, Box<dyn Error>> {
    // Native events and cost records report the same embedded build stamp.
    let identity: Value = serde_json::from_slice(payload)?;
    assert_eq!(identity["v"], 1);
    assert_eq!(identity["crate_version"], server_version);
    let commit = identity["build_commit"]
        .as_str()
        .ok_or("server build commit absent")?;
    assert!(
        commit == "unknown"
            || (matches!(commit.len(), 40 | 64)
                && commit.bytes().all(|byte| byte.is_ascii_hexdigit())),
        "invalid server build commit: {commit}"
    );
    eprintln!("running server build_commit={commit} crate_version={server_version}");
    Ok(commit.to_owned())
}

/// The hard-coded interval needs the exceptional real ten-second delivery wait.
/// Startup, wire delivery and process completion are signals, with no test timer.
#[test]
fn ordinary_start_production_cost_tick_uses_the_fixed_ten_second_exception()
-> Result<(), Box<dyn Error>> {
    let version = server_version()?;
    let root = test_persistence::directory()?;
    let (server, address) = RunningServer::start(root.path())?;
    let proof = ordinary_tick(address, &version);
    let stopped = server.stop();
    match (proof, stopped) {
        (Ok(()), Ok(())) => Ok(()),
        (Err(proof), Err(stopped)) => {
            Err(format!("tick proof: {proof}; cleanup: {stopped}").into())
        }
        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
    }
}

#[test]
fn build_metadata_matches_workspace_head_and_emits_rebuild_watches() -> Result<(), Box<dyn Error>> {
    let directory = Path::new(env!("CARGO_MANIFEST_DIR")).join("../liminal-server");
    let metadata = build_script::build_metadata(&directory);
    assert!(metadata.warning.is_none(), "{:?}", metadata.warning);
    let head = Command::new("git")
        .current_dir(&directory)
        .args(["rev-parse", "HEAD"])
        .output()?;
    assert!(head.status.success(), "git HEAD failed: {}", head.status);
    assert_eq!(metadata.commit, std::str::from_utf8(&head.stdout)?.trim());
    eprintln!(
        "build commit={} watched={:?}",
        metadata.commit, metadata.watched
    );
    let mut output = Vec::new();
    build_script::write_metadata(&metadata, &mut output)?;
    let output = String::from_utf8(output)?;
    assert!(output.lines().any(|line| line == format!("cargo:rustc-env=LIMINAL_BUILD_COMMIT={}", metadata.commit)));
    assert_eq!(
        metadata.watched.len(),
        2,
        "main must watch HEAD and its branch reference"
    );
    for path in metadata.watched {
        assert!(
            output
                .lines()
                .any(|line| line == format!("cargo:rerun-if-changed={}", path.display())),
            "build metadata omitted git rebuild watch: {}",
            path.display()
        );
    }
    assert!(
        output
            .lines()
            .any(|line| line == "cargo:rerun-if-changed=build.rs")
    );
    let root = test_persistence::directory()?;
    let (server, address) = RunningServer::start(root.path())?;
    let (_, embedded_commit, _) = Peer::connect(address)?.first_cost_window(&server_version()?)?;
    server.stop()?;
    assert_eq!(
        embedded_commit, metadata.commit,
        "server build must match the workspace build HEAD"
    );
    Ok(())
}

#[test]
fn server_identity_uses_the_reported_build_when_workspace_head_differs()
-> Result<(), Box<dyn Error>> {
    let metadata = build_script::build_metadata(Path::new(env!("CARGO_MANIFEST_DIR")));
    let commit = if metadata.commit == "0".repeat(40) {
        "1".repeat(40)
    } else {
        "0".repeat(40)
    };
    let payload = serde_json::to_vec(&serde_json::json!({
        "v": 1, "crate_version": "identity-proof", "build_commit": commit
    }))?;
    assert_eq!(server_identity(&payload, "identity-proof")?, commit);
    Ok(())
}