#[path = "support/agent_pass.rs"]
mod agent_pass;
#[path = "link_open_live/fixture.rs"]
mod fixture;
#[path = "runtime_links/logs.rs"]
mod logs;
#[path = "runtime_links/peer.rs"]
mod peer;
use fixture::Fixture;
use liminal_server::health::{ReadinessState, SharedReadinessState, start_health_server};
use liminal_server::link_open::ChoiceConfig;
use std::io::{Read, Write};
use std::net::TcpStream;
use std::time::Duration;
type Result<T = ()> = std::result::Result<T, Box<dyn std::error::Error>>;
const OBSERVATION: Duration = Duration::from_secs(8);
fn configured() -> Result<Fixture> {
let mut fixture = Fixture::new()?;
assert!(fixture.servers[0].bindings.is_none());
fixture.configure(Some(fixture.servers[1].address()?.to_string()))?;
fixture.servers[0].machine()?.carrier_choice = ChoiceConfig::default();
Ok(fixture)
}
fn link(document: &serde_json::Value) -> Result<&serde_json::Value> {
document["links"]
.as_array()
.ok_or("links absent")?
.iter()
.find(|link| link["kind"] == "federated")
.ok_or_else(|| "link absent".into())
}
#[test]
fn an_absent_peer_leaves_the_server_serving_with_named_down_health_and_topology() -> Result {
let mut fixture = configured()?;
fixture.servers[1].pause_listener()?;
let logs = logs::Logs::default();
let dispatcher = logs.dispatcher();
let guard = tracing::dispatcher::set_default(&dispatcher);
let health = start_health_server(
"127.0.0.1:0".parse()?,
SharedReadinessState::new(ReadinessState::default()),
)?;
fixture.servers[0].registry = health.link_bridges();
fixture.servers[0].start_links()?;
fixture.servers[0].publish(7)?;
let mut socket = TcpStream::connect(health.local_addr())?;
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"), "{response}");
let body: serde_json::Value =
serde_json::from_str(response.split_once("\r\n\r\n").ok_or("body absent")?.1)?;
let counts = &body["links"]["orders-link"];
assert_eq!(counts["links_opened"], 0);
assert_eq!(counts["dial_refused"], 1);
assert!(
counts["last_refusal"]
.as_str()
.ok_or("reason absent")?
.starts_with("all_link_carriers_refused")
);
let topology = fixture.servers[0].topology()?;
assert_eq!(link(&topology)?["state"], "down");
assert_eq!(link(&topology)?["reason"], counts["last_refusal"]);
health.shutdown()?;
logs.assert_one_refusal()?;
drop(guard);
Ok(())
}
#[test]
fn a_peer_started_later_opens_and_forwards_without_restarting_the_server() -> Result {
let mut fixture = configured()?;
fixture.servers[1].pause_listener()?;
fixture.servers[0].start_links()?;
fixture.servers[0].publish(11)?;
fixture.servers[1].resume_listener()?;
assert!(fixture.servers[0].registry.observe(OBSERVATION, |counts| {
counts
.get("orders-link")
.is_some_and(|counts| counts.lifecycle.links_opened >= 1)
})?);
assert!(
fixture.servers[0]
.links
.as_ref()
.ok_or("links absent")?
.observe_answers(1, OBSERVATION)
);
let counts = fixture.servers[0].registry.snapshot();
assert_eq!(counts["orders-link"].lifecycle.links_opened, 1);
assert_eq!(counts["orders-link"].crossed, 1);
assert_eq!(link(&fixture.servers[0].topology()?)?["state"], "up");
Ok(())
}
#[test]
fn peer_gone_reopens_the_bridge_and_retains_two_opens_and_one_close() -> Result {
let mut fixture = configured()?;
let logs = logs::Logs::default();
let dispatcher = logs.dispatcher();
let guard = tracing::dispatcher::set_default(&dispatcher);
fixture.servers[0].start_links()?;
fixture.servers[1].pause_listener()?;
fixture.servers[1].resume_listener()?;
fixture.servers[0].publish(13)?;
assert!(fixture.servers[0].registry.observe(OBSERVATION, |counts| {
counts
.get("orders-link")
.is_some_and(|counts| counts.lifecycle.links_opened >= 2)
})?);
assert!(
fixture.servers[0]
.links
.as_ref()
.ok_or("links absent")?
.observe_answers(1, OBSERVATION)
);
let counts = fixture.servers[0].registry.snapshot();
assert_eq!(counts["orders-link"].lifecycle.links_opened, 2);
assert_eq!(counts["orders-link"].lifecycle.links_closed_by_peer_gone, 1);
assert_eq!(counts["orders-link"].crossed, 1);
let topology = fixture.servers[0].topology()?;
assert_eq!(link(&topology)?["state"], "up");
assert_eq!(link(&topology)?["links_opened"], 2);
let text = logs.text()?;
assert!(!text.contains("ERROR"), "{text}");
drop(guard);
Ok(())
}
#[test]
fn an_unknown_peer_holder_remains_a_configuration_fault() -> Result {
let mut fixture = configured()?;
fixture.servers[0].machine()?.links[0].peer_holder = "unknown".to_owned();
let error = fixture.servers[0]
.start_links()
.err()
.ok_or("unknown peer started")?;
assert!(
error
.to_string()
.contains("configuration validation failed: peer_has_no_recorded_machine")
);
Ok(())
}
#[test]
fn configured_links_without_installed_lys_remain_a_configuration_fault() -> Result {
let root = tempfile::tempdir()?;
let config: liminal_server::config::ServerConfig = serde_json::from_value(serde_json::json!({
"listen_address": "127.0.0.1:0", "health_listen_address": "127.0.0.1:0",
"persistence_path": root.path(), "channels": [{"name": "orders", "durable": true}], "routing_rules": [], "drain_timeout_ms": 0,
"auth": {"token": "fixture-bearer", "lys": {"issuer": "https://carrier-live.example",
"network_id": "network-a", "keys_path": root.path().join("keys.json"),
"machine": {"identity_path": root.path().join("identity"), "pass_path": root.path().join("pass"),
"machines": {}, "links": [{"link_id": "orders-link", "peer_holder": "importer", "exports": ["orders"]}]}}}
}))?;
let services = std::sync::Arc::new(
liminal_server::server::connection::LiminalConnectionServices::from_config(&config)?,
);
let result =
liminal_server::server::runtime::link_test_support::without_lys(&config, &services);
services.conversation_supervisor().shutdown();
services.channel_cluster().supervisor().shutdown();
let error = result.err().ok_or("Lys was optional")?;
assert!(error.to_string().contains(
"configuration validation failed: configured outbound links have no installed Lys"
));
Ok(())
}
#[test]
fn a_peer_staying_away_is_retried_silently_and_shutdown_interrupts_backoff() -> Result {
let mut fixture = configured()?;
fixture.servers[1].pause_listener()?;
let logs = logs::Logs::default();
let dispatcher = logs.dispatcher();
let guard = tracing::dispatcher::set_default(&dispatcher);
fixture.servers[0].start_links()?;
assert!(fixture.servers[0].registry.observe(OBSERVATION, |counts| {
counts.get("orders-link").is_some_and(|counts| {
serde_json::to_value(counts).is_ok_and(|counts| {
counts["dial_refused"]
.as_u64()
.is_some_and(|count| count >= 2)
})
})
})?);
let before = fixture.servers[0].registry.snapshot();
fixture.servers[0]
.links
.take()
.ok_or("links absent")?
.finish()?;
let after = fixture.servers[0].registry.snapshot();
assert_eq!(before["orders-link"].lifecycle.links_opened, 0);
assert_eq!(before, after, "shutdown admitted another dial");
logs.assert_one_refusal()?;
drop(guard);
Ok(())
}
#[test]
fn shutdown_cancels_a_retry_waiting_on_peer_admission() -> Result {
let mut fixture = configured()?;
fixture.servers[1].pause_listener()?;
fixture.servers[0].start_links()?;
let listener = std::net::TcpListener::bind(fixture.servers[1].config.listen_address)?;
let (ready, observed) = std::sync::mpsc::sync_channel(1);
let peer = std::thread::spawn(move || -> std::result::Result<(), String> {
let (mut socket, _) = listener.accept().map_err(|error| error.to_string())?;
socket
.set_read_timeout(Some(OBSERVATION))
.map_err(|error| error.to_string())?;
assert!(matches!(
fixture::wire::read_frame(&mut socket).map_err(|error| error.to_string())?,
liminal::protocol::Frame::Connect { .. }
));
ready.send(()).map_err(|error| error.to_string())?;
let mut byte = [0];
assert_eq!(
socket.read(&mut byte).map_err(|error| error.to_string())?,
0
);
Ok(())
});
observed.recv_timeout(OBSERVATION)?;
fixture.servers[0]
.links
.take()
.ok_or("links absent")?
.finish()?;
peer.join()
.map_err(|_| "peer observer panicked")?
.map_err(std::io::Error::other)?;
Ok(())
}
#[test]
fn a_peer_refusing_once_then_accepting_opens_on_the_next_dial_without_exporter_restart() -> Result {
let mut fixture = Fixture::with_open_refusals(1)?;
fixture.configure(Some(fixture.servers[1].address()?.to_string()))?;
fixture.servers[0].machine()?.carrier_choice = ChoiceConfig::default();
fixture.servers[0].start_links()?;
let refused = fixture.servers[0].registry.snapshot();
assert_eq!(refused["orders-link"].refused, 1);
assert_eq!(refused["orders-link"].dial_refused, 0);
assert_eq!(refused["orders-link"].lifecycle.links_opened, 0);
assert!(
refused["orders-link"]
.last_refusal
.as_deref()
.is_some_and(|reason| reason.starts_with("link_refused:"))
);
fixture.servers[0].publish(29)?;
assert!(fixture.servers[0].registry.observe(OBSERVATION, |counts| {
counts
.get("orders-link")
.is_some_and(|counts| counts.lifecycle.links_opened == 1)
})?);
assert!(
fixture.servers[0]
.links
.as_ref()
.ok_or("links absent")?
.observe_answers(1, OBSERVATION)
);
assert!(fixture.servers[0].registry.observe(OBSERVATION, |counts| {
counts
.get("orders-link")
.is_some_and(|counts| counts.crossed == 1)
})?);
let counts = fixture.servers[0].registry.snapshot();
assert_eq!(counts["orders-link"].refused, 1);
assert_eq!(counts["orders-link"].dial_refused, 0);
assert_eq!(counts["orders-link"].crossed, 1);
assert_eq!(counts["orders-link"].lifecycle.links_opened, 1);
assert_eq!(link(&fixture.servers[0].topology()?)?["state"], "up");
Ok(())
}