liminal-server 0.24.9

#[cfg(any(test, feature = "test-support"))]
use super::LiminalConnectionServices;
#[cfg(feature = "test-support")]
use super::install_lys;
use super::{
    AdmissionReadiness, Arc, ConnectionNotifier, ConnectionServices, ConnectionSupervisor,
    ConnectionSupervisorBuilder, LimitsConfig, LysInstall, PassVerifier, ServerConfig, ServerError,
    ServerSignals, ShutdownHandle, SupervisorInner, build_connection_services,
    configured_authentication, configured_lys,
};
#[cfg(test)]
use super::{SubsystemFactory, build_connection_services_via};

impl ConnectionSupervisor {
    pub(in crate::server) fn install_signals(
        &self,
        signals: ServerSignals,
    ) -> Result<(), ServerError> {
        self.inner
            .runtime
            .signals
            .set(signals)
            .map_err(|_| ServerError::ListenerAccept {
                message: "connection signals were already installed".to_owned(),
            })
    }

    #[cfg(test)]
    pub(in crate::server) fn signals_installed(&self) -> bool {
        self.inner.runtime.signals.get().is_some()
    }

    pub(crate) fn current_server_incarnation(&self) -> Result<Option<u64>, ServerError> {
        self.inner
            .incarnations
            .as_ref()
            .map(|authority| authority.current_server_incarnation())
            .transpose()
    }

    /// Creates a connection supervisor backed by the services the config's
    /// `[services]` profile selects: the full liminal channel/conversation stack
    /// (the default) or the capability-scoped worker front door. Profile
    /// enforcement is [`build_connection_services`](super::services::build_connection_services)'s,
    /// so this constructor can never build full services for a worker-front-door
    /// config.
    ///
    /// # Errors
    /// Returns [`ServerError`] when service construction or scheduler startup fails.
    pub fn from_config(config: &ServerConfig) -> Result<Self, ServerError> {
        Self::from_config_with_services(config, build_connection_services(config)?)
    }

    /// Installs a fixture's verified Lys authority and machine before binding.
    ///
    /// # Errors
    /// A Lys authority or machine was already installed on this supervisor.
    #[cfg(feature = "test-support")]
    pub fn install_lys_for_tests(
        &self,
        admitter: Arc<dyn crate::auth_lys::session::LysAdmitter>,
        network_id: String,
        link_host: Arc<crate::link::LinkHost>,
    ) -> Result<(), ServerError> {
        install_lys(
            &self.inner.runtime,
            LysInstall {
                admitter,
                network_id,
                link_host: Some(link_host),
                services: None,
                membership: None,
            },
        )
    }

    /// The installed host used by the connection routes.
    #[cfg(feature = "test-support")]
    #[must_use]
    pub fn link_host_for_tests(&self) -> Option<&Arc<crate::link::LinkHost>> {
        self.inner.runtime.link_host()
    }

    /// The installed admission seam used by the connection routes.
    #[cfg(feature = "test-support")]
    #[must_use]
    pub fn lys_admitter_for_tests(
        &self,
    ) -> Option<&Arc<dyn crate::auth_lys::session::LysAdmitter>> {
        self.inner
            .runtime
            .lys_admitter()
            .map(|(admitter, _)| admitter)
    }

    /// [`Self::from_config`] with the §9 D2 subsystem factory injected.
    ///
    /// The factory is the only route to every scheduler-owning subsystem the
    /// services construction builds, so a recording factory observes exactly what
    /// was constructed; the connection scheduler itself (built below for BOTH
    /// profiles) is the census baseline, not a census entry.
    #[cfg(test)]
    pub(super) fn from_config_via(
        config: &ServerConfig,
        subsystems: &dyn SubsystemFactory,
    ) -> Result<Self, ServerError> {
        Self::from_config_with_services(config, build_connection_services_via(config, subsystems)?)
    }

