arcature 2026.2.1

//! The [`Lifecycle`] coordinator — the request-owned handle an application
//! uses to observe and drive the process lifecycle (AP2.1-10).
//!
//! [`Lifecycle`] is a cheap, clonable handle (one `Arc` increment) sharing
//! the atomic state, the readiness checks, and the drain hooks. The engine
//! creates one in `serve_with_health` / `run_with_health` and passes a
//! reference to the application's `state_fn`; the application clones it into
//! `AppState` so its health routes (or the framework's
//! [`crate::health::router`]) can read `live()` / `ready()` on every request.
//!
//! # No global state (AGENTS.md §20)
//!
//! [`Lifecycle`] is never a process-global singleton. One is created per
//! `serve_with_health` invocation and moves with the owned `Application`. A
//! test that spawns multiple servers has multiple independent lifecycles.
//! Async tasks move between runtime threads; the `Arc` shares cleanly.
//!
//! # State machine
//!
//! See [`state::LifecycleState`] for the transition diagram. The methods
//! below are the only way the state moves; they enforce the legal
//! transitions and are no-ops (or return `false`) for illegal ones, so a
//! double-shutdown or a shutdown-during-startup is safe.

use std::sync::Arc;

use super::hook::{DrainError, DrainHook, ShutdownHooks};
use super::ready::ReadinessChecks;
use super::state::{LifecycleState, StateCell};

/// A cheap, clonable handle to the application lifecycle.
///
/// Cloning shares the underlying state, readiness checks, and drain hooks
/// (one `Arc`). The handle is `Send + Sync` and moves freely across async
/// tasks; it is never a global (AGENTS.md §20).
#[derive(Clone)]
pub struct Lifecycle {
    inner: Arc<Inner>,
}

struct Inner {
    state: StateCell,
    readiness: ReadinessChecks,
    hooks: ShutdownHooks,
}

impl Lifecycle {
    /// Create a new lifecycle in the [`LifecycleState::Starting`] state,
    /// with no readiness checks and no drain hooks.
    #[must_use]
    pub fn new() -> Self {
        Lifecycle {
            inner: Arc::new(Inner {
                state: StateCell::new(),
                readiness: ReadinessChecks::default(),
                hooks: ShutdownHooks::default(),
            }),
        }
    }

    /// The current lifecycle state. Lock-free atomic read (the hot path for
    /// health endpoints).
    #[must_use]
    pub fn state(&self) -> LifecycleState {
        self.inner.state.load()
    }

    /// Whether the process is live (up). True for every state except
    /// `Stopped` — a starting or draining process is still up and may be
    /// serving in-flight requests. Independent of any external dependency
    /// (PROGRAM.md AP2.1-10: "liveness independent of external DB").
    #[must_use]
    pub fn live(&self) -> bool {
        self.state().is_live()
    }

    /// Whether the process is ready to serve new traffic. True **iff** the
    /// state is [`LifecycleState::Ready`] **and** every registered readiness
    /// check returns `true`. This is the gate the `/up/ready` endpoint uses;
    /// it is NOT spoofable — a `Draining` or `Starting` state returns
    /// `false` regardless of the readiness checks (PROGRAM.md AP2.1-10:
    /// "Readiness must NOT become true before required app dependencies are
    /// ready").
    #[must_use]
    pub fn ready(&self) -> bool {
        self.state() == LifecycleState::Ready && self.inner.readiness.all_pass()
    }

    /// Register a readiness check. Returns `self` for chaining. Called from
    /// the application's `state_fn` during startup; the check gates
    /// `/up/ready` once the engine is `Ready`. See [`Self::ready`] for the
    /// non-blocking contract.
    #[must_use]
    pub fn register_readiness<F>(self, check: F) -> Self
    where
        F: Fn() -> bool + Send + Sync + 'static,
    {
        self.inner.readiness.register(std::sync::Arc::new(check));
        self
    }

