use std::sync::Arc;
use std::time::Duration;
use liminal::durability::DurableStore;
use super::tests_memory_store::MemoryStore;
use crate::config::{LimitsConfig, ServerConfig, ServicesConfig};
use crate::server::connection::ConnectionSupervisor;
use crate::server::connection::services::LiminalConnectionServices;
use crate::test_log::CapturedLog;
type TestResult = Result<(), Box<dyn std::error::Error>>;
fn start(store: Arc<dyn DurableStore>) -> Result<ConnectionSupervisor, 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()?,
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: Some(crate::server::participant::test_participant_config()),
websocket: None,
};
let services = Arc::new(LiminalConnectionServices::from_config_with_store(
&config, store,
)?);
let supervisor = ConnectionSupervisor::with_services(services)?;
supervisor.current_server_incarnation()?;
Ok(supervisor)
}
fn assert_start(log: &CapturedLog, previous: &str, level: &str) {
let text = log.contents();
let lines = text
.lines()
.filter(|line| line.contains("restart repair started"))
.collect::<Vec<_>>();
assert_eq!(lines.len(), 1, "{text}");
assert!(lines[0].contains(level), "{text}");
assert!(
lines[0].contains(&format!("previous_stop=\"{previous}\"")),
"{text}"
);
assert!(lines[0].contains("server_incarnation="), "{text}");
assert!(lines[0].contains("conversations=0"), "{text}");
assert!(!text.contains("unclean-restart"), "{text}");
}
#[test]
fn a_start_after_a_clean_stop_logs_previous_stop_clean() -> TestResult {
let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
let first = start(Arc::clone(&store))?;
crate::server::shutdown::sequence::run(&first, Duration::ZERO, || Ok(()), || Ok(()))?;
drop(first);
let log = CapturedLog::default();
let second = log.capture(|| start(store))?;
second.shutdown();
assert_start(&log, "clean", "INFO");
assert!(!log.contents().contains("unclean"));
Ok(())
}
#[test]
fn a_start_after_a_kill_logs_previous_stop_unclean() -> TestResult {
let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
let first = start(Arc::clone(&store))?;
first.scheduler().shutdown();
drop(first);
assert!(
super::capacity_snapshot::read_capacity_snapshot(store.as_ref())?.is_none(),
"a kill published no clean stop"
);
let log = CapturedLog::default();
let second = log.capture(|| start(store))?;
second.shutdown();
assert_start(&log, "unclean", "WARN");
Ok(())
}
#[test]
fn a_first_start_logs_previous_stop_unknown() -> TestResult {
let log = CapturedLog::default();
let server = log.capture(|| start(Arc::new(MemoryStore::default())))?;
server.shutdown();
assert_start(&log, "unknown", "INFO");
Ok(())
}
#[test]
fn an_old_capacity_publication_migrates_on_the_next_clean_stop() -> TestResult {
use super::capacity_snapshot::CAPACITY_KEY;
use super::capacity_snapshot::{CapacitySnapshot, write_capacity_snapshot};
use super::registry::REGISTRY_STREAM_KEY;
use liminal::durability::bridge::block_on;
let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
write_capacity_snapshot(
store.as_ref(),
None,
&CapacitySnapshot::LegacyCleanStop(std::collections::BTreeMap::new()),
)?;
let server = start(Arc::clone(&store))?;
let incarnation = server
.current_server_incarnation()?
.ok_or("no incarnation")?;
crate::server::shutdown::sequence::run(&server, Duration::ZERO, || Ok(()), || Ok(()))?;
let bytes = block_on(store.get_routed(REGISTRY_STREAM_KEY, CAPACITY_KEY))??
.ok_or("no clean-stop publication")?;
assert_eq!(
bytes.get(6..8),
Some([0, 1].as_slice()),
"migration writes codec 1.1"
);
assert_eq!(
bytes.get(24..32),
Some(incarnation.to_be_bytes().as_slice()),
"publication names its incarnation"
);
Ok(())
}
#[test]
fn an_older_clean_stop_cannot_name_a_killed_incarnation_clean() -> TestResult {
let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
let first = start(Arc::clone(&store))?;
crate::server::shutdown::sequence::run(&first, Duration::ZERO, || Ok(()), || Ok(()))?;
drop(first);
let prior_publication = super::capacity_snapshot::read_capacity_snapshot(store.as_ref())?;
let second = start(Arc::clone(&store))?;
second.scheduler().shutdown();
drop(second);
assert_eq!(
super::capacity_snapshot::read_capacity_snapshot(store.as_ref())?,
prior_publication,
"a kill cannot replace the older clean-stop publication"
);
let log = CapturedLog::default();
let third = log.capture(|| start(store))?;
third.shutdown();
assert_start(&log, "unclean", "WARN");
Ok(())
}
#[test]
fn a_codec_1_0_clean_stop_reads_previous_stop_unknown() -> TestResult {
use super::capacity_snapshot::{CapacitySnapshot, write_capacity_snapshot};
let store: Arc<dyn DurableStore> = Arc::new(MemoryStore::default());
let previous = start(Arc::clone(&store))?;
previous.scheduler().shutdown();
drop(previous);
let publication = CapacitySnapshot::LegacyCleanStop(std::collections::BTreeMap::new());
let bytes = publication.encode()?;
assert_eq!(bytes.get(6..8), Some([0, 0].as_slice()));
write_capacity_snapshot(store.as_ref(), None, &publication)?;
let log = CapturedLog::default();
let server = log.capture(|| start(store))?;
assert_eq!(server.current_server_incarnation()?, Some(2));
server.shutdown();
assert_start(&log, "unknown", "INFO");
assert_eq!(
log.contents()
.lines()
.filter(|line| line.contains("WARN"))
.count(),
0,
"{}",
log.contents()
);
Ok(())
}
#[test]
fn a_clean_stop_naming_an_older_incarnation_stays_unclean() -> TestResult {
use super::capacity_snapshot::{CapacitySnapshot, write_capacity_snapshot};
use super::previous_stop::PreviousStop;
let store = MemoryStore::default();
write_capacity_snapshot(
&store,
None,
&CapacitySnapshot::CleanStop {
server_incarnation: 4,
shares: std::collections::BTreeMap::new(),
},
)?;
let log = CapturedLog::default();
log.capture(|| -> TestResult {
let previous = PreviousStop::read(&store, 6)?;
assert_eq!(previous, PreviousStop::Unclean);
previous.started(6, 0);
Ok(())
})?;
assert_start(&log, "unclean", "WARN");
Ok(())
}
#[test]
fn a_refused_clean_stop_publication_stays_unclean() -> TestResult {
use super::capacity_snapshot::{CAPACITY_KEY, CapacitySnapshot};
use super::previous_stop::PreviousStop;
use super::registry::REGISTRY_STREAM_KEY;
use liminal::durability::{RoutedBatchItem, bridge::block_on};
let store = MemoryStore::default();
let mut bytes =
CapacitySnapshot::LegacyCleanStop(std::collections::BTreeMap::new()).encode()?;
*bytes
.last_mut()
.ok_or("capacity publication has no digest")? ^= 1;
block_on(store.apply_routed_batch(
REGISTRY_STREAM_KEY,
vec![RoutedBatchItem {
key: CAPACITY_KEY.to_vec(),
expected: None,
value: bytes,
ttl: None,
}],
))??;
let log = CapturedLog::default();
log.capture(|| -> TestResult {
let previous = PreviousStop::read(&store, 2)?;
assert_eq!(previous, PreviousStop::Unclean);
previous.started(2, 0);
Ok(())
})?;
assert_start(&log, "unclean", "WARN");
let text = log.contents();
assert!(text.contains("capacity snapshot: digest differs"), "{text}");
assert!(
text.contains("previous stop's capacity publication was refused by name"),
"{text}"
);
Ok(())
}