liminal-server 0.28.5

//! Installs the services registry once the listener is bound (SRV-028 R5):
//! the registry is read by one scan, and the services last recorded live
//! are started off the start path, so the start line does not grow with
//! them.

use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::thread::JoinHandle;

use crate::ServerError;
use crate::config::ServerConfig;
use crate::services::lifecycle::{Lifecycle, LysPlacer, Placer};
use crate::services::lys_http::{HttpLys, LysHttp};

/// The startup draft pull (SRV-041 R4), made once the listener binds.
type StartupPull = Box<dyn FnOnce() + Send>;

/// Stops every service process when the server's run ends; their states
/// stay, so a live service starts again with the next start.
pub struct Services {
    lifecycle: Option<Arc<Lifecycle>>,
    /// Pending until the listener binds; then pulling on its own thread,
    /// joined when the stop begins, before the Lys runtime it asks on goes.
    pull: Mutex<(Option<StartupPull>, Option<JoinHandle<()>>)>,
}

impl std::fmt::Debug for Services {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter
            .debug_struct("Services")
            .field("lifecycle", &self.lifecycle)
            .finish_non_exhaustive()
    }
}

impl Services {
    fn new(lifecycle: Option<Arc<Lifecycle>>, pull: Option<StartupPull>) -> Self {
        Self {
            lifecycle,
            pull: Mutex::new((pull, None)),
        }
    }

    /// Gives the registry the address the listener bound and starts every
    /// service last recorded live, off the start path; begins the startup
    /// draft pull on its own thread.
    ///
    /// # Errors
    /// The restart or pull thread could not start.
    pub fn listening(&self, address: std::net::SocketAddr) -> Result<(), ServerError> {
        self.begin_pull()?;
        let Some(lifecycle) = self.lifecycle.as_ref() else {
            return Ok(());
        };
        lifecycle.listening(address.to_string());
        let restarting = Arc::clone(lifecycle);
        std::thread::Builder::new()
            .name("liminal-services-restart".to_owned())
            .spawn(move || {
                let now_ms = std::time::SystemTime::now()
                    .duration_since(std::time::UNIX_EPOCH)
                    .map_or(0, |since| {
                        u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
                    });
                for (service, error) in restarting.restart_live(now_ms) {
                    tracing::warn!(%service, %error, "a live service could not be restarted");
                }
                for (service, error) in restarting.catch_up() {
                    tracing::warn!(%service, %error, "SERVICE_PLACEMENT_FAILED: retry at next boot");
                }
            })
            .map_err(|error| refused(format!("the services restart could not start: {error}")))?;
        Ok(())
    }
}

impl Services {
    fn begin_pull(&self) -> Result<(), ServerError> {
        let mut pull = self
            .pull
            .lock()
            .map_err(|_| refused("the startup draft pull lock is poisoned".to_owned()))?;
        if let Some(pending) = pull.0.take() {
            pull.1 = Some(
                std::thread::Builder::new()
                    .name("liminal-draft-startup".to_owned())
                    .spawn(pending)
                    .map_err(|error| {
                        refused(format!("the startup draft pull could not start: {error}"))
                    })?,
            );
        }
        drop(pull);
        Ok(())
    }

    /// The server's stop was asked: an exit from here keeps its service's
    /// recorded state (see [`Lifecycle::shutdown_begun`]). A startup draft
    /// pull still asking is waited for, its asks bounded by their timeouts.
    pub fn shutdown_begun(&self) {
        let pulling = self.pull.lock().map_or_else(
            |_| {
                tracing::error!("the startup draft pull lock is poisoned; the pull is not joined");
                None
            },
            |mut pull| pull.1.take(),
        );
        if let Some(pulling) = pulling
            && pulling.join().is_err()
        {
            tracing::error!("the startup draft pull panicked");
        }
        if let Some(lifecycle) = self.lifecycle.as_ref() {
            lifecycle.shutdown_begun();
        }
    }
}

impl Drop for Services {
    fn drop(&mut self) {
        if let Some(lifecycle) = self.lifecycle.take() {
            lifecycle.shutdown();
        }
    }
}

const fn refused(message: String) -> ServerError {
    ServerError::ConfigValidation { message }
}