    /// Register a drain hook. Returns `self` for chaining. Called from the
    /// application's `state_fn` during startup; the engine invokes the hook
    /// during shutdown. See [`Self::register_drain_hook`] for the ordering contract.
    #[must_use]
    pub fn register_drain_hook<H>(self, hook: H) -> Self
    where
        H: DrainHook + 'static,
    {
        self.inner.hooks.register(std::sync::Arc::new(hook));
        self
    }

    /// Transition `Starting → Ready`. No-op if the state is not `Starting`
    /// (a shutdown-during-startup may have already moved to `Draining`).
    /// Called by the engine after startup completes and the state is built.
    pub fn mark_ready(&self) {
        let _ = self
            .inner
            .state
            .compare_exchange(LifecycleState::Starting, LifecycleState::Ready);
    }

    /// Transition to `Draining` from any non-`Stopped` state. Idempotent:
    /// calling twice is safe. Readiness goes false immediately (the state is
    /// no longer `Ready`). Called by the engine on a termination signal,
    /// **before** the listener stops accepting (PROGRAM.md AP2.1-10: "move
    /// readiness false first, stop new intake").
    pub fn begin_drain(&self) {
        // From Starting or Ready, move to Draining. From Draining, stay. From
        // Stopped, stay (terminal). A CAS loop handles the Starting/Ready
        // cases uniformly.
        loop {
            let current = self.inner.state.load();
            match current {
                LifecycleState::Stopped | LifecycleState::Draining => return,
                LifecycleState::Starting | LifecycleState::Ready => {
                    if self
                        .inner
                        .state
                        .compare_exchange(current, LifecycleState::Draining)
                        .is_ok()
                    {
                        return;
                    }
                    // CAS failed — state changed under us; retry.
                }
            }
        }
    }

    /// Transition to `Stopped`. Idempotent. Called by the engine after drain
    /// and resource shutdown complete.
    pub fn mark_stopped(&self) {
        self.inner.state.store(LifecycleState::Stopped);
    }

    /// Run every registered drain hook concurrently via the certified Tokio
    /// runtime and await all. Called by the engine's shutdown orchestration
    /// after the HTTP drain. Returns the collected hook errors (empty on
    /// success). Gated by the `macros` feature (uses `tokio::spawn`); a
    /// caller without `macros` drives hooks via
    /// [`run_drain_hooks_sequential`](Self::run_drain_hooks_sequential).
    #[cfg(feature = "macros")]
    pub async fn run_drain_hooks(&self) -> Vec<DrainError> {
        self.inner.hooks.run().await
    }

    /// Run every registered drain hook sequentially (one after another),
    /// collecting errors. Runtime-agnostic (no `tokio::spawn`); available
    /// with no feature flags. A caller on a custom runtime, or one that
    /// wants ordered drain, uses this instead of
    /// `run_drain_hooks`.
    pub async fn run_drain_hooks_sequential(&self) -> Vec<DrainError> {
        self.inner.hooks.run_sequential().await
    }
}

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

