liminal-server 0.24.9

//! Starts configured exporting bridges after the listeners bind and stops them before the store flush.
//! A refused dial leaves the named link down and is retried while the server keeps serving.

use crate::ServerError;
use crate::config::ServerConfig;
use crate::config::types::MachineConfig;
use crate::link::LinkHost;
use crate::link_bridge::BridgeRegistry;
use crate::server::connection::LysInstall;
use crate::server::connection::services::LiminalConnectionServices;
use std::sync::Arc;
#[path = "runtime_links/carriers.rs"]
mod carriers;
#[path = "runtime_links/retry.rs"]
mod retry;
use retry::RetryLinks;
#[cfg(feature = "test-support")]
#[path = "runtime_links/test_support.rs"]
pub mod test_support;

#[derive(Debug)]
struct Configuration {
    machine: MachineConfig,
    host: Arc<LinkHost>,
    services: Arc<LiminalConnectionServices>,
}

#[derive(Debug, Default)]
pub(in crate::server) struct ConfiguredLinks {
    configuration: Option<Configuration>,
    retry: Option<RetryLinks>,
}

pub(super) fn prepare(
    config: &ServerConfig,
    install: Option<&LysInstall>,
    services: &Arc<LiminalConnectionServices>,
) -> Result<ConfiguredLinks, ServerError> {
    let Some(machine) = config
        .auth
        .as_ref()
        .and_then(|auth| auth.lys.as_ref())
        .and_then(|lys| lys.machine.as_ref())
        .filter(|machine| {
            !machine.links.is_empty() || machine.carrier_choice.broker_echo.is_some()
        })
    else {
        return Ok(ConfiguredLinks::default());
    };
    let install =
        install.ok_or_else(|| refusal("configured outbound links have no installed Lys"))?;
    let Some(host) = install.link_host.as_ref() else {
        // The pass source names its boot refusal; serving does not require a link host.
        return Ok(ConfiguredLinks::default());
    };
    Ok(ConfiguredLinks {
        configuration: Some(Configuration {
            machine: machine.clone(),
            host: Arc::clone(host),
            services: Arc::clone(services),
        }),
        retry: None,
    })
}

impl ConfiguredLinks {
    pub(super) fn start(
        &mut self,
        registry: &BridgeRegistry,
        supervisor: &crate::server::ConnectionSupervisor,
    ) -> Result<(), ServerError> {
        let Some(configuration) = &self.configuration else {
            return Ok(());
        };
        self.retry = Some(RetryLinks::start(configuration, registry, supervisor)?);
        Ok(())
    }
    pub(super) fn finish(&mut self) -> Result<(), ServerError> {
        self.retry.take().map_or(Ok(()), |mut retry| retry.finish())
    }
}

fn refusal(message: impl Into<String>) -> ServerError {
    ServerError::ConfigValidation {
        message: message.into(),
    }
}

/// The Lys ask runtime and the install for the configured `[auth.lys]`, on
/// that runtime's handle; none without `[auth]`. The machine pass's first
/// issue waits for [`MachineIssue::begin`] when it can (SRV-041 R4).
pub(super) fn lys(
    config: &ServerConfig,
) -> Result<
    (
        Option<tokio::runtime::Runtime>,
        Option<LysInstall>,
        MachineIssue,
    ),
    ServerError,
> {
    let started = std::time::Instant::now();
    let runtime = crate::server::connection::lys_ask_runtime(config.auth.as_ref())?;
    let (install, pending) = config.auth.as_ref().map_or(Ok((None, None)), |auth| {
        crate::server::connection::configured_lys_after_bind(
            auth,
            runtime.as_ref().map(|runtime| runtime.handle().clone()),
        )
    })?;
    crate::server::boot_steps::finished("lys_install", started);
    Ok((
        runtime,
        install,
        MachineIssue {
            pending,
            asking: None,
        },
    ))
}

/// The machine pass's first issue: pending until the listener binds, then
/// asked on its own thread, joined before the Lys ask runtime goes.
#[derive(Debug, Default)]
pub(in crate::server) struct MachineIssue {
    pending: Option<crate::link_pass::StartIssue>,
    asking: Option<std::thread::JoinHandle<()>>,
}

impl MachineIssue {
    pub(super) fn begin(&mut self) -> Result<(), ServerError> {
        if let Some(pending) = self.pending.take() {
            self.asking = Some(pending.begin()?);
        }
        Ok(())
    }

    pub(super) fn finish(&mut self) -> Result<(), ServerError> {
        self.asking.take().map_or(Ok(()), |asking| {
            asking
                .join()
                .map_err(|_| refusal("the machine pass issue panicked"))
        })
    }
}

pub(in crate::server) fn shutdown(
    supervisor: &crate::server::ConnectionSupervisor,
    bound: std::time::Duration,
    stop_accepting: impl FnOnce() -> Result<(), ServerError>,
    publishers: (
        &mut ConfiguredLinks,
        impl FnOnce() -> Result<(), ServerError>,
    ),
) -> Result<(), ServerError> {
    let (links, finalize) = publishers;
    let links_shutdown = links.finish();
    let shutdown =
        crate::server::shutdown::sequence::run(supervisor, bound, stop_accepting, finalize);
    super::raise_in_precedence(links_shutdown, shutdown)
}