liminal-server 0.19.0

use std::fmt;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};

use signal_hook::consts::signal::{SIGINT, SIGTERM};
use signal_hook::iterator::{Handle as SignalIteratorHandle, Signals};

use crate::ServerError;
use crate::server::connection::{ConnectionSupervisor, WebSocketListener};
use crate::server::listener::ServerListener;

/// The ONE bounded park left in the shutdown path, and it is a HANG STOP — not
/// a grace period, not a drain, and never a delay anything waits out.
///
/// 0.14.3 retired the graceful drain because its stated purpose did not hold.
/// A drain exists to let in-flight requests finish; in this server every write
/// is durable and flushed BEFORE it is acknowledged (`OperationLog::append`
/// flushes the store inside the append), so at the instant shutdown begins
/// there is nothing in flight to finish. What the drain actually did on an
/// estate whose connections are long-lived idle seats — seats that never hang
/// up by themselves — was wait out its whole configured budget (5 s on Tom's
/// estate, measured 2026-09-14) and then force-close anyway. The close was
/// always the real mechanism; the wait in front of it bought nothing.
///
/// Both remaining waits park on an EVENT (the supervisor's delivery-quiescence
/// signal and its TOLD drain-completion notification, W4 leg 3 §4.3) and return
/// the instant that event arrives — with every connection idle they return in
/// microseconds. This bound exists solely so that ONE wedged connection process
/// cannot hold a restart open forever; it is the stop on a hang, which is why
/// it is a number at all and why it is short.
pub(crate) const WEDGED_CONNECTION_STOP: Duration = Duration::from_millis(500);

/// Idempotent shutdown activation handle shared by the runtime and signal thread.
#[derive(Clone)]
pub struct ShutdownHandle {
    inner: Arc<ShutdownState>,
}

impl ShutdownHandle {
    /// Creates a new inactive shutdown handle.
    #[must_use]
    pub fn new() -> Self {
        Self {
            inner: Arc::new(ShutdownState::new()),
        }
    }

    /// Initiates shutdown exactly once.
    ///
    /// Returns `true` for the first caller that transitions the handle to active,
    /// and `false` for subsequent calls.
    pub fn initiate(&self) -> bool {
        if self.inner.initiated.swap(true, Ordering::SeqCst) {
            tracing::debug!("shutdown request ignored because shutdown is already active");
            return false;
        }

        tracing::info!("shutdown requested");
        self.inner.notify();
        true
    }

    /// Blocks until shutdown is initiated.
    pub fn wait(&self) {
        // A test's count of every return from here, taken as this guard
        // drops, so no test waits on a waiter that left without parking.
        #[cfg(test)]
        let _returned = Returned(&self.inner);
        if self.is_initiated() {
            return;
        }
        let Ok(mut guard) = self.inner.wait_lock.lock() else {
            // Behaviour unchanged (return as if shutdown was requested), but a
            // poisoned lock must not masquerade silently as a completed wait.
            tracing::error!(
                "shutdown wait lock poisoned; returning from wait as if shutdown was requested"
            );
            return;
        };
        while !self.is_initiated() {
            // A test's signal that this waiter is about to release the lock by
            // parking, so no test waits on a clock for it.
            #[cfg(test)]
            self.inner.signal_parked();
            let Ok(next_guard) = self.inner.waiter.wait(guard) else {
                tracing::error!(
                    "shutdown wait condvar reacquired a poisoned lock; returning from wait as if shutdown was requested"
                );
                return;
            };
            guard = next_guard;
        }
    }

    /// Returns whether shutdown has been initiated.
    #[must_use]
    pub fn is_initiated(&self) -> bool {
        self.inner.initiated.load(Ordering::SeqCst)
    }
}

impl Default for ShutdownHandle {
    fn default() -> Self {
        Self::new()
    }
}

impl fmt::Debug for ShutdownHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ShutdownHandle")
            .field("initiated", &self.is_initiated())
            .finish()
    }
}