    fn from_config_with_services(
        config: &ServerConfig,
        services: Arc<dyn ConnectionServices>,
    ) -> Result<Self, ServerError> {
        // Absent `[auth]` leaves both `None`, so the connection stays open-access.
        let authentication = config
            .auth
            .as_ref()
            .map(configured_authentication)
            .transpose()?;
        let (auth_token, pass_verifier) =
            authentication.map_or((None, None), |(token, verifier)| (Some(token), verifier));
        SupervisorInner::new(
            services,
            None,
            auth_token,
            pass_verifier,
            config
                .auth
                .as_ref()
                .map(|auth| configured_lys(auth, tokio::runtime::Handle::try_current().ok()))
                .transpose()?
                .flatten(),
            config.limits,
            None,
        )
        .map(|inner| Self {
            inner: Arc::new(inner),
        })
    }

    /// Starts composing a connection supervisor over services the caller built
    /// and keeps: the embedder's constructor.
    ///
    /// Every other public constructor fixes one of the three optional parts —
    /// [`Self::from_config`] installs the configured `[auth]` (bearer AND pass
    /// verifier) but builds its own services, [`Self::with_services_and_auth`]
    /// carries a bearer and never a pass verifier, and
    /// [`Self::with_services_and_notifier`] carries a notifier and no auth at
    /// all. An application that embeds liminal — holding the
    /// `Arc<LiminalConnectionServices>` to publish in-process, verifying
    /// registry passes at Connect, and observing connection lifecycle through a
    /// notifier — needs all three at once, and this is the one route to them.
    /// Nothing set on the builder defaults exactly as the non-config
    /// constructors default it: open access, no notifier, signed default
    /// limits.
    #[must_use]
    pub fn builder(services: Arc<dyn ConnectionServices>) -> ConnectionSupervisorBuilder {
        ConnectionSupervisorBuilder {
            services,
            notifier: None,
            auth_token: None,
            pass_verifier: None,
            lys: None,
            limits: LimitsConfig::default(),
        }
    }

    /// Creates a connection supervisor with no configured channels.
    ///
    /// # Errors
    /// Returns [`ServerError`] when scheduler startup fails.
    #[cfg(any(test, feature = "test-support"))]
    pub fn new() -> Result<Self, ServerError> {
        Self::with_services(Arc::new(LiminalConnectionServices::empty()?))
    }

    /// Creates a connection supervisor using an explicit service adapter.
    ///
    /// # Errors
    /// Returns [`ServerError`] when scheduler startup fails.
    pub fn with_services(services: Arc<dyn ConnectionServices>) -> Result<Self, ServerError> {
        SupervisorInner::new(
            services,
            None,
            None,
            None,
            None,
            LimitsConfig::default(),
            None,
        )
        .map(|inner| Self {
            inner: Arc::new(inner),
        })
    }

    /// Creates a connection supervisor with an explicit service adapter and the
    /// configured connection auth token.
    ///
    /// This is the production constructor for callers that build services
    /// themselves (the runtime needs the shared channel cluster before the
    /// supervisor takes ownership) and therefore cannot use
    /// [`Self::from_config`]: without it the configured `[auth]` token would be
    /// silently dropped and the server would run open-access.
    ///
    /// # Errors
    /// Returns [`ServerError`] when scheduler startup fails.
    pub fn with_services_and_auth(
        services: Arc<dyn ConnectionServices>,
        auth_token: Option<Vec<u8>>,
    ) -> Result<Self, ServerError> {
        Self::with_services_auth_and_limits(services, auth_token, LimitsConfig::default())
    }

    /// Creates a connection supervisor with explicit services, authentication,
    /// and operational limits.
    ///
    /// Production runtime construction uses this form so the durable
    /// incarnation stream's complete-reference bound is the same signed
    /// `max_connections` bound enforced by connection admission.
    ///
    /// # Errors
    /// Returns [`ServerError`] when incarnation startup or scheduler startup fails.
    pub fn with_services_auth_and_limits(
        services: Arc<dyn ConnectionServices>,
        auth_token: Option<Vec<u8>>,
        limits: LimitsConfig,
    ) -> Result<Self, ServerError> {
        Self::with_services_auth_limits_and_fatal_shutdown(services, auth_token, limits, None)
    }