/// Opens and installs the registry on the supervisor before any connection
/// is admitted, so no request finds it missing; holds nothing without
/// templates. [`Services::listening`] then gives it the bound address and
/// starts the restart of live services on its own thread.
///
/// # Errors
/// The registry, legacy placement journal or app credential cannot be read,
/// or Lys cannot be reached as configured.
pub fn install(
    config: &ServerConfig,
    supervisor: &crate::server::connection::ConnectionSupervisor,
    lys_ask: Option<&tokio::runtime::Runtime>,
    services: Option<&Arc<crate::server::connection::services::LiminalConnectionServices>>,
    health: &crate::health::HealthServerHandle,
) -> Result<Services, ServerError> {
    let Some(lys) = config.auth.as_ref().and_then(|auth| auth.lys.as_ref()) else {
        let lifecycle = open_lifecycle(config)?;
        if let Some(lifecycle) = lifecycle.as_ref() {
            // ACCESS-015 R1: the one construction installer, as the declared holder's.
            if let (Some(limits), Some(services)) = (config.open_construction(), services) {
                services
                    .install_open_construction(
                        Arc::clone(lifecycle),
                        limits,
                        config.templates.policy.clone(),
                    )
                    .map_err(|error| refused(error.to_string()))?;
            }
            supervisor.install_lifecycle(Arc::clone(lifecycle))?;
        }
        return Ok(Services::new(lifecycle, None));
    };
    let templates = config.template_sources();
    let placement_path = config
        .persistence_path
        .as_ref()
        .map(|store| tap_placement_journal_path(store, &lys.network_id))
        .transpose()
        .map_err(refused)?;
    let client = lys_client(lys, lys_ask)?;
    if config.services.profile()? == crate::config::types::ServiceProfile::Full
        && let (Some(path), Some((http, credential))) = (placement_path.as_ref(), client.as_ref())
    {
        let installed = crate::console::tap_placement::TapPlacements::start(
            path,
            &lys.network_id,
            Arc::clone(http),
            credential.clone(),
        )
        .and_then(|placements| supervisor.install_tap_placements(placements));
        if let Err(error) = installed {
            tracing::warn!(%error, "TAP_SUBJECT_PLACEMENT_INSTALL_FAILED: taps still serve; retry at next boot");
        }
    }
    if client.is_none() || config.persistence_path.is_none() {
        tracing::warn!(
            "TAP_SUBJECT_PLACEMENT_UNAVAILABLE: app_credential_path, persistence_path and a Lys runtime are required"
        );
    }
    if templates.is_empty() && lys.construction.is_none() && lys.drafts.is_none() {
        return Ok(Services::new(None, None));
    }
    let store = config.persistence_path.as_ref().ok_or_else(|| {
        refused(format!(
            "{}: persistence_path is required",
            config.templates_place()
        ))
    })?;
    let placer: Option<Arc<dyn Placer>> = client.as_ref().map(|(http, credential)| {
        Arc::new(LysPlacer::new(http.clone(), credential.clone())) as Arc<dyn Placer>
    });
    let lifecycle = if let Some(construction) = &lys.construction {
        Lifecycle::open_bounded(
            templates,
            registry_dir(store),
            lys.network_id.clone(),
            placer,
            &construction.limits,
        )
    } else {
        Lifecycle::open(
            templates,
            registry_dir(store),
            lys.network_id.clone(),
            placer,
        )
    }
    .map_err(|error| refused(error.to_string()))?;
    let mut pull = None;
    if let Some(services) = services {
        if lys.construction.is_some() || lys.drafts.is_some() {
            let (http, credential) = client
                .as_ref()
                .ok_or_else(|| refused("construction needs an app connector".into()))?;
            let handle = lys_ask
                .ok_or_else(|| refused("construction needs a Lys runtime".into()))?
                .handle()
                .clone();
            pull = Some(services.install_approval_runtime(
                lys,
                lifecycle.clone(),
                http.clone(),
                credential.clone(),
                handle,
                health,
            )?);
        }
    }
    supervisor.install_lifecycle(Arc::clone(&lifecycle))?;
    Ok(Services::new(Some(lifecycle), pull))
}

/// The HTTP client services are placed through, with this liminal's app
/// credential.
type LysClient = (Arc<dyn LysHttp>, String);

/// The Lys app connector services are placed through: the credential read
/// once from `app_credential_path` and the HTTP client on the ask runtime;
/// none when either is absent.
fn lys_client(
    lys: &crate::config::types::LysConfig,
    lys_ask: Option<&tokio::runtime::Runtime>,
) -> Result<Option<LysClient>, ServerError> {
    let (Some(path), Some(runtime)) = (lys.app_credential_path.as_ref(), lys_ask) else {
        return Ok(None);
    };
    let credential = std::fs::read_to_string(path).map_err(|error| {
        refused(format!(
            "auth.lys.app_credential_path '{}' cannot be read: {}",
            path.display(),
            error.kind()
        ))
    })?;
    let http: Arc<dyn LysHttp> = Arc::new(
        HttpLys::new(
            lys.api_url.as_deref().unwrap_or(&lys.issuer),
            runtime.handle().clone(),
        )
        .map_err(|error| refused(error.to_string()))?,
    );
    Ok(Some((http, credential.trim().to_owned())))
}