#[derive(Debug)]
struct ShutdownState {
    initiated: AtomicBool,
    wait_lock: Mutex<()>,
    waiter: Condvar,
    /// Times a waiter parked on `waiter`, counted under `wait_lock`.
    #[cfg(test)]
    parked: std::sync::atomic::AtomicUsize,
    /// Times `wait` returned, counted under `wait_lock`.
    #[cfg(test)]
    returned: std::sync::atomic::AtomicUsize,
    /// Times `notify` sent its wake to `waiter`.
    #[cfg(test)]
    notified: std::sync::atomic::AtomicUsize,
    /// The test-only condvar `parked` and `returned` are signalled on, kept
    /// apart from `waiter` so a test build never wakes a production waiter.
    #[cfg(test)]
    signal: Condvar,
    /// The test-only lock `signal` pairs with: `parked` and `returned` rise and
    /// are signalled under it, and the test helpers wait under it. It is kept
    /// apart from `wait_lock`, the production lock under test, so a test that
    /// poisons `wait_lock` on purpose still reads every count, and nothing
    /// recovers a poisoned guard. Lock order, the only nesting: `wait_lock`,
    /// then `signal_lock` (in `wait`, around `signal_parked`). Nothing takes
    /// `signal_lock` and then `wait_lock`.
    #[cfg(test)]
    signal_lock: Mutex<()>,
}

/// Test-only: counts one return from [`ShutdownHandle::wait`] as it drops,
/// under `signal_lock`, and signals it. It drops after `wait`'s own guard, so
/// `wait_lock` is not held here.
#[cfg(test)]
struct Returned<'a>(&'a ShutdownState);

#[cfg(test)]
impl Drop for Returned<'_> {
    fn drop(&mut self) {
        self.0.count_and_signal(&self.0.returned);
    }
}

impl ShutdownState {
    const fn new() -> Self {
        Self {
            initiated: AtomicBool::new(false),
            wait_lock: Mutex::new(()),
            waiter: Condvar::new(),
            #[cfg(test)]
            parked: std::sync::atomic::AtomicUsize::new(0),
            #[cfg(test)]
            returned: std::sync::atomic::AtomicUsize::new(0),
            #[cfg(test)]
            notified: std::sync::atomic::AtomicUsize::new(0),
            #[cfg(test)]
            signal: Condvar::new(),
            #[cfg(test)]
            signal_lock: Mutex::new(()),
        }
    }

    /// Test-only: counts a waiter about to park, called under `wait_lock`, and
    /// signals it on the test condvar.
    #[cfg(test)]
    fn signal_parked(&self) {
        self.count_and_signal(&self.parked);
    }

    /// Test-only: raises `count` and signals it under `signal_lock`, so a
    /// helper that read the count under that lock and is about to wait cannot
    /// miss the signal.
    ///
    /// A poisoned `signal_lock` is neither recovered nor a panic (a panic in
    /// `Returned`'s drop during unwinding would abort the test binary): the
    /// count rises and the signal goes with no guard, and the helper's next
    /// lock refuses the poison by name. That arm is unreachable: the lock's
    /// only holders are this count and the helpers, and none of them panics.
    #[cfg(test)]
    fn count_and_signal(&self, count: &std::sync::atomic::AtomicUsize) {
        let guard = self.signal_lock.lock();
        count.fetch_add(1, Ordering::SeqCst);
        self.signal.notify_all();
        drop(guard);
    }

    fn notify(&self) {
        match self.wait_lock.lock() {
            Ok(_guard) => self.waiter.notify_all(),
            Err(poisoned) => {
                tracing::error!("shutdown wait lock poisoned; notifying waiters anyway");
                // Recover the guard from the poison so the notify_all still
                // runs under the lock: a bare notify-without-lock could race a
                // waiter between its under-the-lock `is_initiated` check and
                // its park, losing the wake. Holding the recovered guard keeps
                // the initiate-sets-flag-then-notify ordering airtight while a
                // parked waiter (which released the lock) is still woken.
                let _guard = poisoned.into_inner();
                self.waiter.notify_all();
            }
        }
        // A test's count that the wake above was sent, on either arm.
        #[cfg(test)]
        self.notified.fetch_add(1, Ordering::SeqCst);
    }
}

