liminal-server 0.23.1

//! Auxiliary runtime owners use the same services and shutdown precedence.
use crate::ServerError;
use crate::cluster::{self, ClusterHandle};
use crate::config::types::ClusterConfig;
use crate::health::SharedReadinessState;
use crate::server::ShutdownHandle;
use crate::server::connection::{
    ConnectionSupervisor,
    services::{ChannelCluster, LiminalConnectionServices},
};
use crate::server::self_streams::{
    cost::ServerMetadata, owner::PublisherRuntime, sink::SelfStreams,
};
use std::sync::Arc;
pub(in crate::server) fn start_self_stream_publisher(
    services: &Arc<LiminalConnectionServices>,
    supervisor: &ConnectionSupervisor,
    config: &crate::config::types::ServerConfig,
    shutdown: &ShutdownHandle,
) -> Result<Option<PublisherRuntime>, ServerError> {
    bind_self_streams(services, supervisor, config, shutdown)?
        .map(|streams| {
            PublisherRuntime::start(services.signals().wake(), streams, shutdown.clone())
        })
        .transpose()
}

pub(super) fn finish_self_stream_publisher(
    publisher: &mut Option<PublisherRuntime>,
) -> Result<(), ServerError> {
    publisher.as_mut().map_or(Ok(()), PublisherRuntime::finish)
}

pub(super) fn start_configured_cluster(
    channels: &ChannelCluster,
    config: Option<&ClusterConfig>,
    readiness: &SharedReadinessState,
) -> Result<Option<ClusterHandle>, ServerError> {
    readiness.set_cluster_configured(config.is_some());
    config
        .map(|config| start_cluster(channels, config, readiness))
        .transpose()
}

/// Binds the shared signals and incarnation-owned subjects before cluster startup.
pub(in crate::server) fn bind_self_streams(
    services: &Arc<LiminalConnectionServices>,
    supervisor: &ConnectionSupervisor,
    config: &crate::config::types::ServerConfig,
    shutdown: &ShutdownHandle,
) -> Result<Option<Arc<SelfStreams>>, ServerError> {
    let signals = services.signals();
    supervisor.install_signals(signals.clone())?;
    let Some(incarnation) = supervisor.current_server_incarnation()? else {
        crate::metrics::self_streams_disabled();
        tracing::warn!(
            "SELF_STREAMS_REQUIRE_PARTICIPANT: self streams disabled because no participant is configured"
        );
        // SRV-012 R4: the console streams ride the same publisher owner, so
        // without one no console subject is registered and a console
        // subscribe answers CHANNEL_NOT_REGISTERED.
        crate::metrics::console_streams_disabled();
        tracing::warn!(
            "CONSOLE_STREAMS_REQUIRE_PARTICIPANT: console streams disabled because no participant is configured"
        );
        return Ok(None);
    };
    services.install_self_stream_channels()?;
    // SRV-012 R7: this device's uptime runs from the incarnation now bound.
    services
        .console()
        .incarnation_started(std::time::Instant::now())
        .map_err(|error| ServerError::ListenerAccept {
            message: format!("console hub: {error}"),
        })?;
    let metadata = ServerMetadata::from_config(config, incarnation).map_err(|error| {
        ServerError::ConfigValidation {
            message: error.to_string(),
        }
    })?;
    let streams = Arc::new(SelfStreams::new(
        metadata,
        services,
        signals.wake(),
        shutdown.clone(),
        config.services.self_streams,
    ));
    signals.install(Arc::clone(&streams))?;
    services.install_console_channels()?;
    signals.enable_console()?;
    let wake = signals.wake();
    wake.install_source(&streams)?;
    let counts = services.store_counts();
    let names = ["durable", "ephemeral-dedup"];
    let configured = names
        .get(..counts.store_count())
        .ok_or_else(|| SelfStreams::failure("unexpected number of counted stores"))?;
    counts.install_observer(configured, wake).map_err(|error| {
        SelfStreams::failure(&format!("store count observer installation: {error}"))
    })?;
    Ok(Some(streams))
}

/// Raises the health endpoint's shutdown error ahead of the shutdown
/// sequence's, a precedence chosen deliberately: a health-shutdown failure takes
/// the exit code. The sequence error it masks is logged rather than dropped, so
/// only its exit code goes missing.
pub(super) fn raise_in_precedence(
    health_shutdown: Result<(), ServerError>,
    shutdown_sequence: Result<(), ServerError>,
) -> Result<(), ServerError> {
    if let Err(error) = health_shutdown {
        if let Err(sequence_error) = &shutdown_sequence {
            tracing::error!(
                %sequence_error,
                "shutdown sequence failed; the health endpoint shutdown failure takes the exit code"
            );
        }
        return Err(error);
    }
    shutdown_sequence
}

/// Starts clustering on the shared channel supervisor's scheduler (SRV-005).
///
/// Installs the cluster `sync` as the supervisor's [`ClusterObserver`] so channel
/// subscribe/unsubscribe/publish events drive process-group membership and
/// cross-node fan-out.
///
/// On the success path this marks cluster membership as established on `readiness`
/// (G2) via [`cluster::start`]'s `on_established` hook, so a clustered server's
/// `/ready` endpoint transitions from 503 to 200 once the cluster stack is up.
/// Every early return here (missing resolver, listener bind failure, no reachable
/// seed) leaves the flag unset, so `/ready` stays 503.
pub(super) fn start_cluster(
    channel_cluster: &ChannelCluster,
    cluster_config: &ClusterConfig,
    readiness: &SharedReadinessState,
) -> Result<ClusterHandle, ServerError> {
    let resolver = channel_cluster
        .resolver()
        .cloned()
        .ok_or_else(|| ServerError::ClusterJoin {
            message: "clustering configured but channel supervisor has no distribution resolver"
                .to_owned(),
        })?;
    let scheduler = channel_cluster.supervisor().scheduler();
    let observer = Arc::clone(channel_cluster.observer());
    let readiness = readiness.clone();
    cluster::start(
        &scheduler,
        resolver,
        cluster_config,
        move |sync| {
            // Peers become the console topology's devices and links (R7),
            // set before the initial view is applied.
            let peers = Arc::clone(&observer);
            if !sync.set_peer_listener(Arc::new(move |name, present| peers.peer(name, present))) {
                tracing::error!("cluster peer listener was already installed");
            }
            // The supervisor's one observer forwards to the cluster from now
            // on (SRV-012 R6); it was installed when the supervisor was built.
            if let Err(error) = observer.set_cluster(Arc::new(sync)) {
                tracing::error!(%error, "cluster observer was not installed");
            }
        },
        move || readiness.set_cluster_membership_established(true),
    )
}

pub(super) fn install_ephemeral_sweep_report(
    services: &LiminalConnectionServices,
    health: &crate::health::HealthServerHandle,
) -> Result<(), ServerError> {
    let report = services
        .ephemeral_sweep_counts()
        .ok_or_else(|| ServerError::HealthEndpoint {
            message: "full services have no ephemeral root startup sweep report".to_owned(),
        })?;
    health.install_ephemeral_sweep_report(report)
}