use std::error::Error;
use super::services::subsystem_census::RecordingSubsystems;
use super::services::{LiminalConnectionServices, build_connection_services_via};
use crate::ServerError;
use crate::config::{ServerConfig, validate};
type TestResult = Result<(), Box<dyn Error>>;
fn example() -> Result<ServerConfig, Box<dyn Error>> {
Ok(toml::from_str(include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../config/liminal.example.toml"
)))?)
}
fn refusal<T>(result: Result<T, ServerError>) -> TestResult {
let Err(ServerError::ConfigValidation { message }) = result else {
return Err("missing persistence path did not yield ConfigValidation".into());
};
assert!(message.contains("persistence_path"), "{message}");
assert!(message.contains("LIMINAL_PERSISTENCE_PATH"), "{message}");
assert!(message.contains("no store is opened"), "{message}");
Ok(())
}
#[test]
fn full_profile_validation_refuses_without_a_path() -> TestResult {
let mut config = example()?;
assert!(config.persistence_path.is_none());
refusal(validate(&mut config, None))
}
#[test]
fn missing_path_is_accumulated_with_other_validation_errors() -> TestResult {
let mut config = example()?;
config.listen_address.set_port(0);
let result = validate(&mut config, None);
assert!(
matches!(&result, Err(ServerError::ConfigValidation { message })
if message.contains("listen_address: port must be non-zero"))
);
refusal(result)
}
#[test]
fn profile_dispatch_refuses_before_any_subsystem_or_store() -> TestResult {
let root = tempfile::tempdir()?;
let factory = RecordingSubsystems::rooted(root.path());
let result = build_connection_services_via(&example()?, &factory);
let empty = std::fs::read_dir(root.path())?.count();
let constructed = factory.recorded();
refusal(result)?;
assert_eq!(empty, 0);
assert!(constructed.is_empty(), "{constructed:?}");
Ok(())
}
#[test]
fn direct_config_construction_refuses_without_a_path() -> TestResult {
refusal(LiminalConnectionServices::from_config(&example()?))
}
fn worker() -> Result<ServerConfig, Box<dyn Error>> {
let mut config = example()?;
config.channels.clear();
config.routing_rules.clear();
config.cluster = None;
config.participant = None;
config.websocket = None;
config.services.profile = "worker-front-door".to_owned();
Ok(config)
}
#[test]
fn worker_front_door_needs_no_path_and_builds_no_store() -> TestResult {
let mut config = worker()?;
validate(&mut config, None)?;
let root = tempfile::tempdir()?;
let factory = RecordingSubsystems::rooted(root.path());
let services = build_connection_services_via(&config, &factory)?;
assert!(!services.supports_channel_operations());
assert!(factory.recorded().is_empty());
assert_eq!(std::fs::read_dir(root.path())?.count(), 0);
Ok(())
}
#[test]
fn worker_front_door_still_refuses_a_path() -> TestResult {
let home = tempfile::tempdir()?;
let mut config = worker()?;
config.persistence_path = Some(home.path().to_path_buf());
assert!(matches!(validate(&mut config, None),
Err(ServerError::ConfigValidation { message }) if message.contains("builds no durable store")));
let root = tempfile::tempdir()?;
let factory = RecordingSubsystems::rooted(root.path());
assert!(matches!(build_connection_services_via(&config, &factory),
Err(ServerError::ConfigValidation { message }) if message.contains("builds no durable store")));
assert!(factory.recorded().is_empty());
assert_eq!(std::fs::read_dir(root.path())?.count(), 0);
Ok(())
}
#[test]
fn durable_constructor_has_no_optional_path_or_ephemeral_branch() {
let source = include_str!("services.rs");
let start = source.find("fn build_durable_store(");
assert!(start.is_some(), "durable constructor must exist");
if let Some(start) = start {
let body = &source[start..];
let end = body.find("\n}");
assert!(end.is_some(), "durable constructor must have a body");
if let Some(end) = end {
let constructor = &body[..end];
assert!(constructor.contains("&Path"));
assert!(!constructor.contains("Option<"));
assert!(!constructor.contains("open_ephemeral"));
}
}
}
#[test]
fn pathless_constructors_require_explicit_test_support() -> TestResult {
let manifest: toml::Value = toml::from_str(include_str!(concat!(
env!("CARGO_MANIFEST_DIR"),
"/Cargo.toml"
)))?;
assert!(manifest["features"].get("test-support").is_some());
for (source, function) in [
(include_str!("services.rs"), "pub fn empty()"),
(
include_str!("supervisor/supervisor_config.rs"),
"pub fn new()",
),
] {
let prefix = source
.split(function)
.next()
.ok_or("constructor is absent")?;
assert!(
prefix.ends_with("#[cfg(any(test, feature = \"test-support\"))]\n "),
"pathless constructor must be gated"
);
}
Ok(())
}
fn root_entries(path: &std::path::Path) -> Result<Vec<std::path::PathBuf>, Box<dyn Error>> {
let mut paths = Vec::new();
for entry in std::fs::read_dir(path.join("ephemeral"))? {
let entry = entry?;
if entry
.file_name()
.to_string_lossy()
.starts_with("liminal-durability-")
{
paths.push(entry.path());
}
}
paths.sort();
Ok(paths)
}
fn pin_config(path: &std::path::Path) -> Result<ServerConfig, Box<dyn Error>> {
let mut config = example()?;
config.persistence_path = Some(path.to_path_buf());
config.cluster = None;
config.participant = None;
config.auth = None;
config.websocket = None;
Ok(config)
}
struct ServerChild {
process: std::process::Child,
output: Option<std::io::BufReader<std::process::ChildStdout>>,
ready: serde_json::Value,
log: std::path::PathBuf,
}
impl Drop for ServerChild {
fn drop(&mut self) {
match self.process.try_wait() {
Ok(Some(_)) => return,
Ok(None) => {}
Err(error) => tracing::error!(%error, "server pin child status failed"),
}
if let Err(error) = self.process.kill() {
tracing::error!(%error, "server pin child kill failed");
}
if let Err(error) = self.process.wait() {
tracing::error!(%error, "server pin child reap failed");
}
}
}
fn server_child(
home: &std::path::Path,
temporary: &std::path::Path,
log: &std::path::Path,
mode: &str,
) -> Result<ServerChild, Box<dyn Error>> {
use std::io::BufRead;
let process = std::process::Command::new(std::env::current_exe()?)
.args([
"--exact",
"server::connection::tests_no_path_refusal::ephemeral_server_child",
"--nocapture",
])
.env("LIMINAL_SERVER_ROOT_PIN", home)
.env("LIMINAL_SERVER_ROOT_MODE", mode)
.env("TMPDIR", temporary)
.stdin(std::process::Stdio::piped())
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::from(std::fs::File::create(log)?))
.spawn()?;
let mut child = ServerChild {
process,
output: None,
ready: serde_json::Value::Null,
log: log.to_path_buf(),
};
child.output = Some(std::io::BufReader::new(
child
.process
.stdout
.take()
.ok_or("server pin stdout absent")?,
));
loop {
let mut line = String::new();
if child
.output
.as_mut()
.ok_or("server pin reader absent")?
.read_line(&mut line)?
== 0
{
break;
}
if let Some(json) = line.trim_end().strip_prefix("server-root-ready ") {
child.ready = serde_json::from_str(json)?;
return Ok(child);
}
}
Err(format!(
"server child ended before ready: {}",
std::fs::read_to_string(log)?
)
.into())
}
fn tell_server(child: &mut ServerChild, byte: u8) -> TestResult {
use std::io::Write;
child
.process
.stdin
.as_mut()
.ok_or("server pin stdin absent")?
.write_all(&[byte])?;
Ok(())
}
fn health_answer(child: &ServerChild) -> Result<serde_json::Value, Box<dyn Error>> {
use std::io::{Read, Write};
let address = child.ready["health"]
.as_str()
.ok_or("server health address absent")?;
let mut stream = std::net::TcpStream::connect(address)?;
stream.write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n")?;
let mut text = String::new();
stream.read_to_string(&mut text)?;
assert!(text.starts_with("HTTP/1.1 200"), "{text}");
let (_, body) = text.split_once("\r\n\r\n").ok_or("health body absent")?;
Ok(serde_json::from_str(body)?)
}
fn disk_bytes(path: &std::path::Path) -> Result<u64, Box<dyn Error>> {
let mut bytes = 0_u64;
for entry in std::fs::read_dir(path)? {
let entry = entry?;
let metadata = entry.metadata()?;
bytes += if metadata.is_dir() {
disk_bytes(&entry.path())?
} else {
metadata.len()
};
}
Ok(bytes)
}
#[test]
fn recorded_production_start_roots_lazy_stores_and_never_touches_tmpdir() -> TestResult {
let home = tempfile::tempdir()?;
let temporary = tempfile::tempdir()?;
let evidence = tempfile::tempdir()?;
drop(LiminalConnectionServices::from_config(&pin_config(
home.path(),
)?)?);
let mut child = server_child(
home.path(),
temporary.path(),
&evidence.path().join("server.log"),
"publish",
)?;
assert_eq!(child.ready["before"], 0);
let entries = root_entries(home.path())?;
assert_eq!(entries.len(), 1);
assert!(entries[0].starts_with(home.path().join("ephemeral")));
assert_eq!(std::fs::read_dir(temporary.path())?.count(), 0);
assert_eq!(
health_answer(&child)?["ephemeral_sweep"],
serde_json::json!({"removed_count":0,"removed_bytes":0,"held_count":0})
);
assert!(std::fs::read_to_string(&child.log)?.contains("ephemeral root startup sweep"));
tell_server(&mut child, b'S')?;
assert!(child.process.wait()?.success());
assert!(root_entries(home.path())?.is_empty());
assert!(home.path().join("ephemeral/root.lock").is_file());
Ok(())
}
#[test]
fn crashed_production_entry_is_removed_with_matching_log_and_retained_health_counts() -> TestResult
{
let home = tempfile::tempdir()?;
let temporary = tempfile::tempdir()?;
let evidence = tempfile::tempdir()?;
drop(LiminalConnectionServices::from_config(&pin_config(
home.path(),
)?)?);
let mut crashed = server_child(
home.path(),
temporary.path(),
&evidence.path().join("crash.log"),
"publish",
)?;
let entries = root_entries(home.path())?;
assert_eq!(entries.len(), 1);
tell_server(&mut crashed, b'C')?;
assert_eq!(crashed.process.wait()?.code(), Some(73));
assert!(entries[0].exists());
let bytes = disk_bytes(&entries[0])?;
assert!(bytes > 0);
let mut rebooted = server_child(
home.path(),
temporary.path(),
&evidence.path().join("reboot.log"),
"quiet",
)?;
let expected = serde_json::json!({"removed_count":1,"removed_bytes":bytes,"held_count":0});
assert_eq!(health_answer(&rebooted)?["ephemeral_sweep"], expected);
assert_eq!(health_answer(&rebooted)?["ephemeral_sweep"], expected);
assert!(!entries[0].exists());
let log = std::fs::read_to_string(&rebooted.log)?;
assert!(log.contains(&entries[0].display().to_string()), "{log}");
assert!(log.contains(&format!("bytes={bytes}")), "{log}");
assert!(log.contains("removed_count=1"), "{log}");
tell_server(&mut rebooted, b'S')?;
assert!(rebooted.process.wait()?.success());
Ok(())
}
#[test]
fn a_live_entry_holder_is_kept_and_appears_in_production_health() -> TestResult {
let home = tempfile::tempdir()?;
let temporary = tempfile::tempdir()?;
let evidence = tempfile::tempdir()?;
std::fs::create_dir(home.path().join("ephemeral"))?;
let mut holder = server_child(
home.path(),
temporary.path(),
&evidence.path().join("holder.log"),
"holder",
)?;
let entries = root_entries(home.path())?;
assert_eq!(entries.len(), 1);
let mut server = server_child(
home.path(),
temporary.path(),
&evidence.path().join("server.log"),
"quiet",
)?;
assert_eq!(
health_answer(&server)?["ephemeral_sweep"],
serde_json::json!({"removed_count":0,"removed_bytes":0,"held_count":1})
);
assert!(entries[0].is_dir());
let log = std::fs::read_to_string(&server.log)?;
assert!(log.contains(&entries[0].display().to_string()), "{log}");
assert!(log.contains("held_count=1"), "{log}");
tell_server(&mut server, b'S')?;
assert!(server.process.wait()?.success());
tell_server(&mut holder, b'S')?;
assert!(holder.process.wait()?.success());
Ok(())
}
#[test]
fn competing_production_start_refuses_root_before_opening_the_database() -> TestResult {
let home = tempfile::tempdir()?;
let root = liminal::durability::EphemeralRoot::open(home.path())?;
let entry = home.path().join("ephemeral/liminal-durability-dead");
std::fs::create_dir(&entry)?;
std::fs::write(entry.join("payload"), b"retained")?;
let result = super::services::build_connection_services(&pin_config(home.path())?);
let error = result.err().ok_or("competing production start succeeded")?;
assert!(
error.to_string().contains("ephemeral root lock held"),
"{error}"
);
assert!(
!home.path().join("durability").exists(),
"database opener was reached before root refusal"
);
assert!(entry.is_dir());
assert_eq!(root.sweep_counts().removed_count, 0);
Ok(())
}
#[test]
fn production_factories_cannot_reach_the_unrooted_test_opener() -> TestResult {
let source = include_str!("services.rs");
let body = source
.split("impl SubsystemFactory for RootedProductionSubsystems {")
.nth(1)
.ok_or("root-owning production factory absent")?
.split("/// Test-only recording implementation")
.next()
.ok_or("factory ending absent")?;
assert!(body.contains("root.open_store("));
assert!(!body.contains("open_ephemeral("));
assert!(source.contains(
"#[cfg(any(test, feature = \"test-support\"))]\npub(super) struct ProductionSubsystems;"
));
Ok(())
}
#[test]
fn ephemeral_server_child() -> TestResult {
use std::io::{Read, Write};
let Some(home) = std::env::var_os("LIMINAL_SERVER_ROOT_PIN") else {
return Ok(());
};
let home = std::path::PathBuf::from(home);
let mode = std::env::var("LIMINAL_SERVER_ROOT_MODE")?;
tracing_subscriber::fmt()
.with_ansi(false)
.without_time()
.with_writer(std::io::stderr)
.try_init()
.map_err(|error| error.to_string())?;
if mode == "holder" {
let entry = home.join("ephemeral/liminal-durability-live-holder");
std::fs::create_dir(&entry)?;
std::fs::write(entry.join("payload"), b"alive")?;
let holder = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(entry.join("holder.lock"))?;
rustix::fs::flock(&holder, rustix::fs::FlockOperation::LockExclusive)?;
writeln!(std::io::stdout(), "server-root-ready {{}}")?;
std::io::stdout().flush()?;
std::io::stdin().read_exact(&mut [0_u8])?;
return Ok(());
}
let services = LiminalConnectionServices::from_config(&pin_config(&home)?)?;
let before = root_entries(&home)?.len();
let health = crate::health::endpoint::start_health_server(
"127.0.0.1:0".parse()?,
crate::health::checks::SharedReadinessState::default(),
)?;
health.install_ephemeral_sweep_report(
services
.ephemeral_sweep_counts()
.ok_or("production sweep report absent")?,
)?;
if mode == "publish" {
use super::services::ConnectionServices;
let counters = services
.durable_store()
.store_counts()
.ok_or("persistent counts absent")?;
let persistent_before = counters.snapshot();
let envelope = liminal::protocol::MessageEnvelope::new(
liminal::protocol::SchemaId::new([0_u8; liminal::protocol::SchemaId::WIRE_LEN]),
liminal::protocol::CausalContext::independent(),
b"{}".to_vec(),
);
let subscription = services.subscribe_handle_for_test("orders-audit")?;
services.publish(7, "orders-audit", &envelope, Some("ephemeral-root-pin"))?;
assert!(subscription.try_next()?.is_some());
assert_eq!(
counters.snapshot(),
persistent_before,
"ephemeral keyed publish touched persistent store"
);
drop(subscription);
} else {
assert_eq!(mode, "quiet");
}
let ready = serde_json::json!({"health":health.local_addr().to_string(),"before":before});
writeln!(std::io::stdout(), "server-root-ready {ready}")?;
std::io::stdout().flush()?;
let mut command = [0_u8];
std::io::stdin().read_exact(&mut command)?;
if command[0] == b'C' {
std::process::exit(73);
}
assert_eq!(command[0], b'S');
drop(services);
health.shutdown()?;
Ok(())
}