/// Process-global OS signal registration for graceful shutdown.
#[derive(Debug)]
pub struct SignalShutdownRegistration {
    signal_handle: SignalIteratorHandle,
    worker: Option<JoinHandle<()>>,
}

impl SignalShutdownRegistration {
    const fn new(signal_handle: SignalIteratorHandle, worker: JoinHandle<()>) -> Self {
        Self {
            signal_handle,
            worker: Some(worker),
        }
    }
}

impl Drop for SignalShutdownRegistration {
    fn drop(&mut self) {
        self.signal_handle.close();
        let Some(worker) = self.worker.take() else {
            return;
        };
        if worker.join().is_err() {
            tracing::warn!("shutdown signal worker terminated unexpectedly");
        }
    }
}

/// Registers SIGTERM and SIGINT handlers that initiate the supplied handle.
///
/// # Errors
/// Returns [`ServerError::ListenerAccept`] when the OS signal registration fails.
pub fn register_signal_handlers(
    handle: ShutdownHandle,
) -> Result<SignalShutdownRegistration, ServerError> {
    let mut signals =
        Signals::new([SIGTERM, SIGINT]).map_err(|error| ServerError::ListenerAccept {
            message: format!("failed to register shutdown signal handlers: {error}"),
        })?;
    let signal_handle = signals.handle();
    let worker = thread::spawn(move || {
        for signal in signals.forever() {
            tracing::info!(signal, "received shutdown signal");
            handle.initiate();
        }
    });
    Ok(SignalShutdownRegistration::new(signal_handle, worker))
}

/// Runs the shutdown sequence after the handle has been activated.
///
/// Five steps, in this order, with no timer between them: stop accepting, close
/// every connection this server holds, settle the connection fates the close
/// parked, flush durable channel state, exit.
///
/// The optional sibling WebSocket listener (LP-WS-TRANSPORT R1) stops
/// accepting — and interrupts its in-flight upgrade handshakes — in the same
/// pre-notification window as the main listener, so no connection on EITHER
/// transport can slip past the shutdown broadcast. Already-admitted WebSocket
/// connections live in the shared supervisor and are closed by the same
/// sequence below.
///
/// # The retired drain (0.14.3)
///
/// Until 0.14.3 this function waited up to `drain_timeout` for connections to
/// hang up by themselves before closing them. `drain_timeout` is now IGNORED:
/// it is still accepted so an existing config file and an existing embedder
/// call site both keep compiling and loading, and it is logged once, by name,
/// as ignored. See [`WEDGED_CONNECTION_STOP`] for why the wait bought nothing
/// and what the one remaining bound is for.
///
/// The close itself is unchanged and was always the orderly one: each
/// connection process enqueues a protocol `Disconnect`, drains its outbound
/// buffer, completes its connection fate as `ServerShutdown`, and — on the
/// WebSocket transport — writes a close frame carrying `CloseCode::Away` and
/// the reason "server shutdown" before exiting `Normal`. A peer reads a
/// shutdown, never a reset.
///
/// # Errors
/// Returns [`ServerError`] when stop-accepting or durable flush fails, and
/// [`ServerError::ShutdownUnsettledConnectionFates`] when a parked connection
/// fate could not be completed before exit.
pub fn run_shutdown_sequence(
    listener: &mut ServerListener,
    websocket_listener: Option<&mut WebSocketListener>,
    supervisor: &ConnectionSupervisor,
    drain_timeout: Duration,
) -> Result<(), ServerError> {
    run_shutdown_sequence_before_flush(
        listener,
        websocket_listener,
        supervisor,
        drain_timeout,
        || Ok(()),
    )
}

