arcature 2026.2.0

//! The drain/shutdown hook seam (AP2.1-10).
//!
//! [`DrainHook`] is the trait the Realtime lane (AP2.1-8) and the jobs lane
//! implement to participate in the coordinated graceful drain. The engine
//! calls every registered hook **after** the HTTP listener stops accepting
//! and drains in-flight HTTP connections, and **before** the engine's
//! subsystem resources (database, cache, storage, mail) are closed. The
//! hook is the place to close long-lived connections (WebSockets, SSE
//! streams, active job tasks) so the engine's resource teardown sees no
//! outstanding work.
//!
//! # Ordering contract (precise — for the master wiring AP2.1-8)
//!
//! On a termination signal the [`super::Lifecycle`] orchestration runs:
//!
//! 1. `lifecycle.begin_drain()` — readiness goes false **first** (the
//!    upstream load balancer observing `/up/ready` stops sending traffic).
//! 2. The HTTP listener stops accepting new connections (axum graceful
//!    shutdown is triggered).
//! 3. **Concurrently**, every registered [`DrainHook::drain`] is invoked so
//!    long-lived connections (WebSockets) close and the in-flight HTTP drain
//!    completes. The hooks and the HTTP drain race; the orchestration awaits
//!    both.
//! 4. The engine's [`super::super::shutdown`] tears down subsystem resources
//!    in reverse startup order (mail, storage, cache, worker, db).
//! 5. `lifecycle.mark_stopped()` — the process is about to exit.
//!
//! A hook that needs to close WebSockets should signal them to close in
//! `drain()` and await their closure. The Realtime lane's
//! `realtime::shutdown()` (AP2.1-8, self-contained this wave) is the
//! expected implementor; the master wires it by registering a `DrainHook`
//! that calls `realtime::shutdown().await` inside `drain()`.
//!
//! # Failure semantics
//!
//! A hook returning `Err` is recorded; the orchestration continues draining
//! the remaining hooks and resources — a partial drain is better than
//! leaving connections open. The first hook error is surfaced to the
//! operator alongside the shutdown result. A hook that panics is caught by
//! the orchestration's `JoinHandle` and recorded as a [`DrainError::Hook`];
//! it does not abort the drain (AGENTS.md §17: no panic path aborts
//! production teardown).
//!
//! # Feature gating
//!
//! The [`DrainHook`] trait, [`ShutdownHooks`] registration, and
//! [`DrainError`] are pure `std` — they compile with no feature flags so an
//! expert user on a custom runtime can register hooks and drive their own
//! execution. The concurrent [`ShutdownHooks::run`] executor uses the
//! certified Tokio runtime (`tokio::spawn`) and is therefore gated behind
//! the `macros` feature. Without `macros`, a caller drives hooks via
//! [`ShutdownHooks::iter`] (sequential, runtime-agnostic) or their own
//! executor.

use std::sync::Arc;

/// A typed drain/shutdown hook failure. Preserved (not collapsed to `String`)
/// so the operator sees which hook failed and why (AGENTS.md §18).
#[derive(Debug)]
pub enum DrainError {
    /// A registered hook returned an error. The `name` identifies the hook
    /// for diagnostics (the hook's `name()`); the `message` is the hook's
    /// own description of the failure.
    Hook {
        /// The hook's reported name (never a secret).
        name: &'static str,
        /// The hook's failure description.
        message: String,
    },
    /// A hook's drain task panicked. The panic message is captured for
    /// diagnostics; the drain continues with the remaining hooks.
    Panic {
        /// The hook's reported name.
        name: &'static str,
        /// The captured panic payload as a string (best-effort).
        message: String,
    },
}

impl std::fmt::Display for DrainError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Hook { name, message } => {
                write!(f, "drain hook {name:?} failed: {message}")
            }
            Self::Panic { name, message } => {
                write!(f, "drain hook {name:?} panicked: {message}")
            }
        }
    }
}

impl std::error::Error for DrainError {}

