use std::error::Error;
use std::sync::atomic::Ordering;
use std::thread;
use std::time::Duration;
use super::{ShutdownHandle, wait_after_force_close};
use crate::server::connection::ConnectionSupervisor;
#[test]
fn shutdown_handle_initiates_once() {
let handle = ShutdownHandle::new();
assert!(!handle.is_initiated());
assert!(handle.initiate());
assert!(handle.is_initiated());
assert!(!handle.initiate());
}
const SIGNAL_LOCK_POISONED: &str =
"the test signal lock was poisoned: another test thread panicked holding it";
fn await_parked(handle: &ShutdownHandle) -> Result<(), Box<dyn Error>> {
let state = &handle.inner;
let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
loop {
if state.returned.load(Ordering::SeqCst) > 0 {
return Err("the waiter returned from wait() without parking".into());
}
if state.parked.load(Ordering::SeqCst) > 0 {
drop(guard);
return Ok(());
}
guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
}
}
fn await_returned(handle: &ShutdownHandle, parked_before: usize) -> Result<(), Box<dyn Error>> {
let state = &handle.inner;
if state.notified.load(Ordering::SeqCst) == 0 {
return Err("initiate() never notified the parked waiter".into());
}
let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
while state.returned.load(Ordering::SeqCst) == 0 {
if state.parked.load(Ordering::SeqCst) > parked_before {
return Err("the waiter woke and parked again: it did not see the shutdown".into());
}
guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
}
drop(guard);
Ok(())
}
#[test]
fn shutdown_handle_wait_unblocks_on_initiate() -> Result<(), Box<dyn std::error::Error>> {
let handle = ShutdownHandle::new();
let waiter = handle.clone();
let worker = thread::spawn(move || {
waiter.wait();
waiter.is_initiated()
});
await_parked(&handle)?;
let parked = handle.inner.parked.load(Ordering::SeqCst);
assert!(handle.initiate());
await_returned(&handle, parked)?;
let observed = worker.join().map_err(|_| "wait worker panicked")?;
assert!(observed);
Ok(())
}
fn wait_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
&state.wait_lock
}
fn signal_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
&state.signal_lock
}
fn poison(handle: &ShutdownHandle, lock: fn(&super::ShutdownState) -> &std::sync::Mutex<()>) {
let poisoner = handle.clone();
let outcome = thread::spawn(move || {
let _guard = lock(&poisoner.inner).lock();
std::panic::resume_unwind(Box::new("poison a shutdown lock"));
})
.join();
assert!(outcome.is_err(), "the poisoner thread must have panicked");
}
fn poison_wait_lock(handle: &ShutdownHandle) {
poison(handle, wait_lock_of);
}
#[test]
fn a_poisoned_test_signal_lock_is_refused_by_name() {
let handle = ShutdownHandle::new();
poison(&handle, signal_lock_of);
assert!(handle.initiate());
let refused =
await_returned(&handle, 0).map_or_else(|error| error.to_string(), |()| String::new());
assert_eq!(refused, SIGNAL_LOCK_POISONED);
let refused = await_parked(&handle).map_or_else(|error| error.to_string(), |()| String::new());
assert_eq!(refused, SIGNAL_LOCK_POISONED);
}
#[test]
fn wait_on_poisoned_lock_returns_and_logs_error() {
let handle = ShutdownHandle::new();
poison_wait_lock(&handle);
let log = crate::test_log::CapturedLog::default();
log.capture(|| handle.wait());
let output = log.contents();
assert!(
output.contains("ERROR"),
"a poisoned wait lock must be logged at error level, got: {output}"
);
}
#[test]
fn notify_on_poisoned_lock_still_wakes_parked_waiter() -> Result<(), Box<dyn std::error::Error>> {
let handle = ShutdownHandle::new();
let waiter = handle.clone();
let worker = thread::spawn(move || waiter.wait());
await_parked(&handle)?;
poison_wait_lock(&handle);
let parked = handle.inner.parked.load(Ordering::SeqCst);
assert!(handle.initiate());
await_returned(&handle, parked)?;
worker.join().map_err(|_| "wait worker panicked")?;
Ok(())
}
#[test]
fn the_close_settle_returns_immediately_when_no_connections_are_active()
-> Result<(), Box<dyn std::error::Error>> {
let supervisor = ConnectionSupervisor::new()?;
let started = std::time::Instant::now();
wait_after_force_close(&supervisor);
let elapsed = started.elapsed();
assert!(
elapsed < Duration::from_millis(50),
"the close settle took {elapsed:?} with no connections tracked"
);
supervisor.shutdown();
Ok(())
}
#[test]
fn shutdown_source_has_no_drain_and_no_reap_count_sleep_loop() {
let implementation = shutdown_implementation();
for forbidden in [
"DRAIN_PROGRESS_INTERVAL",
"FORCE_CLOSE_SETTLE_TIMEOUT",
"FORCE_CLOSE_POLL_INTERVAL",
"reap_crashed_connections",
"fn drain_connections",
] {
assert!(
!implementation.contains(forbidden),
"retired poll/reap/drain token `{forbidden}` must not appear in the shutdown implementation"
);
}
}
#[test]
fn the_configured_drain_timeout_is_never_turned_into_a_deadline() {
let implementation = shutdown_implementation();
assert!(
!implementation.contains("+ drain_timeout"),
"drain_timeout must never be added to an Instant to form a shutdown deadline"
);
assert!(
implementation.contains("ignored_drain_timeout = ?drain_timeout"),
"drain_timeout must still be named in the shutdown log line that reports it ignored"
);
}
fn shutdown_implementation() -> String {
[
include_str!("shutdown.rs"),
include_str!("shutdown_sequence.rs"),
]
.map(|source| source.split("mod tests").next().unwrap_or(source))
.join("\n")
}