use std::error::Error;
use std::time::Duration;
use crate::ServerError;
use crate::config::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::ConnectionSupervisor;
use crate::server::listener::ServerListener;
use crate::server::self_streams::tests::R3Fixture;
use crate::test_log::CapturedLog;
use super::{ShutdownHandle, run_shutdown_sequence_before_flush};
type TestResult = Result<(), Box<dyn Error>>;
struct Fixture {
listener: ServerListener,
supervisor: ConnectionSupervisor,
source: R3Fixture,
stopped: bool,
}
impl Fixture {
fn new() -> Result<Self, Box<dyn Error>> {
let source = R3Fixture::new()?;
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()?,
drain_timeout_ms: 0,
channels: Vec::new(),
routing_rules: Vec::new(),
persistence_path: None,
cluster: None,
auth: None,
services: ServicesConfig::default(),
limits: LimitsConfig::default(),
participant: None,
websocket: None,
};
let supervisor = ConnectionSupervisor::with_fatal_shutdown(
source.services.clone(),
None,
None,
config.limits,
ShutdownHandle::new(),
)?;
let listener = ServerListener::bind(&config, supervisor.clone())?;
Ok(Self {
listener,
supervisor,
source,
stopped: false,
})
}
fn stop(
&mut self,
finalize: impl FnOnce() -> Result<(), ServerError>,
) -> Result<(), ServerError> {
self.stopped = true;
run_shutdown_sequence_before_flush(
&mut self.listener,
None,
&self.supervisor,
Duration::ZERO,
finalize,
)
}
}
impl Drop for Fixture {
fn drop(&mut self) {
if !self.stopped {
if let Err(error) = self.stop(|| Ok(())) {
eprintln!("shutdown fixture stop failed: {error}");
assert!(
std::thread::panicking(),
"shutdown fixture stop failed: {error}"
);
}
}
}
}
fn shutdown_completion(log: &CapturedLog) -> String {
let contents = log.contents();
let completions: Vec<_> = contents
.lines()
.filter(|line| line.contains("shutdown sequence complete"))
.collect();
assert_eq!(
completions.len(),
1,
"expected one shutdown completion verdict, actual log: {contents}"
);
completions[0].to_owned()
}
#[test]
fn unresolved_self_publish_finalize_keeps_its_error_and_reports_unclean() -> TestResult {
let mut fixture = Fixture::new()?;
let log = CapturedLog::default();
let mut finalized = 0;
let result = log.capture(|| {
fixture.stop(|| {
finalized += 1;
Err(ServerError::PublishUnresolved {
channel: "liminal.cost".to_owned(),
idempotency_key: None,
sequence: Some(2396),
message: "self append had started when its wait ended".to_owned(),
})
})
});
assert_eq!(finalized, 1, "expected finalize once, actual {finalized}");
assert!(
matches!(&result, Err(ServerError::PublishUnresolved {
channel, idempotency_key: None, sequence: Some(2396), message
}) if channel == "liminal.cost" && message == "self append had started when its wait ended"),
"expected the unresolved liminal.cost error and its first evidence, actual {result:?}"
);
let completion = shutdown_completion(&log);
assert!(
completion.contains("clean=false"),
"expected clean=false when self publication finalization fails, actual {completion}"
);
assert!(
!completion.contains("clean=true"),
"expected no clean=true verdict for failed finalization, actual {completion}"
);
Ok(())
}
#[test]
fn self_publish_finalize_precedes_the_shared_durable_flush() -> TestResult {
let mut fixture = Fixture::new()?;
let counts = fixture.source.services.store_counts();
let before = counts.snapshot();
let mut flushes_at_finalize = None;
let result = fixture.stop(|| {
flushes_at_finalize = Some(counts.snapshot().since(&before).flushes);
Ok(())
});
assert!(
result.is_ok(),
"expected a successful shutdown, actual {result:?}"
);
assert_eq!(
flushes_at_finalize,
Some(0),
"expected finalize before any shared durable flush, actual observed flushes {flushes_at_finalize:?}"
);
let flushed = counts.snapshot().since(&before).flushes;
assert_eq!(
flushed, 1,
"expected one later shared durable flush, actual {flushed}"
);
Ok(())
}
#[test]
fn successful_self_publish_finalize_reports_clean_after_one_shared_flush() -> TestResult {
let mut fixture = Fixture::new()?;
let counts = fixture.source.services.store_counts();
let before = counts.snapshot();
let log = CapturedLog::default();
let mut finalized = 0;
let result = log.capture(|| {
fixture.stop(|| {
finalized += 1;
Ok(())
})
});
assert!(
result.is_ok(),
"expected a successful shutdown, actual {result:?}"
);
assert_eq!(finalized, 1, "expected finalize once, actual {finalized}");
let completion = shutdown_completion(&log);
assert!(
completion.contains("clean=true"),
"expected clean=true after successful finalization, actual {completion}"
);
assert!(
!completion.contains("clean=false"),
"expected no unclean verdict for successful finalization, actual {completion}"
);
let flushed = counts.snapshot().since(&before).flushes;
assert_eq!(
flushed, 1,
"expected one shared durable flush, actual {flushed}"
);
Ok(())
}