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_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
);
assert_eq!(
cost["counts"].as_object().ok_or("missing counters")?.len(),
17
);
let previous = cost["previous_snapshot"]
.as_object()
.ok_or("missing baseline")?;
assert_eq!(previous.len(), 17);
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)
}
#[test]
fn ordinary_start_production_cost_tick_uses_the_fixed_ten_second_exception()
-> Result<(), Box<dyn Error>> {
let root = tempfile::tempdir()?;
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),
}
}