arcature 2026.2.1

//! Realtime shutdown configuration — the shared signal the WS/SSE endpoints
//! and the connection registry use to drain (PROGRAM.md §AP2.1-8).
//!
//! This is a small, self-contained signal (an `Arc<AtomicBool>` + a
//! `tokio::sync::Notify` + the global connection cap) so the realtime module
//! can drain without touching
//! `crates/arcature/src/application/shutdown.rs` (the Production lane owns
//! the global drain state machine). The application constructs one
//! [`ShutdownConfig`], stores it in `AppState`, flips it to draining from
//! its own shutdown hook, and calls [`crate::realtime::registry::Registry::drain`].
//!
//! # Integration seam (for the master)
//!
//! After the Production lane lands the global lifecycle state machine, the
//! master wires this seam into it: the global drain sets `ShutdownConfig` to
//! draining (one atomic store + `notify_waiters`) and awaits
//! `Registry::drain`. The realtime module needs no edit for that — the seam
//! is the `ShutdownConfig` value and the `Registry` it shares. See the
//! report's integration-seam note.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use tokio::sync::Notify;

/// The shared realtime shutdown signal. Cheap to clone (`Arc`-backed); the
/// same value is held by the WS endpoint, the SSE endpoint, the connection
/// registry, and `AppState`.
#[derive(Clone, Debug)]
pub struct ShutdownConfig {
    inner: Arc<ShutdownInner>,
}

#[derive(Debug)]
struct ShutdownInner {
    draining: AtomicBool,
    /// Notified when `begin_drain` flips the signal, so long-running
    /// connection loops can `select!` against it instead of polling.
    drain_notify: Notify,
    /// The global connection cap the registry enforces. Recorded here so
    /// the connection task can read it without a second parameter.
    max_connections: usize,
}

impl ShutdownConfig {
    /// Construct the shutdown config. `max_connections` is the global cap
    /// across WS + SSE connections; the registry rejects new connections
    /// past it (AGENTS.md §29).
    #[must_use]
    pub fn new(max_connections: usize) -> Self {
        Self {
            inner: Arc::new(ShutdownInner {
                draining: AtomicBool::new(false),
                drain_notify: Notify::new(),
                max_connections,
            }),
        }
    }

    /// Flip the signal to draining. Idempotent. Notifies any connection
    /// loop `select!`ing against [`Self::drain_notified`] so it exits
    /// promptly instead of waiting for the next heartbeat tick. After this,
    /// new broadcast/SSE frames stop being forwarded and the connection
    /// task exits its loop, closing the connection gracefully.
    pub fn begin_drain(&self) {
        self.inner.draining.store(true, Ordering::Relaxed);
        self.inner.drain_notify.notify_waiters();
    }

    /// Whether the server is currently draining.
    #[must_use]
    pub fn is_draining(&self) -> bool {
        self.inner.draining.load(Ordering::Relaxed)
    }

    /// The configured global connection cap.
    #[must_use]
    pub fn max_connections(&self) -> usize {
        self.inner.max_connections
    }

    /// A future that completes when the server begins draining. Connection
    /// loops `select!` against this so a drain exits the loop promptly even
    /// when the loop is blocked on a client `recv` with no incoming message
    /// and the heartbeat interval is long.
    ///
    /// The returned `Future` is itself `#[must_use]`, so the caller is
    /// already warned if it drops the awaitable; no redundant function-level
    /// `#[must_use]` is needed.
    pub fn drain_notified(&self) -> impl std::future::Future<Output = ()> + '_ {
        self.inner.drain_notify.notified()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn drain_signal_is_set_and_idempotent() {
        let s = ShutdownConfig::new(8);
        assert!(!s.is_draining());
        s.begin_drain();
        assert!(s.is_draining());
        // Idempotent: a second call is a no-op.
        s.begin_drain();
        assert!(s.is_draining());
    }

    #[test]
    fn clones_share_the_signal() {
        let s = ShutdownConfig::new(8);
        let s2 = s.clone();
        s.begin_drain();
        assert!(s2.is_draining(), "clones observe the same atomic");
    }

    #[test]
    fn max_connections_is_recorded() {
        let s = ShutdownConfig::new(256);
        assert_eq!(s.max_connections(), 256);
    }

    #[tokio::test]
    async fn drain_notified_completes_after_begin_drain() {
        let s = ShutdownConfig::new(8);
        let s2 = s.clone();
        // A waiter task parks on `drain_notified` before the drain begins.
        let waiter = tokio::spawn(async move {
            s2.drain_notified().await;
        });
        tokio::task::yield_now().await;
        s.begin_drain();
        tokio::time::timeout(std::time::Duration::from_secs(2), waiter)
            .await
            .expect("drain_notified did not hang")
            .expect("waiter task did not panic");
    }
}