Skip to main content

liminal_server/server/
shutdown.rs

1use std::fmt;
2use std::sync::atomic::{AtomicBool, Ordering};
3use std::sync::{Arc, Condvar, Mutex};
4use std::thread::{self, JoinHandle};
5use std::time::{Duration, Instant};
6
7use signal_hook::consts::signal::{SIGINT, SIGTERM};
8use signal_hook::iterator::{Handle as SignalIteratorHandle, Signals};
9
10use crate::ServerError;
11use crate::server::connection::{ConnectionSupervisor, WebSocketListener};
12use crate::server::listener::ServerListener;
13
14/// The ONE bounded park left in the shutdown path, and it is a HANG STOP — not
15/// a grace period, not a drain, and never a delay anything waits out.
16///
17/// 0.14.3 retired the graceful drain because its stated purpose did not hold.
18/// A drain exists to let in-flight requests finish; in this server every write
19/// is durable and flushed BEFORE it is acknowledged (`OperationLog::append`
20/// flushes the store inside the append), so at the instant shutdown begins
21/// there is nothing in flight to finish. What the drain actually did on an
22/// estate whose connections are long-lived idle seats — seats that never hang
23/// up by themselves — was wait out its whole configured budget (5 s on Tom's
24/// estate, measured 2026-09-14) and then force-close anyway. The close was
25/// always the real mechanism; the wait in front of it bought nothing.
26///
27/// Both remaining waits park on an EVENT (the supervisor's delivery-quiescence
28/// signal and its TOLD drain-completion notification, W4 leg 3 §4.3) and return
29/// the instant that event arrives — with every connection idle they return in
30/// microseconds. This bound exists solely so that ONE wedged connection process
31/// cannot hold a restart open forever; it is the stop on a hang, which is why
32/// it is a number at all and why it is short.
33pub(crate) const WEDGED_CONNECTION_STOP: Duration = Duration::from_millis(500);
34
35/// Idempotent shutdown activation handle shared by the runtime and signal thread.
36#[derive(Clone)]
37pub struct ShutdownHandle {
38    inner: Arc<ShutdownState>,
39}
40
41impl ShutdownHandle {
42    /// Creates a new inactive shutdown handle.
43    #[must_use]
44    pub fn new() -> Self {
45        Self {
46            inner: Arc::new(ShutdownState::new()),
47        }
48    }
49
50    /// Initiates shutdown exactly once.
51    ///
52    /// Returns `true` for the first caller that transitions the handle to active,
53    /// and `false` for subsequent calls.
54    pub fn initiate(&self) -> bool {
55        if self.inner.initiated.swap(true, Ordering::SeqCst) {
56            tracing::debug!("shutdown request ignored because shutdown is already active");
57            return false;
58        }
59
60        tracing::info!("shutdown requested");
61        self.inner.notify();
62        true
63    }
64
65    /// Blocks until shutdown is initiated.
66    pub fn wait(&self) {
67        // A test's count of every return from here, taken as this guard
68        // drops, so no test waits on a waiter that left without parking.
69        #[cfg(test)]
70        let _returned = Returned(&self.inner);
71        if self.is_initiated() {
72            return;
73        }
74        let Ok(mut guard) = self.inner.wait_lock.lock() else {
75            // Behaviour unchanged (return as if shutdown was requested), but a
76            // poisoned lock must not masquerade silently as a completed wait.
77            tracing::error!(
78                "shutdown wait lock poisoned; returning from wait as if shutdown was requested"
79            );
80            return;
81        };
82        while !self.is_initiated() {
83            // A test's signal that this waiter is about to release the lock by
84            // parking, so no test waits on a clock for it.
85            #[cfg(test)]
86            self.inner.signal_parked();
87            let Ok(next_guard) = self.inner.waiter.wait(guard) else {
88                tracing::error!(
89                    "shutdown wait condvar reacquired a poisoned lock; returning from wait as if shutdown was requested"
90                );
91                return;
92            };
93            guard = next_guard;
94        }
95    }
96
97    /// Returns whether shutdown has been initiated.
98    #[must_use]
99    pub fn is_initiated(&self) -> bool {
100        self.inner.initiated.load(Ordering::SeqCst)
101    }
102}
103
104impl Default for ShutdownHandle {
105    fn default() -> Self {
106        Self::new()
107    }
108}
109
110impl fmt::Debug for ShutdownHandle {
111    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
112        formatter
113            .debug_struct("ShutdownHandle")
114            .field("initiated", &self.is_initiated())
115            .finish()
116    }
117}
118
119#[derive(Debug)]
120struct ShutdownState {
121    initiated: AtomicBool,
122    wait_lock: Mutex<()>,
123    waiter: Condvar,
124    /// Times a waiter parked on `waiter`, counted under `wait_lock`.
125    #[cfg(test)]
126    parked: std::sync::atomic::AtomicUsize,
127    /// Times `wait` returned, counted under `wait_lock`.
128    #[cfg(test)]
129    returned: std::sync::atomic::AtomicUsize,
130    /// Times `notify` sent its wake to `waiter`.
131    #[cfg(test)]
132    notified: std::sync::atomic::AtomicUsize,
133    /// The test-only condvar `parked` and `returned` are signalled on, kept
134    /// apart from `waiter` so a test build never wakes a production waiter.
135    #[cfg(test)]
136    signal: Condvar,
137    /// The test-only lock `signal` pairs with: `parked` and `returned` rise and
138    /// are signalled under it, and the test helpers wait under it. It is kept
139    /// apart from `wait_lock`, the production lock under test, so a test that
140    /// poisons `wait_lock` on purpose still reads every count, and nothing
141    /// recovers a poisoned guard. Lock order, the only nesting: `wait_lock`,
142    /// then `signal_lock` (in `wait`, around `signal_parked`). Nothing takes
143    /// `signal_lock` and then `wait_lock`.
144    #[cfg(test)]
145    signal_lock: Mutex<()>,
146}
147
148/// Test-only: counts one return from [`ShutdownHandle::wait`] as it drops,
149/// under `signal_lock`, and signals it. It drops after `wait`'s own guard, so
150/// `wait_lock` is not held here.
151#[cfg(test)]
152struct Returned<'a>(&'a ShutdownState);
153
154#[cfg(test)]
155impl Drop for Returned<'_> {
156    fn drop(&mut self) {
157        self.0.count_and_signal(&self.0.returned);
158    }
159}
160
161impl ShutdownState {
162    const fn new() -> Self {
163        Self {
164            initiated: AtomicBool::new(false),
165            wait_lock: Mutex::new(()),
166            waiter: Condvar::new(),
167            #[cfg(test)]
168            parked: std::sync::atomic::AtomicUsize::new(0),
169            #[cfg(test)]
170            returned: std::sync::atomic::AtomicUsize::new(0),
171            #[cfg(test)]
172            notified: std::sync::atomic::AtomicUsize::new(0),
173            #[cfg(test)]
174            signal: Condvar::new(),
175            #[cfg(test)]
176            signal_lock: Mutex::new(()),
177        }
178    }
179
180    /// Test-only: counts a waiter about to park, called under `wait_lock`, and
181    /// signals it on the test condvar.
182    #[cfg(test)]
183    fn signal_parked(&self) {
184        self.count_and_signal(&self.parked);
185    }
186
187    /// Test-only: raises `count` and signals it under `signal_lock`, so a
188    /// helper that read the count under that lock and is about to wait cannot
189    /// miss the signal.
190    ///
191    /// A poisoned `signal_lock` is neither recovered nor a panic (a panic in
192    /// `Returned`'s drop during unwinding would abort the test binary): the
193    /// count rises and the signal goes with no guard, and the helper's next
194    /// lock refuses the poison by name. That arm is unreachable: the lock's
195    /// only holders are this count and the helpers, and none of them panics.
196    #[cfg(test)]
197    fn count_and_signal(&self, count: &std::sync::atomic::AtomicUsize) {
198        let guard = self.signal_lock.lock();
199        count.fetch_add(1, Ordering::SeqCst);
200        self.signal.notify_all();
201        drop(guard);
202    }
203
204    fn notify(&self) {
205        match self.wait_lock.lock() {
206            Ok(_guard) => self.waiter.notify_all(),
207            Err(poisoned) => {
208                tracing::error!("shutdown wait lock poisoned; notifying waiters anyway");
209                // Recover the guard from the poison so the notify_all still
210                // runs under the lock: a bare notify-without-lock could race a
211                // waiter between its under-the-lock `is_initiated` check and
212                // its park, losing the wake. Holding the recovered guard keeps
213                // the initiate-sets-flag-then-notify ordering airtight while a
214                // parked waiter (which released the lock) is still woken.
215                let _guard = poisoned.into_inner();
216                self.waiter.notify_all();
217            }
218        }
219        // A test's count that the wake above was sent, on either arm.
220        #[cfg(test)]
221        self.notified.fetch_add(1, Ordering::SeqCst);
222    }
223}
224
225/// Process-global OS signal registration for graceful shutdown.
226#[derive(Debug)]
227pub struct SignalShutdownRegistration {
228    signal_handle: SignalIteratorHandle,
229    worker: Option<JoinHandle<()>>,
230}
231
232impl SignalShutdownRegistration {
233    const fn new(signal_handle: SignalIteratorHandle, worker: JoinHandle<()>) -> Self {
234        Self {
235            signal_handle,
236            worker: Some(worker),
237        }
238    }
239}
240
241impl Drop for SignalShutdownRegistration {
242    fn drop(&mut self) {
243        self.signal_handle.close();
244        let Some(worker) = self.worker.take() else {
245            return;
246        };
247        if worker.join().is_err() {
248            tracing::warn!("shutdown signal worker terminated unexpectedly");
249        }
250    }
251}
252
253/// Registers SIGTERM and SIGINT handlers that initiate the supplied handle.
254///
255/// # Errors
256/// Returns [`ServerError::ListenerAccept`] when the OS signal registration fails.
257pub fn register_signal_handlers(
258    handle: ShutdownHandle,
259) -> Result<SignalShutdownRegistration, ServerError> {
260    let mut signals =
261        Signals::new([SIGTERM, SIGINT]).map_err(|error| ServerError::ListenerAccept {
262            message: format!("failed to register shutdown signal handlers: {error}"),
263        })?;
264    let signal_handle = signals.handle();
265    let worker = thread::spawn(move || {
266        for signal in signals.forever() {
267            tracing::info!(signal, "received shutdown signal");
268            handle.initiate();
269        }
270    });
271    Ok(SignalShutdownRegistration::new(signal_handle, worker))
272}
273
274/// Runs the shutdown sequence after the handle has been activated.
275///
276/// Five steps, in this order, with no timer between them: stop accepting, close
277/// every connection this server holds, settle the connection fates the close
278/// parked, flush durable channel state, exit.
279///
280/// The optional sibling WebSocket listener (LP-WS-TRANSPORT R1) stops
281/// accepting — and interrupts its in-flight upgrade handshakes — in the same
282/// pre-notification window as the main listener, so no connection on EITHER
283/// transport can slip past the shutdown broadcast. Already-admitted WebSocket
284/// connections live in the shared supervisor and are closed by the same
285/// sequence below.
286///
287/// # The retired drain (0.14.3)
288///
289/// Until 0.14.3 this function waited up to `drain_timeout` for connections to
290/// hang up by themselves before closing them. `drain_timeout` is now IGNORED:
291/// it is still accepted so an existing config file and an existing embedder
292/// call site both keep compiling and loading, and it is logged once, by name,
293/// as ignored. See [`WEDGED_CONNECTION_STOP`] for why the wait bought nothing
294/// and what the one remaining bound is for.
295///
296/// The close itself is unchanged and was always the orderly one: each
297/// connection process enqueues a protocol `Disconnect`, drains its outbound
298/// buffer, completes its connection fate as `ServerShutdown`, and — on the
299/// WebSocket transport — writes a close frame carrying `CloseCode::Away` and
300/// the reason "server shutdown" before exiting `Normal`. A peer reads a
301/// shutdown, never a reset.
302///
303/// # Errors
304/// Returns [`ServerError`] when stop-accepting or durable flush fails, and
305/// [`ServerError::ShutdownUnsettledConnectionFates`] when a parked connection
306/// fate could not be completed before exit.
307pub fn run_shutdown_sequence(
308    listener: &mut ServerListener,
309    websocket_listener: Option<&mut WebSocketListener>,
310    supervisor: &ConnectionSupervisor,
311    drain_timeout: Duration,
312) -> Result<(), ServerError> {
313    run_shutdown_sequence_before_flush(
314        listener,
315        websocket_listener,
316        supervisor,
317        drain_timeout,
318        || Ok(()),
319    )
320}
321
322pub(in crate::server) fn run_shutdown_sequence_before_flush(
323    listener: &mut ServerListener,
324    websocket_listener: Option<&mut WebSocketListener>,
325    supervisor: &ConnectionSupervisor,
326    drain_timeout: Duration,
327    finalize: impl FnOnce() -> Result<(), ServerError>,
328) -> Result<(), ServerError> {
329    let started = Instant::now();
330    tracing::info!(
331        ignored_drain_timeout = ?drain_timeout,
332        "starting shutdown sequence; the configured drain timeout has been ignored since 0.14.3 \
333         because every write is durable before it is acknowledged, so no request is in flight to \
334         drain"
335    );
336    // Stop accepting new connections first so none can slip into the accept
337    // window after shutdown begins and miss the notification broadcast below.
338    if let Some(websocket_listener) = websocket_listener {
339        websocket_listener.stop_accepting()?;
340    }
341    listener.stop_accepting()?;
342
343    // FIX A-ii: flush accepted-but-unfanned-out publishes to their subscriber
344    // connections BEFORE broadcasting the shutdown Disconnect. Accept is now
345    // stopped, so the set of accepted publishes is bounded; this TOLD barrier
346    // parks on the delivery-quiescence signal (a connection parks only once every
347    // accepted publish has been pumped to its socket) and returns the instant it
348    // arrives. Without it, `notify_shutdown_subscribers` below could enqueue a
349    // subscriber's Disconnect ahead of an in-flight fan-out (measured 8-131 ms)
350    // and the subscriber's reader would exit before delivery. The bound is the
351    // hang stop, not a budget anything waits out; missing it is logged, not
352    // fatal — the close and flush legs below still run.
353    if !supervisor.wait_for_delivery_quiesced(Instant::now() + WEDGED_CONNECTION_STOP) {
354        tracing::warn!(
355            stop = ?WEDGED_CONNECTION_STOP,
356            "delivery flush barrier completion was not observed before the wedged-connection stop; \
357             proceeding to shutdown notification"
358        );
359    }
360
361    supervisor.notify_shutdown_subscribers();
362
363    // The close is the mechanism, and it is now the only route. Nothing waits
364    // for a peer to hang up first.
365    supervisor.force_close_active_connections();
366    wait_after_force_close(supervisor);
367
368    // A connection fate refused by candidate-lane precedence during the close
369    // is parked, and nothing will publish to drain its lane now. Settle it
370    // here, as the next boot otherwise would, so a clean stop leaves nothing
371    // to reconcile; a fate that cannot be settled is named and the stop is
372    // reported unclean once the flush and teardown below have still run.
373    let settled = supervisor.settle_parked_connection_fates();
374
375    supervisor.shutdown();
376    let finalized = finalize();
377    supervisor.stop_participant_warm();
378    let flushed = flush_durable_state(supervisor);
379    tracing::info!(
380        elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
381        clean = settled.is_ok() && finalized.is_ok() && flushed.is_ok(),
382        "shutdown sequence complete"
383    );
384    let mut failure = None;
385    for result in [finalized, flushed, settled] {
386        if let Err(error) = result {
387            tracing::error!(%error, "shutdown phase failed");
388            if failure.is_none() {
389                failure = Some(error);
390            }
391        }
392    }
393    failure.map_or(Ok(()), Err)
394}
395
396/// Waits for the closed connections to deliver their exits.
397///
398/// Parks on the supervisor's TOLD exit notification (§4.3) — every exit route
399/// funnels through the one `remove()` teardown that bumps the drain generation
400/// — and returns the instant the last one lands. There is no poll loop and no
401/// reap scan; [`WEDGED_CONNECTION_STOP`] only stops a wedged process from
402/// holding the restart open.
403///
404/// The name is kept from the pre-0.14.3 shape, where this ran only after a
405/// drain had expired. Since 0.14.3 the close is the sole route, so this runs on
406/// every shutdown.
407pub(crate) fn wait_after_force_close(supervisor: &ConnectionSupervisor) {
408    let deadline = Instant::now() + WEDGED_CONNECTION_STOP;
409    if supervisor.wait_for_connections_drained(deadline) {
410        return;
411    }
412    let remaining = supervisor.active_connection_count();
413    if remaining > 0 {
414        tracing::warn!(
415            active_connections = remaining,
416            stop = ?WEDGED_CONNECTION_STOP,
417            "connections remained active after the wedged-connection stop"
418        );
419    }
420}
421
422fn flush_durable_state(supervisor: &ConnectionSupervisor) -> Result<(), ServerError> {
423    tracing::info!("flushing durable channel state");
424    supervisor.flush_durable_state().map_err(|error| {
425        tracing::error!(%error, "durable state flush failed during shutdown");
426        match error {
427            ServerError::ShutdownFlush { .. } => error,
428            other => ServerError::ShutdownFlush {
429                message: other.to_string(),
430            },
431        }
432    })?;
433    tracing::info!("durable channel state flushed");
434    Ok(())
435}
436
437#[cfg(test)]
438#[path = "shutdown_inflight_tests.rs"]
439mod shutdown_inflight_tests;
440
441#[cfg(test)]
442mod tests {
443    use std::error::Error;
444    use std::sync::atomic::Ordering;
445    use std::thread;
446    use std::time::Duration;
447
448    use super::{ShutdownHandle, wait_after_force_close};
449    use crate::server::connection::ConnectionSupervisor;
450
451    #[test]
452    fn shutdown_handle_initiates_once() {
453        let handle = ShutdownHandle::new();
454
455        assert!(!handle.is_initiated());
456        assert!(handle.initiate());
457        assert!(handle.is_initiated());
458        assert!(!handle.initiate());
459    }
460
461    /// What a helper answers when the test signal lock is poisoned.
462    const SIGNAL_LOCK_POISONED: &str =
463        "the test signal lock was poisoned: another test thread panicked holding it";
464
465    /// Blocks until a waiter has parked in `wait`, or fails by name if it
466    /// returned without parking. Both test-only counts rise under
467    /// `signal_lock` and are signalled on the test condvar, and this wakes
468    /// holding that lock, so neither can be missed and nothing here waits on
469    /// something that might never come.
470    fn await_parked(handle: &ShutdownHandle) -> Result<(), Box<dyn Error>> {
471        let state = &handle.inner;
472        let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
473        loop {
474            if state.returned.load(Ordering::SeqCst) > 0 {
475                return Err("the waiter returned from wait() without parking".into());
476            }
477            if state.parked.load(Ordering::SeqCst) > 0 {
478                drop(guard);
479                return Ok(());
480            }
481            guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
482        }
483    }
484
485    /// After `initiate`, blocks until the parked waiter has returned, or fails
486    /// by name when it cannot: `notify` never reached its condvar, so no wake
487    /// was sent; or the waiter woke and parked again (`parked` rose past
488    /// `parked_before`, read before `initiate`), so it did not see the
489    /// shutdown and nothing will wake it again. Every rise of `parked` and
490    /// `returned` is signalled on the test condvar under `signal_lock`, which
491    /// is refused by name if poisoned, never recovered. `wait_lock` is not
492    /// taken here, so a test that poisons it still reads the waiter's return.
493    fn await_returned(handle: &ShutdownHandle, parked_before: usize) -> Result<(), Box<dyn Error>> {
494        let state = &handle.inner;
495        if state.notified.load(Ordering::SeqCst) == 0 {
496            return Err("initiate() never notified the parked waiter".into());
497        }
498        let mut guard = state.signal_lock.lock().map_err(|_| SIGNAL_LOCK_POISONED)?;
499        while state.returned.load(Ordering::SeqCst) == 0 {
500            if state.parked.load(Ordering::SeqCst) > parked_before {
501                return Err("the waiter woke and parked again: it did not see the shutdown".into());
502            }
503            guard = state.signal.wait(guard).map_err(|_| SIGNAL_LOCK_POISONED)?;
504        }
505        drop(guard);
506        Ok(())
507    }
508
509    #[test]
510    fn shutdown_handle_wait_unblocks_on_initiate() -> Result<(), Box<dyn std::error::Error>> {
511        let handle = ShutdownHandle::new();
512        let waiter = handle.clone();
513        let worker = thread::spawn(move || {
514            waiter.wait();
515            waiter.is_initiated()
516        });
517
518        await_parked(&handle)?;
519        let parked = handle.inner.parked.load(Ordering::SeqCst);
520        assert!(handle.initiate());
521        await_returned(&handle, parked)?;
522        let observed = worker.join().map_err(|_| "wait worker panicked")?;
523
524        assert!(observed);
525        Ok(())
526    }
527
528    /// The production lock under test.
529    fn wait_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
530        &state.wait_lock
531    }
532
533    /// The test helpers' own lock.
534    fn signal_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
535        &state.signal_lock
536    }
537
538    /// Poisons the handle's lock that `lock` names by panicking a thread
539    /// while it holds the guard.
540    #[allow(clippy::panic)]
541    fn poison(handle: &ShutdownHandle, lock: fn(&super::ShutdownState) -> &std::sync::Mutex<()>) {
542        let poisoner = handle.clone();
543        let outcome = thread::spawn(move || {
544            let _guard = lock(&poisoner.inner).lock();
545            panic!("poison a shutdown lock");
546        })
547        .join();
548        assert!(outcome.is_err(), "the poisoner thread must have panicked");
549    }
550
551    /// Poisons the handle's `wait_lock`.
552    fn poison_wait_lock(handle: &ShutdownHandle) {
553        poison(handle, wait_lock_of);
554    }
555
556    /// A poisoned test signal lock is refused by name, never recovered: the
557    /// helpers answer an error that names it.
558    #[test]
559    fn a_poisoned_test_signal_lock_is_refused_by_name() {
560        let handle = ShutdownHandle::new();
561        poison(&handle, signal_lock_of);
562        assert!(handle.initiate());
563
564        let refused =
565            await_returned(&handle, 0).map_or_else(|error| error.to_string(), |()| String::new());
566        assert_eq!(refused, SIGNAL_LOCK_POISONED);
567        let refused =
568            await_parked(&handle).map_or_else(|error| error.to_string(), |()| String::new());
569        assert_eq!(refused, SIGNAL_LOCK_POISONED);
570    }
571
572    /// A poisoned wait lock must not silently masquerade as a completed
573    /// shutdown: `wait()` still returns (behaviour unchanged) but logs at
574    /// error level.
575    #[test]
576    fn wait_on_poisoned_lock_returns_and_logs_error() {
577        let handle = ShutdownHandle::new();
578        poison_wait_lock(&handle);
579
580        let log = crate::test_log::CapturedLog::default();
581        log.capture(|| handle.wait());
582
583        let output = log.contents();
584        assert!(
585            output.contains("ERROR"),
586            "a poisoned wait lock must be logged at error level, got: {output}"
587        );
588    }
589
590    /// A waiter already parked in `wait()` must still be woken by `notify()`
591    /// when the lock has been poisoned after it parked: `initiate()` must never
592    /// silently skip the wake.
593    #[test]
594    fn notify_on_poisoned_lock_still_wakes_parked_waiter() -> Result<(), Box<dyn std::error::Error>>
595    {
596        let handle = ShutdownHandle::new();
597        let waiter = handle.clone();
598        let worker = thread::spawn(move || waiter.wait());
599        // The waiter parks on the condvar, releasing the lock, so the poisoner
600        // below can take it.
601        await_parked(&handle)?;
602        poison_wait_lock(&handle);
603
604        let parked = handle.inner.parked.load(Ordering::SeqCst);
605        assert!(handle.initiate());
606
607        // A parked waiter must wake after notify under a poisoned lock.
608        await_returned(&handle, parked)?;
609        worker.join().map_err(|_| "wait worker panicked")?;
610        Ok(())
611    }
612
613    /// The close-settle wait returns immediately when nothing is tracked, so a
614    /// shutdown with no connections spends no time here at all.
615    #[test]
616    fn the_close_settle_returns_immediately_when_no_connections_are_active()
617    -> Result<(), Box<dyn std::error::Error>> {
618        let supervisor = ConnectionSupervisor::new()?;
619
620        let started = std::time::Instant::now();
621        wait_after_force_close(&supervisor);
622        let elapsed = started.elapsed();
623
624        assert!(
625            elapsed < Duration::from_millis(50),
626            "the close settle took {elapsed:?} with no connections tracked"
627        );
628        supervisor.shutdown();
629        Ok(())
630    }
631
632    /// Oracle 13 (W4 leg 3, §4.3) — absence proof over the close/settle
633    /// implementation (this module before its `mod tests`): none of the retired
634    /// poll constants nor the per-iteration reap scan survive, AND neither does
635    /// the graceful drain retired in 0.14.3. The forbid-list literals below live
636    /// in the test section, so `split` excludes them from the implementation
637    /// slice under inspection.
638    #[test]
639    fn shutdown_source_has_no_drain_and_no_reap_count_sleep_loop() {
640        let source = include_str!("shutdown.rs");
641        // `split` always yields a first segment; `unwrap_or` keeps this panic-free
642        // under the workspace lint deny while never falling back in practice.
643        let implementation = source.split("mod tests").next().unwrap_or(source);
644        for forbidden in [
645            "DRAIN_PROGRESS_INTERVAL",
646            "FORCE_CLOSE_SETTLE_TIMEOUT",
647            "FORCE_CLOSE_POLL_INTERVAL",
648            "reap_crashed_connections",
649            "fn drain_connections",
650        ] {
651            assert!(
652                !implementation.contains(forbidden),
653                "retired poll/reap/drain token `{forbidden}` must not appear in the shutdown implementation"
654            );
655        }
656    }
657
658    /// The configured drain timeout reaches the shutdown path and is used for
659    /// exactly one thing: being named in the line that says it is ignored. If a
660    /// future edit re-arms it as a deadline, this goes red.
661    #[test]
662    fn the_configured_drain_timeout_is_never_turned_into_a_deadline() {
663        let source = include_str!("shutdown.rs");
664        let implementation = source.split("mod tests").next().unwrap_or(source);
665        assert!(
666            !implementation.contains("+ drain_timeout"),
667            "drain_timeout must never be added to an Instant to form a shutdown deadline"
668        );
669        assert!(
670            implementation.contains("ignored_drain_timeout = ?drain_timeout"),
671            "drain_timeout must still be named in the shutdown log line that reports it ignored"
672        );
673    }
674}