liminal-server 0.14.3

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.
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) {
        if self.is_initiated() {
            return;
        }
        let Ok(mut guard) = self.inner.wait_lock.lock() else {
            return;
        };
        while !self.is_initiated() {
            match self.inner.waiter.wait(guard) {
                Ok(next_guard) => guard = next_guard,
                Err(_) => return,
            }
        }
    }

    /// 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,
}

impl ShutdownState {
    const fn new() -> Self {
        Self {
            initiated: AtomicBool::new(false),
            wait_lock: Mutex::new(()),
            waiter: Condvar::new(),
        }
    }

    fn notify(&self) {
        if let Ok(_guard) = self.wait_lock.lock() {
            self.waiter.notify_all();
        }
    }
}

/// 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::debug!("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.
///
/// Four steps, in this order, with no timer between them: stop accepting, close
/// every connection this server holds, 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.
pub fn run_shutdown_sequence(
    listener: &mut ServerListener,
    websocket_listener: Option<&mut WebSocketListener>,
    supervisor: &ConnectionSupervisor,
    drain_timeout: Duration,
) -> 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 did not quiesce 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);

    flush_durable_state(supervisor)?;
    supervisor.shutdown();
    tracing::info!(
        elapsed_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX),
        "shutdown sequence complete"
    );
    Ok(())
}

/// 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)]
mod tests {
    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());
    }

    #[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()
        });

        thread::sleep(Duration::from_millis(10));
        assert!(handle.initiate());
        let observed = worker.join().map_err(|_| "wait worker panicked")?;

        assert!(observed);
        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"
        );
    }
}