/// The trait a lane implements to participate in the coordinated drain.
///
/// Register an implementor with
/// [`Lifecycle::register_drain_hook`](super::Lifecycle::register_drain_hook)
/// during startup (typically from the `state_fn`). The engine invokes
/// [`DrainHook::drain`] on every registered hook during shutdown, after the
/// HTTP listener stops accepting and in-flight HTTP is draining.
///
/// `drain()` must be idempotent and safe to call once. It should signal
/// long-lived connections to close and await their closure; it must not
/// close resources the engine's subsystem shutdown owns (the database pool,
/// cache socket, etc. — those are closed by the engine after the hooks).
pub trait DrainHook: Send + Sync + 'static {
    /// A short, stable name identifying the hook (e.g. `"realtime"`,
    /// `"jobs"`). Used in diagnostics; never a secret.
    fn name(&self) -> &'static str;

    /// Drain the lane's in-flight work and close long-lived connections.
    /// Called once per hook during shutdown, concurrently with the HTTP
    /// drain. Return `Err` with a description on failure; the orchestration
    /// records it and continues.
    fn drain(
        &self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + '_>>;
}

/// The registry of drain hooks for one [`Lifecycle`](super::Lifecycle).
///
/// Built during startup; invoked once during shutdown. Stored behind an
/// `Arc` so the lifecycle handle and the orchestration share the same list.
#[derive(Default)]
pub(crate) struct ShutdownHooks {
    hooks: std::sync::RwLock<Vec<Arc<dyn DrainHook>>>,
}

impl ShutdownHooks {
    /// Register a drain hook. Called during startup (from `state_fn`).
    /// Pure `std` — available with no feature flags.
    pub(crate) fn register(&self, hook: Arc<dyn DrainHook>) {
        let mut guard = self
            .hooks
            .write()
            .expect("shutdown-hooks lock poisoned: a registration task panicked");
        guard.push(hook);
    }

    /// The registered hooks, cloned out for a caller that drives its own
    /// executor. Available with no feature flags (the concurrent
    /// [`Self::run`] executor is `macros`-gated).
    #[allow(dead_code)]
    pub(crate) fn iter(&self) -> Vec<Arc<dyn DrainHook>> {
        match self.hooks.read() {
            Ok(guard) => guard.iter().cloned().collect(),
            Err(_) => Vec::new(),
        }
    }

    /// Run every registered hook's `drain()` concurrently via the certified
    /// Tokio runtime, awaiting all. Returns the collected errors (empty on
    /// full success). A panicking hook is recorded as a [`DrainError::Panic`]
    /// and does not abort the others (AGENTS.md §17). Gated by `macros`
    /// (`tokio::spawn`); without `macros` a caller drives hooks via
    /// [`Self::iter`].
    #[cfg(feature = "macros")]
    pub(crate) async fn run(&self) -> Vec<DrainError> {
        let hooks = self.iter();
        if hooks.is_empty() {
            return Vec::new();
        }
        // Spawn each hook concurrently and join. The hooks close long-lived
        // connections while the HTTP drain proceeds in parallel.
        let mut joins = Vec::with_capacity(hooks.len());
        for hook in hooks {
            let name = hook.name();
            let join = tokio::spawn(async move { (name, hook.drain().await) });
            joins.push(join);
        }
        let mut errors = Vec::new();
        for join in joins {
            match join.await {
                Ok((_name, Ok(()))) => {}
                Ok((name, Err(message))) => errors.push(DrainError::Hook { name, message }),
                Err(join_err) => {
                    // A hook task panicked. Surface the panic message; the
                    // drain continues. `tokio::task::JoinError` is panic or
                    // cancel — a drain task is never cancelled here (the
                    // orchestration does not cancel it), so this is a panic.
                    let name = "?";
                    let message = join_err.to_string();
                    errors.push(DrainError::Panic { name, message });
                }
            }
        }
        errors
    }

    /// The sequential, runtime-agnostic executor: run each hook's `drain()`
    /// one after another, collecting errors. Used when the `macros` feature
    /// (tokio runtime) is off, or by a caller that wants ordered drain.
    /// Available with no feature flags.
    #[allow(dead_code)]
    pub(crate) async fn run_sequential(&self) -> Vec<DrainError> {
        let hooks = self.iter();
        let mut errors = Vec::new();
        for hook in hooks {
            let name = hook.name();
            match hook.drain().await {
                Ok(()) => {}
                Err(message) => errors.push(DrainError::Hook { name, message }),
            }
        }
        errors
    }
}