    /// Production composition with the process-wide shutdown activation that a
    /// post-Open participant fatal must join, without a Lys install: the
    /// fixtures' constructor.
    #[cfg(test)]
    pub(crate) fn with_fatal_shutdown(
        services: Arc<dyn ConnectionServices>,
        auth_token: Option<Vec<u8>>,
        pass_verifier: Option<PassVerifier>,
        limits: LimitsConfig,
        fatal_shutdown: ShutdownHandle,
    ) -> Result<Self, ServerError> {
        Self::with_lys_and_fatal_shutdown(
            services,
            auth_token,
            pass_verifier,
            None,
            limits,
            fatal_shutdown,
        )
    }

    /// The production constructor: `[auth]` whole, the Lys install
    /// (`[auth.lys]`: admitter, link host, service bindings) included, with
    /// the fatal shutdown handle. The server's `run` builds through here so
    /// Lys reaches every connection it admits (ACCESS-003 R1 FOUND: the
    /// path without `lys` installed none in production).
    pub(crate) fn with_lys_and_fatal_shutdown(
        services: Arc<dyn ConnectionServices>,
        auth_token: Option<Vec<u8>>,
        pass_verifier: Option<PassVerifier>,
        lys: Option<LysInstall>,
        limits: LimitsConfig,
        fatal_shutdown: ShutdownHandle,
    ) -> Result<Self, ServerError> {
        SupervisorInner::new(
            services,
            None,
            auth_token,
            pass_verifier,
            lys,
            limits,
            Some(fatal_shutdown),
        )
        .map(|inner| Self {
            inner: Arc::new(inner),
        })
    }

    fn with_services_auth_limits_and_fatal_shutdown(
        services: Arc<dyn ConnectionServices>,
        auth_token: Option<Vec<u8>>,
        limits: LimitsConfig,
        fatal_shutdown: Option<ShutdownHandle>,
    ) -> Result<Self, ServerError> {
        SupervisorInner::new(
            services,
            None,
            auth_token,
            None,
            None,
            limits,
            fatal_shutdown,
        )
        .map(|inner| Self {
            inner: Arc::new(inner),
        })
    }

    /// Creates a connection supervisor with an explicit service adapter and a
    /// connection-keyed worker-registration notifier.
    ///
    /// The `notifier` is invoked when a worker registers on a connection and when
    /// such a connection closes, and when a pass-stamped connection attaches
    /// and closes. Supervisors built via [`Self::with_services`],
    /// [`Self::from_config`], or [`Self::new`] carry no notifier, so liminal still
    /// runs standalone; a `WorkerRegister` frame is then accepted without any
    /// application callback.
    ///
    /// # Errors
    /// Returns [`ServerError`] when scheduler startup fails.
    pub fn with_services_and_notifier(
        services: Arc<dyn ConnectionServices>,
        notifier: Arc<dyn ConnectionNotifier>,
    ) -> Result<Self, ServerError> {
        SupervisorInner::new(
            services,
            Some(notifier),
            None,
            None,
            None,
            LimitsConfig::default(),
            None,
        )
        .map(|inner| Self {
            inner: Arc::new(inner),
        })
    }

    /// A handle on whether this server can admit a connection, for the
    /// readiness probe (P0 #56 R4).
    ///
    /// Handed to `SharedReadinessState::track_admission` once the supervisor
    /// exists. The health endpoint binds BEFORE the supervisor is built —
    /// liveness has to be answerable while the rest of the server is still
    /// coming up — so this cannot be wired at readiness construction and is
    /// installed afterwards instead.
    #[must_use]
    pub fn admission_readiness(&self) -> AdmissionReadiness {
        self.inner.admission_readiness.clone()
    }

    /// The channel membership gate `[auth.lys.membership]` installed
    /// (ACCESS-006 R2), for the ACCESS-005 grant owner: it installs itself
    /// with [`crate::auth_membership::MembershipGate::install_authority`] and
    /// drives each applied change and resume into the gate through
    /// [`crate::auth_membership::MembershipListener`]. `None` without the
    /// section.
    #[must_use]
    pub fn membership_gate(&self) -> Option<Arc<crate::auth_membership::MembershipGate>> {
        self.inner.runtime.membership_gate().cloned()
    }
}