liminal-server 0.20.2

#[path = "../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-server"))
            .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 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) -> Result<(), Box<dyn Error>> {
    let mut peer = Peer::connect(address)?;
    peer.send(&Frame::Subscribe {
        flags: 0,
        stream_id: 1,
        channel: "liminal.cost".to_owned(),
        accepted_schemas: Vec::new(),
        max_in_flight: 1,
    })?;
    assert!(matches!(
        peer.read()?,
        Frame::SubscribeAck { stream_id: 1, .. }
    ));
    peer.send(&Frame::new_publish(
        2,
        "ordinary",
        MessageEnvelope::new(
            SchemaId::from_schema_bytes(b"{}"),
            CausalContext::default(),
            b"{}".to_vec(),
        ),
    )?)?;
    assert!(matches!(
        peer.read()?,
        Frame::PublishAck { stream_id: 2, .. }
    ));
    let Frame::Deliver {
        stream_id: 1,
        delivery_seq,
        envelope,
        ..
    } = peer.read()?
    else {
        return Err("production owner did not deliver its native cost tick".into());
    };
    let cost: Value = serde_json::from_slice(&envelope.payload)?;
    assert_eq!(cost["v"], 1);
    assert_eq!(cost["publisher"], "liminal.local");
    assert_eq!(cost["device"], "local");
    assert_eq!(cost["crate_version"], env!("CARGO_PKG_VERSION"));
    assert_eq!(cost["build_commit"], env!("LIMINAL_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)
}

/// 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 root = test_persistence::directory()?;
    let (server, address) = RunningServer::start(root.path())?;
    let proof = ordinary_tick(address);
    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),
    }
}