pub(in crate::server) fn run_shutdown_sequence_before_flush(
    listener: &mut ServerListener,
    websocket_listener: Option<&mut WebSocketListener>,
    supervisor: &ConnectionSupervisor,
    drain_timeout: Duration,
    finalize: impl FnOnce() -> Result<(), ServerError>,
) -> Result<(), ServerError> {
    let started = Instant::now();
    tracing::info!(
        ignored_drain_timeout = ?drain_timeout,
        "starting shutdown sequence; the configured drain timeout has been ignored since 0.14.3 \
         because every write is durable before it is acknowledged, so no request is in flight to \
         drain"
    );
    // Stop accepting new connections first so none can slip into the accept
    // window after shutdown begins and miss the notification broadcast below.
    if let Some(websocket_listener) = websocket_listener {
        websocket_listener.stop_accepting()?;
    }
    listener.stop_accepting()?;

    // FIX A-ii: flush accepted-but-unfanned-out publishes to their subscriber
    // connections BEFORE broadcasting the shutdown Disconnect. Accept is now
    // stopped, so the set of accepted publishes is bounded; this TOLD barrier
    // parks on the delivery-quiescence signal (a connection parks only once every
    // accepted publish has been pumped to its socket) and returns the instant it
    // arrives. Without it, `notify_shutdown_subscribers` below could enqueue a
    // subscriber's Disconnect ahead of an in-flight fan-out (measured 8-131 ms)
    // and the subscriber's reader would exit before delivery. The bound is the
    // hang stop, not a budget anything waits out; missing it is logged, not
    // fatal — the close and flush legs below still run.
    if !supervisor.wait_for_delivery_quiesced(Instant::now() + WEDGED_CONNECTION_STOP) {
        tracing::warn!(
            stop = ?WEDGED_CONNECTION_STOP,
            "delivery flush barrier completion was not observed before the wedged-connection stop; \
             proceeding to shutdown notification"
        );
    }

    supervisor.notify_shutdown_subscribers();

    // The close is the mechanism, and it is now the only route. Nothing waits
    // for a peer to hang up first.
    supervisor.force_close_active_connections();
    wait_after_force_close(supervisor);

    // A connection fate refused by candidate-lane precedence during the close
    // is parked, and nothing will publish to drain its lane now. Settle it
    // here, as the next boot otherwise would, so a clean stop leaves nothing
    // to reconcile; a fate that cannot be settled is named and the stop is
    // reported unclean once the flush and teardown below have still run.
    let settled = supervisor.settle_parked_connection_fates();

    supervisor.shutdown();
    let finalized = finalize();
    let flushed = flush_durable_state(supervisor);
    tracing::info!(
        elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
        clean = settled.is_ok() && finalized.is_ok() && flushed.is_ok(),
        "shutdown sequence complete"
    );
    let mut failure = None;
    for result in [finalized, flushed, settled] {
        if let Err(error) = result {
            tracing::error!(%error, "shutdown phase failed");
            if failure.is_none() {
                failure = Some(error);
            }
        }
    }
    failure.map_or(Ok(()), Err)
}

/// Waits for the closed connections to deliver their exits.
///
/// Parks on the supervisor's TOLD exit notification (§4.3) — every exit route
/// funnels through the one `remove()` teardown that bumps the drain generation
/// — and returns the instant the last one lands. There is no poll loop and no
/// reap scan; [`WEDGED_CONNECTION_STOP`] only stops a wedged process from
/// holding the restart open.
///
/// The name is kept from the pre-0.14.3 shape, where this ran only after a
/// drain had expired. Since 0.14.3 the close is the sole route, so this runs on
/// every shutdown.
pub(crate) fn wait_after_force_close(supervisor: &ConnectionSupervisor) {
    let deadline = Instant::now() + WEDGED_CONNECTION_STOP;
    if supervisor.wait_for_connections_drained(deadline) {
        return;
    }
    let remaining = supervisor.active_connection_count();
    if remaining > 0 {
        tracing::warn!(
            active_connections = remaining,
            stop = ?WEDGED_CONNECTION_STOP,
            "connections remained active after the wedged-connection stop"
        );
    }
}

fn flush_durable_state(supervisor: &ConnectionSupervisor) -> Result<(), ServerError> {
    tracing::info!("flushing durable channel state");
    supervisor.flush_durable_state().map_err(|error| {
        tracing::error!(%error, "durable state flush failed during shutdown");
        match error {
            ServerError::ShutdownFlush { .. } => error,
            other => ServerError::ShutdownFlush {
                message: other.to_string(),
            },
        }
    })?;
    tracing::info!("durable channel state flushed");
    Ok(())
}

#[cfg(test)]
#[path = "shutdown_inflight_tests.rs"]
mod shutdown_inflight_tests;

