liminal_server/server/connection/supervisor/

supervisor_shutdown.rs

use super::{
    ConnectionControl, ConnectionSupervisor, Instant, ParticipantServiceFatal, ServerError,
};

impl ConnectionSupervisor {
    /// How many connections have departed this supervisor: the count of
    /// connection records removed, raised by the one removal funnel every exit
    /// route reaches. A teardown that completes its connection's fate does so
    /// before the record is removed, so a departure counted here is one whose
    /// fate has already completed. Pass it to [`Self::await_departures`].
    ///
    /// # Errors
    /// Returns [`ServerError`] when the departure count's lock is poisoned.
    pub fn departures(&self) -> Result<u64, ServerError> {
        self.inner.runtime.departures()
    }

    /// Parks until `count` connections have departed since `since`, a value
    /// [`Self::departures`] returned, or until `deadline` elapses. Returns
    /// `true` when they have departed and `false` when the deadline won.
    ///
    /// Woken by each departure through the removal `Condvar`, never by a timer
    /// or a count scan; the deadline is the one bound on the park.
    ///
    /// # Errors
    /// Returns [`ServerError`] when the departure count's lock is poisoned.
    pub fn await_departures(
        &self,
        since: u64,
        count: u64,
        deadline: Instant,
    ) -> Result<bool, ServerError> {
        self.inner.runtime.await_departures(since, count, deadline)
    }

    /// Parks until every tracked connection has been removed or `deadline`
    /// elapses, returning `true` when the drain completed and `false` when the
    /// single admitted deadline won.
    ///
    /// The TOLD drain-completion replacement (W4 leg 3, §4.3): the waiter is woken
    /// by the one `remove()` funnel every connection exit reaches — never by a
    /// reap/count timer. Both the graceful drain and the force-close settle call
    /// this same waiter, each with its own one-shot deadline; there is no second
    /// settle poll loop.
    #[must_use]
    pub(crate) fn wait_for_connections_drained(&self, deadline: Instant) -> bool {
        let drained = self
            .inner
            .runtime
            .wait_for_active_connections_drained(deadline);
        let retired = self.inner.retirements.wait(deadline);
        if !retired {
            tracing::warn!(
                "force-closed native connection retirement was not observed before the wedged-connection stop"
            );
        }
        drained && retired
    }

    /// FIX A-ii shutdown flush barrier: parks until every active connection has
    /// fanned out its accepted publishes to its socket, or `deadline` elapses.
    /// Called in `run_shutdown_sequence` BEFORE the shutdown Disconnect broadcast
    /// so an accepted-but-unfanned-out publish can no longer be overtaken by it.
    #[must_use]
    pub(crate) fn wait_for_delivery_quiesced(&self, deadline: Instant) -> bool {
        self.inner.runtime.wait_for_delivery_quiesced(deadline)
    }

    /// Returns the first latched post-Open participant fatal, if any.
    ///
    /// The production runtime reads this after its existing shutdown handle wakes,
    /// then returns the typed fatal after the ordinary drain and durable flush.
    pub(crate) fn participant_service_fatal(
        &self,
    ) -> Result<Option<ParticipantServiceFatal>, ServerError> {
        self.inner.runtime.participant_service_fatal()
    }

    /// Broadcasts a best-effort shutdown notification to active connections.
    ///
    /// Connections with no active subscriptions ignore the notification. Failures
    /// to enqueue the control message are logged and skipped; they are not retried.
    pub fn notify_shutdown_subscribers(&self) {
        self.inner
            .broadcast_control(&ConnectionControl::NotifyShutdown);
    }

    /// Sends the orderly close control message to every tracked connection
    /// process.
    ///
    /// Each live process enqueues one 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
    /// normally. A peer reads a shutdown, not an error. Enqueue failures are
    /// logged and skipped.
    ///
    /// Since 0.14.3 this is the ONLY close route in `run_shutdown_sequence`:
    /// nothing waits for a peer to hang up first, so the log line below is the
    /// ordinary shutdown record rather than a warning about an expired drain.
    pub fn force_close_active_connections(&self) {
        for connection in self.inner.runtime.active_connections() {
            self.inner
                .runtime
                .link_life
                .close(connection.pid, crate::link::LinkEnd::Shutdown);
            tracing::info!(
                connection_pid = connection.pid,
                peer_addr = ?connection.peer_addr,
                "closing connection for server shutdown"
            );
            if !self
                .inner
                .enqueue_control(connection.pid, ConnectionControl::ForceClose)
            {
                // Observation, not diagnosis — see `broadcast_control`.
                tracing::warn!(
                    connection_pid = connection.pid,
                    peer_addr = ?connection.peer_addr,
                    "forceful connection close was not published to the process"
                );
            }
        }
    }

    /// Flushes durable channel state through the configured liminal services.
    ///
    /// # Errors
    /// Returns [`ServerError::ShutdownFlush`] when the underlying service flush fails.
    pub fn flush_durable_state(&self) -> Result<(), ServerError> {
        self.inner.runtime.services().flush_durable_state()
    }

    /// Stops the participant warm and records the clean stop's capacity
    /// snapshot, before the final flush (DUR-020 R3, R5).
    pub(crate) fn stop_participant_warm(&self) {
        self.inner.runtime.services().stop_participant_warm();
    }

    /// Completes, before a clean stop reports success, every connection fate
    /// parked on candidate-lane `Precedence`.
    ///
    /// # Errors
    /// Returns [`ServerError::ShutdownUnsettledConnectionFates`] naming every
    /// durable Open the stop could not complete.
    pub(crate) fn settle_parked_connection_fates(&self) -> Result<(), ServerError> {
        self.inner.runtime.settle_parked_connection_fates()
    }

    /// Stops the beamr scheduler used by connection processes.
    pub fn shutdown(&self) {
        // A connection waiting on a conversation's restore is woken and
        // answered at once, so the scheduler join below never waits on a
        // room's replay or rebuild (Gaia, 11 October 2026: 60 s).
        self.stop_participant_warm();
        // Remove every host record while the readiness owner is still live. The
        // removal path ACKs deregistration and only then releases each fd guard;
        // scheduler shutdown subsequently drops the process-owned handles.
        let removed: Vec<_> = self
            .inner
            .runtime
            .active_connections()
            .into_iter()
            .filter_map(|connection| {
                self.inner
                    .runtime
                    .remove(connection.pid)
                    .map(|record| (connection.pid, record))
            })
            .collect();
        if let Some(notifier) = self.inner.runtime.notifier.as_ref() {
            let detached: Vec<_> = removed
                .iter()
                .filter_map(|(pid, record)| {
                    record
                        .pass_principal
                        .as_ref()
                        .map(|principal| (*pid, principal))
                })
                .collect();
            let unregistered: Vec<_> = removed
                .iter()
                .filter_map(|(pid, record)| record.registration.as_ref().map(|_| *pid))
                .collect();
            notifier.on_shutdown(&detached, &unregistered);
        }
        self.inner.scheduler.shutdown();
    }
}