use super::cost::{CostWindow, SelfWork};
use super::events::ServerEvent;
use super::publisher::CostSink;
use super::sink::SelfStreams;
use crate::config::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::link::LinkLife;
use crate::link_bridge::LinkBridge;
use crate::link_carrier::{WireCarrier, read_frame, write_frame};
use crate::link_dial::DialedLink;
use crate::link_handshake::{LinkHandshake, LocalMachine, NOISE_OVERHEAD};
use crate::server::connection::{ConnectionSupervisor, LiminalConnectionServices};
use crate::server::listener::ServerListener;
use crate::server::shutdown::{ShutdownHandle, run_shutdown_sequence_before_flush};
use crate::test_log::CapturedLog;
use liminal::durability::StoreCounts;
use liminal::protocol::Frame;
use lys_core::keys::Ed25519Identity;
use std::net::{Shutdown, TcpListener, TcpStream};
use std::path::Path;
use std::sync::{Arc, mpsc};
use std::thread::JoinHandle;
use std::time::Duration;
use zeroize::Zeroizing;
type TestError = Box<dyn std::error::Error + Send + Sync>;
type TestResult<T = ()> = Result<T, TestError>;
const OBSERVATION: Duration = Duration::from_secs(5);
struct Server {
listener: ServerListener,
supervisor: ConnectionSupervisor,
services: Arc<LiminalConnectionServices>,
sink: Arc<SelfStreams>,
stopped: bool,
}
impl Server {
fn boot(root: &Path) -> TestResult<Self> {
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: Vec::new(),
routing_rules: Vec::new(),
persistence_path: Some(root.to_owned()),
cluster: None,
auth: None,
drain_timeout_ms: 0,
services: ServicesConfig::default(),
limits: LimitsConfig::default(),
participant: Some(crate::server::participant::test_participant_config()),
websocket: None,
};
let services = Arc::new(LiminalConnectionServices::from_config(&config)?);
let shutdown = ShutdownHandle::new();
let supervisor = ConnectionSupervisor::with_fatal_shutdown(
services.clone(),
None,
None,
config.limits,
shutdown.clone(),
)?;
let sink =
crate::server::runtime::bind_self_streams(&services, &supervisor, &config, &shutdown)?
.ok_or("self streams absent")?;
let listener = ServerListener::bind(&config, supervisor.clone())?;
Ok(Self {
listener,
supervisor,
services,
sink,
stopped: false,
})
}
fn cost(&self) -> TestResult {
let window = CostWindow::new(StoreCounts::default(), SelfWork::default());
let record = window
.prepare(self.services.store_counts().snapshot(), SelfWork::default())
.ok_or("fixture has no store work")?;
CostSink::publish(&mut Arc::clone(&self.sink), &record)?;
Ok(())
}
fn event(&self, connection: u64) -> TestResult {
self.sink
.enqueue_event_for_test(ServerEvent::ConnectionOpened { connection })?;
self.sink.drain_all()?;
Ok(())
}
fn stop(&mut self) -> TestResult {
self.stopped = true;
run_shutdown_sequence_before_flush(
&mut self.listener,
None,
&self.supervisor,
Duration::ZERO,
|| self.sink.drain_all(),
)?;
Ok(())
}
}
impl Drop for Server {
fn drop(&mut self) {
if !self.stopped {
if let Err(error) = self.stop() {
eprintln!("link-refusal fixture stop failed: {error}");
assert!(
std::thread::panicking(),
"link-refusal fixture stop failed: {error}"
);
}
}
}
}
struct RefusingPeer {
first: mpsc::Receiver<()>,
release: Option<mpsc::Sender<()>>,
interrupt: TcpStream,
worker: Option<JoinHandle<TestResult<Vec<String>>>>,
}
impl RefusingPeer {
fn open() -> TestResult<(DialedLink, Self)> {
let listener = TcpListener::bind("127.0.0.1:0")?;
let socket = TcpStream::connect(listener.local_addr()?)?;
let (mut peer, _) = listener.accept()?;
let interrupt = peer.try_clone()?;
let local = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([1; 32])));
let remote = LocalMachine::new(Ed25519Identity::from_seed(&Zeroizing::new([2; 32])));
let mut sending = LinkHandshake::initiator(&local, remote.machine())?;
let mut receiving = LinkHandshake::responder(&remote, local.machine())?;
let mut bytes = [0; NOISE_OVERHEAD];
let length = sending.write(&mut bytes)?;
receiving.read(&bytes[..length])?;
let length = receiving.write(&mut bytes)?;
sending.read(&bytes[..length])?;
let mut carrier = WireCarrier::new(socket, 1)?;
carrier.install_life(LinkLife::new(remote.machine()))?;
let link = DialedLink {
link_id: "self-link".into(),
exports: vec!["liminal".into()],
transport: sending.finish()?,
carrier,
};
let (entered, first) = mpsc::sync_channel(1);
let (release, released) = mpsc::channel();
let worker = std::thread::spawn(move || {
let mut subjects = Vec::new();
while let Some(frame) = read_frame(&mut peer)? {
let Frame::LinkMessage {
stream_id, subject, ..
} = frame
else {
return Err("peer received a non-crossing frame".into());
};
if subjects.is_empty() {
entered.send(())?;
released.recv()?;
}
subjects.push(subject);
write_frame(
&mut peer,
&Frame::PublishError {
flags: 0,
stream_id,
reason_code: 0x0203,
message: Some("crossing refused at import: link_import_not_granted".into()),
},
)?;
}
Ok(subjects)
});
Ok((
link,
Self {
first,
release: Some(release),
interrupt,
worker: Some(worker),
},
))
}
fn release(&mut self) -> TestResult {
if let Some(release) = self.release.take() {
release.send(())?;
}
Ok(())
}
fn finish(mut self) -> TestResult<Vec<String>> {
self.release()?;
self.worker
.take()
.ok_or("peer worker absent")?
.join()
.map_err(|_| "peer worker panicked")?
}
}
impl Drop for RefusingPeer {
fn drop(&mut self) {
if let Some(release) = self.release.take() {
if let Err(error) = release.send(()) {
eprintln!("peer release failed: {error}");
}
}
if let Err(error) = self.interrupt.shutdown(Shutdown::Both) {
if error.kind() != std::io::ErrorKind::NotConnected {
eprintln!("peer interrupt failed: {error}");
}
}
if let Some(worker) = self.worker.take() {
match worker.join() {
Ok(Ok(_)) => {}
Ok(Err(error)) => eprintln!("peer fixture failed: {error}"),
Err(_) => eprintln!("peer fixture panicked"),
}
}
}
}
#[test]
fn a_refused_self_stream_crossing_cannot_block_append_clean_shutdown_or_restart() -> TestResult {
let root = tempfile::tempdir()?;
let log = CapturedLog::default();
log.capture(|| {
let mut server = Server::boot(root.path())?;
let (link, mut peer) = RefusingPeer::open()?;
let bridge = LinkBridge::start(link, Arc::clone(&server.services))?;
server.cost()?;
peer.first.recv_timeout(OBSERVATION)?;
let appended = (|| {
let before = server.services.self_stream_head("liminal.events")?;
for start in [0, 600] {
server.sink.lifecycle_batch(
1,
(start..start + 600).map(|generation| (generation, "opened")),
)?;
server.sink.drain_all()?;
}
assert!(server.services.self_stream_head("liminal.events")? >= before + 1200);
server.event(1200)
})();
peer.release()?;
assert!(
appended.is_ok(),
"self-stream append failed behind a crossing reader: {appended:?}"
);
let observed = bridge.observe_answers(2, OBSERVATION);
let counts = bridge.counters();
bridge.finish()?;
peer.finish()?;
assert!(
appended.is_ok(),
"self-stream append failed behind a crossing reader: {appended:?}"
);
assert!(observed, "refused crossing did not settle: {counts:?}");
assert!(counts.refused >= 2);
let events = server.services.self_stream_head("liminal.events")?;
let cost = server.services.self_stream_head("liminal.cost")?;
server.stop()?;
assert!(
log.contents()
.lines()
.any(|line| line.contains("shutdown sequence complete")
&& line.contains("clean=true")),
"{}",
log.contents()
);
drop(server);
let mut restarted = Server::boot(root.path())?;
assert!(restarted.services.self_stream_head("liminal.events")? >= events);
assert!(restarted.services.self_stream_head("liminal.cost")? >= cost);
assert!(TcpStream::connect(restarted.listener.local_addr()).is_ok());
restarted.event(1201)?;
restarted.stop()?;
Ok(())
})
}
#[test]
fn self_stream_crossing_checkpoints_do_not_publish_more_self_stream_work() -> TestResult {
let root = tempfile::tempdir()?;
let mut server = Server::boot(root.path())?;
let (link, mut peer) = RefusingPeer::open()?;
let bridge = LinkBridge::start(link, Arc::clone(&server.services))?;
server.cost()?;
peer.first.recv_timeout(OBSERVATION)?;
server.event(1)?;
let head = server.services.self_stream_head("liminal.events")?;
peer.release()?;
assert!(bridge.observe_answers(2, OBSERVATION));
let finished = bridge.finish();
let peer_finished = peer.finish();
server.sink.drain_all()?;
assert_eq!(
server.services.self_stream_head("liminal.events")?,
head,
"bridge checkpoint flushes must not publish new events for the bridge to cross"
);
finished?;
peer_finished?;
server.stop()?;
Ok(())
}
#[test]
fn repeated_self_stream_crossing_refusals_are_counted_and_logged_once_per_subject() -> TestResult {
let root = tempfile::tempdir()?;
let log = CapturedLog::default();
log.capture(|| {
let mut server = Server::boot(root.path())?;
let (link, mut peer) = RefusingPeer::open()?;
let bridge = LinkBridge::start(link, Arc::clone(&server.services))?;
server.cost()?;
peer.first.recv_timeout(OBSERVATION)?;
server.event(1)?;
server.cost()?;
server.event(2)?;
peer.release()?;
assert!(bridge.observe_answers(4, OBSERVATION));
let counts = bridge.counters();
let finished = bridge.finish();
let peer_finished = peer.finish();
assert_eq!(counts.refused, 4);
let crossed = peer_finished?;
assert_eq!(crossed.len(), 4, "checkpoint work fed back into the bridge");
for subject in ["liminal.cost", "liminal.events"] {
let named: Vec<_> = log
.contents()
.lines()
.filter(|line| {
line.contains("link crossing refused;")
&& line.contains(&format!("subject={subject}"))
})
.map(str::to_owned)
.collect();
assert_eq!(
named.len(),
1,
"each refused subject must be logged once: {named:?}"
);
assert!(named[0].contains("link_import_not_granted"));
}
finished?;
server.stop()?;
Ok(())
})
}