#[cfg(test)]
mod tests {
    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());
    }

    /// What a helper answers when the test signal lock is poisoned.
    const SIGNAL_LOCK_POISONED: &str =
        "the test signal lock was poisoned: another test thread panicked holding it";

    /// Blocks until a waiter has parked in `wait`, or fails by name if it
    /// returned without parking. Both test-only counts rise under
    /// `signal_lock` and are signalled on the test condvar, and this wakes
    /// holding that lock, so neither can be missed and nothing here waits on
    /// something that might never come.
    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)?;
        }
    }

    /// After `initiate`, blocks until the parked waiter has returned, or fails
    /// by name when it cannot: `notify` never reached its condvar, so no wake
    /// was sent; or the waiter woke and parked again (`parked` rose past
    /// `parked_before`, read before `initiate`), so it did not see the
    /// shutdown and nothing will wake it again. Every rise of `parked` and
    /// `returned` is signalled on the test condvar under `signal_lock`, which
    /// is refused by name if poisoned, never recovered. `wait_lock` is not
    /// taken here, so a test that poisons it still reads the waiter's return.
    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(())
    }

    /// The production lock under test.
    fn wait_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
        &state.wait_lock
    }

    /// The test helpers' own lock.
    fn signal_lock_of(state: &super::ShutdownState) -> &std::sync::Mutex<()> {
        &state.signal_lock
    }

    /// Poisons the handle's lock that `lock` names by panicking a thread
    /// while it holds the guard.
    #[allow(clippy::panic)]
    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();
            panic!("poison a shutdown lock");
        })
        .join();
        assert!(outcome.is_err(), "the poisoner thread must have panicked");
    }

    /// Poisons the handle's `wait_lock`.
    fn poison_wait_lock(handle: &ShutdownHandle) {
        poison(handle, wait_lock_of);
    }

    /// A poisoned test signal lock is refused by name, never recovered: the
    /// helpers answer an error that names it.
    #[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);
    }

    /// A poisoned wait lock must not silently masquerade as a completed
    /// shutdown: `wait()` still returns (behaviour unchanged) but logs at
    /// error level.
    #[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}"
        );
    }

    /// A waiter already parked in `wait()` must still be woken by `notify()`
    /// when the lock has been poisoned after it parked: `initiate()` must never
    /// silently skip the wake.
    #[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());
        // The waiter parks on the condvar, releasing the lock, so the poisoner
        // below can take it.
        await_parked(&handle)?;
        poison_wait_lock(&handle);

        let parked = handle.inner.parked.load(Ordering::SeqCst);
        assert!(handle.initiate());

        // A parked waiter must wake after notify under a poisoned lock.
        await_returned(&handle, parked)?;
        worker.join().map_err(|_| "wait worker panicked")?;
        Ok(())
    }

    /// The close-settle wait returns immediately when nothing is tracked, so a
    /// shutdown with no connections spends no time here at all.
    #[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(())
    }

    /// Oracle 13 (W4 leg 3, §4.3) — absence proof over the close/settle
    /// implementation (this module before its `mod tests`): none of the retired
    /// poll constants nor the per-iteration reap scan survive, AND neither does
    /// the graceful drain retired in 0.14.3. The forbid-list literals below live
    /// in the test section, so `split` excludes them from the implementation
    /// slice under inspection.
    #[test]
    fn shutdown_source_has_no_drain_and_no_reap_count_sleep_loop() {
        let source = include_str!("shutdown.rs");
        // `split` always yields a first segment; `unwrap_or` keeps this panic-free
        // under the workspace lint deny while never falling back in practice.
        let implementation = source.split("mod tests").next().unwrap_or(source);
        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"
            );
        }
    }

    /// The configured drain timeout reaches the shutdown path and is used for
    /// exactly one thing: being named in the line that says it is ignored. If a
    /// future edit re-arms it as a deadline, this goes red.
    #[test]
    fn the_configured_drain_timeout_is_never_turned_into_a_deadline() {
        let source = include_str!("shutdown.rs");
        let implementation = source.split("mod tests").next().unwrap_or(source);
        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"
        );
    }
}