/// The lifecycle of an exchange with no `[auth]` (ACCESS-012 R1): its
/// top-level `[[templates]]` boot a registry beside the store under the
/// network `open`, placing nothing in Lys; `[open.construction]` bounds it
/// and boots it with no template (ACCESS-015 R1). Neither, no lifecycle.
///
/// # Errors
/// The registry cannot be made or read, or the templates have no store.
pub fn open_lifecycle(config: &ServerConfig) -> Result<Option<Arc<Lifecycle>>, ServerError> {
    let construction = config.open_construction();
    if config.templates.is_empty() && construction.is_none() {
        return Ok(None);
    }
    let store = config
        .persistence_path
        .as_ref()
        .ok_or_else(|| refused("templates: persistence_path is required".to_owned()))?;
    let network = Lifecycle::OPEN_NETWORK.to_owned();
    match construction {
        Some(limits) => Lifecycle::open_bounded(
            &config.templates,
            registry_dir(store),
            network,
            None,
            &limits,
        ),
        None => Lifecycle::open(&config.templates, registry_dir(store), network, None),
    }
    .map(Some)
    .map_err(|error| refused(error.to_string()))
}

/// The registry sits beside the store, never inside it: the store directory
/// belongs to its engine.
pub fn registry_dir(store: &std::path::Path) -> PathBuf {
    let name = store.file_name().map_or_else(
        || "liminal".into(),
        |name| name.to_string_lossy().into_owned(),
    );
    store.with_file_name(format!("{name}-services"))
}

/// Keeps the journal beside both persistence directories and migrates it
/// before the registry's service-record scan.
pub fn tap_placement_journal_path(
    store: &std::path::Path,
    network: &str,
) -> Result<PathBuf, String> {
    let name = store.file_name().map_or_else(
        || "liminal".into(),
        |name| name.to_string_lossy().into_owned(),
    );
    let path = store.with_file_name(format!("{name}-console-tap-placements.json"));
    crate::console::tap_placement::PlacementBook::relocate_legacy(
        &registry_dir(store).join("console-tap-placements.json"),
        &path,
        network,
    )?;
    Ok(path)
}

#[cfg(test)]
mod tests {
    use super::{open_lifecycle, registry_dir};
    use crate::config::types::ServerConfig;

    /// An open exchange (no `[auth]`) with `templates` at the top level and a
    /// store in `dir`.
    fn open_config(dir: &std::path::Path, templates: &str) -> ServerConfig {
        let text = format!(
            "listen_address = \"127.0.0.1:1\"\nhealth_listen_address = \"127.0.0.1:2\"\n\
             drain_timeout_ms = 5000\nchannels = []\nrouting_rules = []\n\
             persistence_path = \"{}\"\n{templates}",
            dir.join("store").display()
        );
        toml::from_str(&text).unwrap_or_else(|error| unreachable!("fixture parses: {error}"))
    }

    const ECHO: &str = "[[templates]]\nname = \"echo\"\ncommand = \"/bin/sh\"\n\
                        args = [\"-c\", \"true\"]\nsubjects = [\"echo\"]\nstop_timeout_ms = 500\n";

    #[test]
    fn access012_an_open_exchange_with_a_template_boots_a_lifecycle_under_open_with_no_placer()
    -> Result<(), Box<dyn std::error::Error>> {
        let dir = std::env::temp_dir().join(format!(
            "liminal-open-boot-{}-{}",
            std::process::id(),
            std::time::SystemTime::UNIX_EPOCH.elapsed()?.as_nanos()
        ));
        std::fs::create_dir_all(&dir)?;
        let lifecycle = open_lifecycle(&open_config(&dir, ECHO))?
            .ok_or("one top-level template boots the lifecycle")?;
        assert_eq!(lifecycle.network_id(), "open");
        assert!(
            !lifecycle.places(),
            "an open exchange places nothing in Lys"
        );
        assert!(
            registry_dir(&dir.join("store")).is_dir(),
            "the registry is made beside the store"
        );
        std::fs::remove_dir_all(&dir)?;
        Ok(())
    }

    #[test]
    fn access012_an_open_exchange_with_no_templates_boots_no_lifecycle()
    -> Result<(), Box<dyn std::error::Error>> {
        let dir =
            std::env::temp_dir().join(format!("liminal-open-boot-none-{}", std::process::id()));
        assert!(open_lifecycle(&open_config(&dir, ""))?.is_none());
        Ok(())
    }

    #[test]
    fn access015_open_construction_boots_a_lifecycle_with_no_template()
    -> Result<(), Box<dyn std::error::Error>> {
        let dir = tempfile::tempdir()?;
        let construction = "[open.construction]\nsites = 8\nsubjects = 8\nschema_bytes = 16384\n\
                            text_bytes = 1024\nsteps = 32\ntransitions = 128\ndrafts = 128\n\
                            approvers = 2\nsnapshot_bytes = 1048576\nrequest_bytes = 65536\n";
        let lifecycle = open_lifecycle(&open_config(dir.path(), construction))?
            .ok_or("[open.construction] boots the lifecycle")?;
        assert_eq!(lifecycle.network_id(), "open");
        assert!(!lifecycle.places());
        Ok(())
    }

    #[test]
    fn the_registry_sits_beside_the_store() {
        assert_eq!(
            registry_dir(std::path::Path::new("/data/liminal/store")),
            std::path::PathBuf::from("/data/liminal/store-services")
        );
    }
}