impl std::fmt::Debug for Lifecycle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Lifecycle")
            .field("state", &self.state())
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    use super::Lifecycle;
    use super::LifecycleState;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicBool, Ordering};

    #[test]
    fn new_lifecycle_starts_in_starting_not_ready() {
        let lifecycle = Lifecycle::new();
        assert_eq!(lifecycle.state(), LifecycleState::Starting);
        assert!(lifecycle.live());
        // Not ready: state is Starting, not Ready.
        assert!(!lifecycle.ready());
    }

    #[test]
    fn mark_ready_transitions_to_ready() {
        let lifecycle = Lifecycle::new();
        lifecycle.mark_ready();
        assert_eq!(lifecycle.state(), LifecycleState::Ready);
        assert!(lifecycle.live());
        // No readiness checks → ready() true once state is Ready.
        assert!(lifecycle.ready());
    }

    #[test]
    fn readiness_check_gates_ready() {
        let flag = Arc::new(AtomicBool::new(false));
        let flag_for_check = flag.clone();
        let lifecycle =
            Lifecycle::new().register_readiness(move || flag_for_check.load(Ordering::SeqCst));
        lifecycle.mark_ready();
        // State is Ready but the check is false → not ready.
        assert!(!lifecycle.ready());
        flag.store(true, Ordering::SeqCst);
        assert!(lifecycle.ready());
    }

    #[test]
    fn begin_drain_makes_not_ready_immediately() {
        let lifecycle = Lifecycle::new();
        lifecycle.mark_ready();
        assert!(lifecycle.ready());
        lifecycle.begin_drain();
        assert_eq!(lifecycle.state(), LifecycleState::Draining);
        // Draining is still live (in-flight may be served) but not ready.
        assert!(lifecycle.live());
        assert!(!lifecycle.ready());
    }

    #[test]
    fn begin_drain_during_startup_is_safe() {
        let lifecycle = Lifecycle::new();
        // Drain before ready (shutdown during startup).
        lifecycle.begin_drain();
        assert_eq!(lifecycle.state(), LifecycleState::Draining);
        // A late mark_ready is a no-op (state is already Draining).
        lifecycle.mark_ready();
        assert_eq!(lifecycle.state(), LifecycleState::Draining);
        assert!(!lifecycle.ready());
    }

    #[test]
    fn mark_stopped_is_terminal_and_idempotent() {
        let lifecycle = Lifecycle::new();
        lifecycle.mark_ready();
        lifecycle.begin_drain();
        lifecycle.mark_stopped();
        assert_eq!(lifecycle.state(), LifecycleState::Stopped);
        assert!(!lifecycle.live());
        assert!(!lifecycle.ready());
        // Idempotent.
        lifecycle.mark_stopped();
        assert_eq!(lifecycle.state(), LifecycleState::Stopped);
    }

    #[tokio::test]
    #[cfg(feature = "macros")]
    async fn drain_hooks_run_and_report_errors() {
        use std::sync::atomic::AtomicUsize;
        use std::time::Duration;

        use super::DrainHook;

        struct CountingHook {
            name: &'static str,
            counter: Arc<AtomicUsize>,
            fail: bool,
        }

        impl DrainHook for CountingHook {
            fn name(&self) -> &'static str {
                self.name
            }
            fn drain(
                &self,
            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<(), String>> + Send + '_>>
            {
                let counter = self.counter.clone();
                let fail = self.fail;
                Box::pin(async move {
                    tokio::time::sleep(Duration::from_millis(5)).await;
                    counter.fetch_add(1, Ordering::SeqCst);
                    if fail {
                        Err("simulated drain failure".to_owned())
                    } else {
                        Ok(())
                    }
                })
            }
        }

        let counter = Arc::new(AtomicUsize::new(0));
        let lifecycle = Lifecycle::new()
            .register_drain_hook(CountingHook {
                name: "ok-hook",
                counter: counter.clone(),
                fail: false,
            })
            .register_drain_hook(CountingHook {
                name: "fail-hook",
                counter: counter.clone(),
                fail: true,
            });

        let errors = lifecycle.run_drain_hooks().await;
        // Both hooks ran (concurrently) despite the failure.
        assert_eq!(counter.load(Ordering::SeqCst), 2);
        // The failing hook's error is surfaced.
        assert_eq!(errors.len(), 1);
        assert!(matches!(
            errors[0],
            super::DrainError::Hook {
                name: "fail-hook",
                ..
            }
        ));
    }

    #[test]
    fn clone_shares_state() {
        let lifecycle = Lifecycle::new();
        let clone = lifecycle.clone();
        lifecycle.mark_ready();
        // The clone observes the same state (shared Arc).
        assert_eq!(clone.state(), LifecycleState::Ready);
        assert!(clone.ready());
    }
}