use liminal::protocol::{CausalContext, Frame, MessageEnvelope, SchemaId};
use liminal_server::config::{ChannelDef, LimitsConfig, ServerConfig, ServicesConfig};
use liminal_server::link_bridge::LinkBridge;
use liminal_server::link_dial::{LinkDialOutcome, LinkDialer};
use liminal_server::server::connection::services::{ConnectionServices, LiminalConnectionServices};
use liminal_server::server::{ConnectionSupervisor, ServerListener};
use std::io::{ErrorKind, Read, Write};
use std::net::TcpStream;
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::time::Duration;
#[path = "link_bridge/net023.rs"]
mod net023;
#[path = "link_bridge/net025.rs"]
mod net025;
pub mod support;
use support::federation::{Federation, LiveOpen, read_frame, write_frame};
type TestResult = Result<(), Box<dyn std::error::Error>>;
const OBSERVATION: Duration = Duration::from_secs(3);
#[derive(Clone, Default)]
struct CapturedLog(Arc<Mutex<Vec<u8>>>);
impl Write for CapturedLog {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.0
.lock()
.map_err(|_| std::io::Error::other("log lock poisoned"))?
.extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
fn health_counts(
address: std::net::SocketAddr,
) -> Result<serde_json::Value, Box<dyn std::error::Error>> {
let mut socket = TcpStream::connect(address)?;
socket.set_read_timeout(Some(OBSERVATION))?;
socket.write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n")?;
let mut response = String::new();
socket.read_to_string(&mut response)?;
assert!(response.starts_with("HTTP/1.1 200 "));
let (_, body) = response
.split_once("\r\n\r\n")
.ok_or("health body absent")?;
Ok(serde_json::from_str(body)?)
}
struct Exporter {
services: Arc<LiminalConnectionServices>,
listener: ServerListener,
}
impl Exporter {
fn boot(path: &Path) -> Result<Self, Box<dyn std::error::Error>> {
Self::boot_with_subjects(path, &[])
}
fn boot_with_subjects(path: &Path, extra: &[&str]) -> Result<Self, Box<dyn std::error::Error>> {
let config = ServerConfig {
listen_address: "127.0.0.1:0".parse()?,
extra_listen_addresses: Vec::new(),
health_listen_address: "127.0.0.1:0".parse()?,
channels: [
"orders",
"orders.created",
"orders.secret",
"orders_other",
"console.topology",
]
.into_iter()
.chain(extra.iter().copied())
.map(|name| ChannelDef {
name: name.to_owned(),
schema_ref: None,
durable: name != "console.topology",
loaded_schema: None,
})
.collect(),
routing_rules: Vec::new(),
persistence_path: Some(path.to_owned()),
cluster: None,
auth: None,
drain_timeout_ms: 0,
services: ServicesConfig::default(),
limits: LimitsConfig {
max_channels: Some(8),
..LimitsConfig::default()
},
participant: None,
websocket: None,
open: None,
templates: liminal_server::services::lifecycle::templates::TemplateSources::default(),
agents: None,
};
let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
let supervisor = ConnectionSupervisor::with_services(services.clone())?;
Ok(Self {
listener: ServerListener::bind(&config, supervisor)?,
services,
})
}
fn bridge(&self, pair: &Federation) -> Result<LinkBridge, Box<dyn std::error::Error>> {
let outcome =
LinkDialer::new(&pair.exporter.host, &pair.exporter.config).open(&pair.outbound())?;
let LinkDialOutcome::Opened(link) = outcome else {
return Err("link did not open".into());
};
Ok(LinkBridge::start(*link, self.services.clone())?)
}
fn publish(&self, subject: &str, payload: &[u8]) -> TestResult {
self.services.publish(
0,
subject,
&MessageEnvelope::new(
SchemaId::new([0; 32]),
CausalContext::independent(),
payload.to_vec(),
),
None,
)?;
Ok(())
}
fn stop(self) -> TestResult {
let supervisor = self.listener.supervisor();
self.listener.shutdown()?;
supervisor.force_close_active_connections();
supervisor.scheduler().shutdown();
self.services
.channel_cluster()
.supervisor()
.scheduler()
.shutdown();
self.services
.conversation_supervisor()
.scheduler()
.shutdown();
Ok(())
}
}
fn subscriber(pair: &Federation) -> Result<TcpStream, Box<dyn std::error::Error>> {
subscriber_on(pair, "orders.created")
}
fn subscriber_on(
pair: &Federation,
subject: &str,
) -> Result<TcpStream, Box<dyn std::error::Error>> {
let mut socket = pair.importer.connect_as(&pair.importer.config.pass_path)?;
socket.set_read_timeout(Some(OBSERVATION))?;
write_frame(
&mut socket,
&Frame::Subscribe {
flags: 0,
stream_id: 1,
channel: subject.to_owned(),
accepted_schemas: Vec::new(),
max_in_flight: 32,
},
)?;
assert!(matches!(
read_frame(&mut socket)?,
Frame::SubscribeAck { .. }
));
Ok(socket)
}
#[test]
fn a_subject_registered_under_an_exported_prefix_after_start_also_crosses() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber_on(&pair, "orders.cancelled")?;
let bridge = exporter.bridge(&pair)?;
exporter.services.register_channel(
&liminal_server::server::connection::channel_registry::ChannelRegistration {
name: "orders.cancelled".to_owned(),
schema_bytes: None,
durable: true,
},
)?;
exporter.publish("orders.cancelled", b"{\"late\":true}")?;
delivered(&mut imported, b"{\"late\":true}")?;
assert!(bridge.observe_answers(1, OBSERVATION));
assert_eq!(bridge.counters().crossed, 1);
bridge.finish()?;
exporter.stop()?;
pair.shutdown()?;
Ok(())
}
fn delivered(socket: &mut TcpStream, expected: &[u8]) -> TestResult {
let frame = read_frame(socket)?;
assert!(matches!(frame, Frame::Deliver { envelope, .. } if envelope.payload == expected));
Ok(())
}
fn local_subscriber(
exporter: &Exporter,
subject: &str,
) -> Result<TcpStream, Box<dyn std::error::Error>> {
let mut socket = TcpStream::connect(exporter.listener.local_addr())?;
socket.set_read_timeout(Some(OBSERVATION))?;
write_frame(
&mut socket,
&Frame::Connect {
flags: 0,
min_version: liminal::protocol::ProtocolVersion::new(1, 0),
max_version: liminal::protocol::ProtocolVersion::new(1, 0),
auth_token: Vec::new(),
},
)?;
assert!(matches!(read_frame(&mut socket)?, Frame::ConnectAck { .. }));
write_frame(
&mut socket,
&Frame::Subscribe {
flags: 0,
stream_id: 1,
channel: subject.to_owned(),
accepted_schemas: Vec::new(),
max_in_flight: 32,
},
)?;
assert!(matches!(
read_frame(&mut socket)?,
Frame::SubscribeAck { .. }
));
Ok(socket)
}
#[test]
fn the_topology_receives_the_bridges_live_crossing_counters() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber(&pair)?;
let mut topology = local_subscriber(&exporter, "console.topology")?;
let bridge = exporter.bridge(&pair)?;
exporter.publish("orders.created", b"{}")?;
delivered(&mut imported, b"{}")?;
assert!(bridge.observe_answers(1, OBSERVATION));
exporter
.services
.console()
.pump(exporter.services.as_ref())?;
let Frame::Deliver { envelope, .. } = read_frame(&mut topology)? else {
return Err("topology was not delivered".into());
};
let document: serde_json::Value = serde_json::from_slice(&envelope.payload)?;
let link = document["links"]
.as_array()
.ok_or("topology links absent")?
.iter()
.find(|link| link["kind"] == "federated")
.ok_or("bridge absent from topology")?;
assert_eq!(link["crossed"], 1);
assert_eq!(link["refused"], 0);
assert_eq!(link["carrier"], "wire");
bridge.finish()?;
exporter.stop()?;
pair.shutdown()?;
Ok(())
}
fn absent(socket: &mut TcpStream) -> TestResult {
socket.set_read_timeout(Some(Duration::from_millis(100)))?;
let error = read_frame(socket)
.err()
.ok_or("unexpected second crossing")?;
let io = error
.downcast_ref::<std::io::Error>()
.ok_or("absence was not an observation bound")?;
assert!(matches!(
io.kind(),
ErrorKind::WouldBlock | ErrorKind::TimedOut
));
socket.set_read_timeout(Some(OBSERVATION))?;
Ok(())
}
#[test]
fn a_message_published_on_an_exported_prefix_arrives_at_the_importers_subscriber() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber(&pair)?;
let bridge = exporter.bridge(&pair)?;
let health = liminal_server::health::start_health_server(
"127.0.0.1:0".parse()?,
liminal_server::health::SharedReadinessState::default(),
liminal_server::health::AccessHealth {
mode: "gated",
reach: None,
decisions: None,
},
)?;
bridge.install_observations(&health.link_bridges());
for value in [b"{\"n\":1}".as_slice(), b"{\"n\":2}", b"{\"n\":3}"] {
exporter.publish("orders.created", value)?;
}
for value in [b"{\"n\":1}".as_slice(), b"{\"n\":2}", b"{\"n\":3}"] {
delivered(&mut imported, value)?;
}
assert!(bridge.observe_answers(3, OBSERVATION));
assert_eq!(bridge.counters().crossed, 3);
let observed = health_counts(health.local_addr())?;
assert_eq!(observed["links"]["orders-link"]["crossed"], 3);
assert_eq!(
observed["links"]["orders-link"]["resumed_from"]["orders.created"],
0
);
bridge.finish()?;
health.shutdown()?;
exporter.stop()?;
pair.shutdown()?;
Ok(())
}
#[test]
fn a_message_outside_the_exported_prefix_never_crosses() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber(&pair)?;
let bridge = exporter.bridge(&pair)?;
exporter.publish("orders_other", b"{}")?;
absent(&mut imported)?;
let counts = bridge.counters();
assert_eq!(counts.crossed, 0);
assert_eq!(counts.refused, 0);
assert_eq!(counts.too_large, 0);
assert!(
counts.resumed_from.contains_key("orders.created"),
"exported prefix owns no subscription"
);
assert!(!counts.resumed_from.contains_key("orders_other"));
bridge.finish()?;
exporter.stop()?;
pair.shutdown()?;
Ok(())
}
#[test]
fn a_refused_crossing_is_counted_and_not_retried() -> TestResult {
let log = CapturedLog::default();
let writer = log.clone();
let log_subscriber = tracing_subscriber::fmt()
.without_time()
.with_ansi(false)
.with_writer(move || writer.clone())
.finish();
let subscriber_guard = tracing::subscriber::set_default(log_subscriber);
let pair = Federation::with_first_import_refused()?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber(&pair)?;
let bridge = exporter.bridge(&pair)?;
exporter.publish("orders.secret", b"{}")?;
assert!(
bridge.observe_answers(1, OBSERVATION),
"refused crossing was never answered"
);
assert_eq!(bridge.counters().refused, 1);
exporter.publish("orders.created", b"{\"after\":true}")?;
delivered(&mut imported, b"{\"after\":true}")?;
assert!(bridge.observe_answers(2, OBSERVATION));
assert_eq!(bridge.counters().refused, 1);
assert_eq!(bridge.counters().crossed, 1);
absent(&mut imported)?;
bridge.finish()?;
drop(subscriber_guard);
let text = String::from_utf8(log.0.lock().map_err(|_| "log lock poisoned")?.clone())?;
assert_eq!(
text.matches("link crossing refused; subscription advances without retry")
.count(),
1
);
assert!(text.contains("crossing refused at import"));
exporter.stop()?;
pair.shutdown()?;
Ok(())
}
#[test]
fn a_message_too_large_for_one_noise_frame_is_refused_by_name() -> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber(&pair)?;
let bridge = exporter.bridge(&pair)?;
let payload = format!(
"{{\"data\":\"{}\"}}",
"x".repeat(liminal_server::link_handshake::NOISE_MAX_MESSAGE)
);
exporter.publish("orders.created", payload.as_bytes())?;
assert!(
bridge.observe_answers(1, OBSERVATION),
"too_large was not counted"
);
assert_eq!(bridge.counters().too_large, 1);
assert_eq!(bridge.counters().last_refusal.as_deref(), Some("too_large"));
exporter.publish("orders.created", b"{}")?;
delivered(&mut imported, b"{}")?;
assert!(bridge.observe_answers(2, OBSERVATION));
assert_eq!(bridge.counters().crossed, 1);
absent(&mut imported)?;
bridge.finish()?;
exporter.stop()?;
pair.shutdown()?;
Ok(())
}
#[test]
fn after_a_restart_the_bridge_resumes_at_its_subscriptions_position_and_sends_nothing_twice()
-> TestResult {
let pair = Federation::start(LiveOpen::Outright)?;
let root = tempfile::tempdir()?;
let exporter = Exporter::boot(root.path())?;
let mut imported = subscriber(&pair)?;
let bridge = exporter.bridge(&pair)?;
exporter.publish("orders.created", b"{\"before\":true}")?;
delivered(&mut imported, b"{\"before\":true}")?;
assert!(bridge.observe_answers(1, OBSERVATION));
bridge.finish()?;
exporter.publish("orders.created", b"{\"offline\":true}")?;
exporter.stop()?;
let restarted = Exporter::boot(root.path())?;
let bridge = restarted.bridge(&pair)?;
assert_eq!(
bridge.counters().resumed_from.get("orders.created"),
Some(&1)
);
delivered(&mut imported, b"{\"offline\":true}")?;
restarted.publish("orders.created", b"{\"after\":true}")?;
delivered(&mut imported, b"{\"after\":true}")?;
assert!(bridge.observe_answers(2, OBSERVATION));
assert_eq!(bridge.counters().crossed, 2);
absent(&mut imported)?;
bridge.finish()?;
restarted.stop()?;
pair.shutdown()?;
Ok(())
}