liminal-server 0.25.4

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> {
    crate::server::runtime::runtime_links::shutdown(
        supervisor,
        drain_timeout,
        || stop_accepting(listener, websocket_listener),
        (
            &mut crate::server::runtime::runtime_links::ConfiguredLinks::default(),
            finalize,
        ),
    )
}

pub(in crate::server) fn stop_accepting(
    listener: &mut ServerListener,
    websocket_listener: Option<&mut WebSocketListener>,
) -> Result<(), ServerError> {
    let websocket_result = websocket_listener.map_or(Ok(()), WebSocketListener::stop_accepting);
    let wire_result = listener.stop_accepting();
    match (websocket_result, wire_result) {
        (Err(first), Err(error)) => {
            tracing::error!(%error, "wire listeners also failed to stop accepting");
            Err(first)
        }
        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
        (Ok(()), Ok(())) => Ok(()),
    }
}

/// Waits for the closed connections to deliver their exits.
///
/// Parks on host removal and native handler disposal, which follows removal
/// from the scheduler's process table, and returns when 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)]
#[path = "shutdown_tests.rs"]
mod tests;

#[path = "shutdown_sequence.rs"]
pub(in crate::server) mod sequence;