liminal-server 0.24.9

//! The services registry and its lifecycle (SRV-028 R2 to R5).
//!
//! Read-only configuration templates and approved live records share the same
//! process lifecycle. Construction begins with contracts and an empty template,
//! then a held act records the program that completion leaves asleep.
//!
//! The durable store keeps one complete bounded snapshot for construction,
//! including prepared acts, approvals, receipts and notification intents. An old
//! file registry is imported once before admission; historical records remain.
//! Caller authority is decided through the admitted pass before mutations.

use std::collections::{BTreeMap, HashMap};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, MutexGuard};

use crate::config::templates::is_segment;
use crate::config::types::ServiceTemplateConfig;

#[path = "lifecycle_run.rs"]
mod run;

#[path = "plan.rs"]
pub mod plan;
#[path = "site/processes.rs"]
mod processes;
#[path = "site.rs"]
pub mod site;
#[path = "templates.rs"]
pub mod templates;
#[path = "trial.rs"]
pub mod trial;

/// The Lys app that owns the `liminal.*` kinds.
const APP: &str = "liminal";

#[path = "site/lifecycle_records.rs"]
mod records;
pub use records::{ServiceRecord, ServiceState, Transition};

/// Why a lifecycle request did nothing, by name.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum LifecycleError {
    /// No template of that name.
    UnknownTemplate(String),
    /// The name is not one segment.
    NameInvalid,
    /// A service of that template already has that name.
    NameTaken(String),
    /// No service of that id.
    UnknownService(String),
    /// The verb does not apply in the service's state.
    NotNow {
        /// The verb asked.
        verb: &'static str,
        /// The state it was asked in.
        state: ServiceState,
        /// Whether a process was running.
        running: bool,
    },
    /// Lys refused or could not be asked to place the service.
    Placement(String),
    /// No app credential is configured, so nothing can be placed.
    NoPlacer,
    /// The registry could not be read or written.
    Store(String),
    /// The template's command could not be started or stopped.
    Process(String),
}

impl std::fmt::Display for LifecycleError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::UnknownTemplate(name) => write!(formatter, "unknown_template: {name}"),
            Self::NameInvalid => formatter
                .write_str("name_invalid: a name is one segment of letters, digits, '-' or '_'"),
            Self::NameTaken(name) => write!(formatter, "name_taken: {name}"),
            Self::UnknownService(id) => write!(formatter, "unknown_service: {id}"),
            Self::NotNow {
                verb,
                state,
                running,
            } => write!(
                formatter,
                "not_now: {verb} does not apply to a service {state:?} with a process {}",
                if *running { "running" } else { "stopped" }
            ),
            Self::Placement(why) => write!(formatter, "placement_refused: {why}"),
            Self::NoPlacer => {
                formatter.write_str("placement_unavailable: no app_credential_path is configured")
            }
            Self::Store(why) => write!(formatter, "registry_unavailable: {why}"),
            Self::Process(why) => write!(formatter, "process_failed: {why}"),
        }
    }
}

impl std::error::Error for LifecycleError {}

/// Places a resource under its parent in Lys.
pub trait Placer: Send + Sync + std::fmt::Debug {
    /// Places `child` (kind, id) under `parent` (kind, id) once, under
    /// `operation`.
    ///
    /// # Errors
    /// Lys's refusal by its name, or why it could not be asked.
    fn place(
        &self,
        operation: &str,
        child: (&str, &str),
        parent: (&str, &str),
    ) -> Result<(), String>;
}

#[path = "site/placement.rs"]
mod placement;
pub use placement::LysPlacer;

/// What a service process's own credential admits: registering its
/// subjects, and nothing else.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ProcessGrant {
    /// The service the process runs.
    pub service: String,
    /// The only subjects it may register.
    pub subjects: Vec<String>,
}

impl ProcessGrant {
    /// Whether the process may serve `subject`.
    #[must_use]
    pub fn serves(&self, subject: &str) -> bool {
        self.subjects.iter().any(|own| own == subject)
    }
}

/// How long a start admitted before the bound address is handed over waits
/// to be told it.
const ADDRESS_WAIT: std::time::Duration = std::time::Duration::from_secs(5);

/// The registry, its templates and the processes it started.
pub struct Lifecycle {
    templates: BTreeMap<String, ServiceTemplateConfig>,
    dir: PathBuf,
    network: String,
    /// The address a started process connects to, set once the listener binds.
    server: std::sync::OnceLock<String>,
    /// Woken when `server` is set: the accept worker starts at bind, before
    /// the runtime hands the address over, so a start admitted in between
    /// waits to be told it rather than being refused.
    told: (Mutex<()>, std::sync::Condvar),
    placer: Option<Arc<dyn Placer>>,
    registry: Mutex<Registry>,
    site_config: std::sync::OnceLock<site::storage::SiteBindings>,
    changed: std::sync::Condvar,
    site_console: std::sync::OnceLock<std::sync::Weak<crate::console::ConsoleHub>>,
}

#[derive(Default)]
struct Registry {
    records: BTreeMap<String, ServiceRecord>,
    running: HashMap<String, run::Running>,
    tokens: HashMap<[u8; 32], ProcessGrant>,
    stopping_all: bool,
    construction: site::Construction,
    construction_hash: Option<haematite::tree::Hash>,
    trials: BTreeMap<String, BTreeMap<u64, trial::Registration>>,
    poisoned: bool,
}

impl Lifecycle {
    /// Opens the registry in `dir`, reading every record once.
    ///
    /// `network` is the Lys network services are placed under; the address a
    /// started process connects to is given by [`Self::listening`].
    ///
    /// # Errors
    /// The directory cannot be made or read, or a record in it is not one.
    pub fn open(
        templates: &[ServiceTemplateConfig],
        dir: PathBuf,
        network: String,
        placer: Option<Arc<dyn Placer>>,
    ) -> Result<Arc<Self>, LifecycleError> {
        site::create_private_dir(&dir).map_err(|error| store(&dir, &error))?;
        let snapshot = site::storage::load(&dir)?;
        Ok(Self::open_snapshot(
            templates, dir, network, placer, snapshot,
        ))
    }

    fn open_snapshot(
        templates: &[ServiceTemplateConfig],
        dir: PathBuf,
        network: String,
        placer: Option<Arc<dyn Placer>>,
        snapshot: site::storage::Snapshot,
    ) -> Arc<Self> {
        Arc::new(Self {
            templates: templates
                .iter()
                .map(|template| (template.name.clone(), template.clone()))
                .collect(),
            dir,
            network,
            server: std::sync::OnceLock::new(),
            told: (Mutex::new(()), std::sync::Condvar::new()),
            placer,
            site_config: std::sync::OnceLock::new(),
            changed: std::sync::Condvar::new(),
            site_console: std::sync::OnceLock::new(),
            registry: Mutex::new(Registry {
                records: snapshot.records,
                construction: snapshot.construction,
                ..Registry::default()
            }),
        })
    }

    /// Records the address the server listens on; a process is started only
    /// after it is known. A second call changes nothing.
    pub fn listening(&self, address: String) {
        let _first = self.server.set(address);
        let (lock, told) = &self.told;
        drop(
            lock.lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner),
        );
        told.notify_all();
    }

    /// The address the server listens on, waiting up to [`ADDRESS_WAIT`] to be
    /// told it: the runtime sets it straight after the bind, so the wait is
    /// only ever the few instructions between the two; the bound covers a
    /// listener that never binds.
    pub(super) fn server_address(&self) -> Option<&str> {
        if let Some(address) = self.server.get() {
            return Some(address);
        }
        let (lock, told) = &self.told;
        drop(
            told.wait_timeout_while(
                lock.lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner),
                ADDRESS_WAIT,
                |()| self.server.get().is_none(),
            )
            .unwrap_or_else(std::sync::PoisonError::into_inner),
        );
        self.server.get().map(String::as_str)
    }

    fn lock(&self) -> Result<MutexGuard<'_, Registry>, LifecycleError> {
        let registry = self
            .registry
            .lock()
            .map_err(|_| LifecycleError::Store("registry_lock_poisoned".into()))?;
        if registry.poisoned {
            return Err(LifecycleError::Store(
                "registry_commit_uncertain: reopen required".into(),
            ));
        }
        Ok(registry)
    }

    /// The templates and every service `visible` keeps, with whether a
    /// process is running for it, as the `services.list` answer.
    ///
    /// # Errors
    /// The registry lock is poisoned.
    pub fn list(
        &self,
        visible: impl Fn(&ServiceRecord) -> bool,
    ) -> Result<serde_json::Value, LifecycleError> {
        let registry = self.lock()?;
        let services: Vec<serde_json::Value> = registry
            .records
            .values()
            .filter(|record| visible(record))
            .map(|record| {
                let mut value = serde_json::to_value(record)
                    .map_err(|_| LifecycleError::Store("service_encoding_failed".into()))?;
                if let Some(object) = value.as_object_mut() {
                    object.insert(
                        "running".to_owned(),
                        registry.running.contains_key(&record.id).into(),
                    );
                }
                Ok(value)
            })
            .collect::<Result<_, LifecycleError>>()?;
        let mut templates: Vec<serde_json::Value> = self
            .templates
            .values()
            .map(|template| {
                serde_json::json!({"name": template.name, "subjects": template.subjects})
            })
            .collect();
        templates.extend(registry.construction.templates.values().filter(|record| !record.retired)
            .map(|record| serde_json::json!({"name":record.proposal.name,"subjects":record.proposal.subjects})));
        drop(registry);
        Ok(serde_json::json!({"templates": templates, "services": services}))
    }

    /// The record of `id`.
    ///
    /// # Errors
    /// No such service, or the lock is poisoned.
    pub fn record(&self, id: &str) -> Result<ServiceRecord, LifecycleError> {
        self.lock()?
            .records
            .get(id)
            .cloned()
            .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))
    }

    /// Creates a service named `name` from `template` for `holder`, asleep,
    /// after placing it and its subjects in Lys. The same `operation` from
    /// the same holder answers the same service and places nothing twice.
    ///
    /// # Errors
    /// By name: the template, the name, the placement or the store.
    pub fn create(
        &self,
        template: &str,
        name: &str,
        holder: &str,
        operation: &str,
        now_ms: u64,
    ) -> Result<ServiceRecord, LifecycleError> {
        let found = self
            .templates
            .get(template)
            .ok_or_else(|| LifecycleError::UnknownTemplate(template.to_owned()))?;
        if !is_segment(name) {
            return Err(LifecycleError::NameInvalid);
        }
        let id = format!("svc-{}", digest(&["liminal-service/v1", holder, operation]));
        {
            let registry = self.lock()?;
            if let Some(record) = registry.records.get(&id) {
                return Ok(record.clone());
            }
            if registry
                .records
                .values()
                .any(|record| record.template.as_deref() == Some(template) && record.name == name)
            {
                return Err(LifecycleError::NameTaken(name.to_owned()));
            }
        }
        let subjects: Vec<String> = found
            .subjects
            .iter()
            .map(|root| format!("{root}.{name}"))
            .collect();
        let placer = self.placer.as_ref().ok_or(LifecycleError::NoPlacer)?;
        let network = ("liminal.network", self.network.as_str());
        let mut children = vec![("liminal.service", id.as_str())];
        children.extend(
            subjects
                .iter()
                .map(|subject| ("liminal.subject", subject.as_str())),
        );
        for child in children {
            let op = format!("op-{}", digest(&["liminal-place/v1", child.0, child.1]));
            placer
                .place(&op, child, network)
                .map_err(LifecycleError::Placement)?;
        }
        let record = ServiceRecord {
            id: id.clone(),
            name: name.to_owned(),
            template: Some(template.to_owned()),
            purpose: None,
            contracts: BTreeMap::new(),
            builder: None,
            builder_grants: None,
            builder_bound: false,
            approved_by: Vec::new(),
            responsible: None,
            plan: plan::Plan::default(),
            approvals: Vec::new(),
            subjects,
            state: ServiceState::Asleep,
            created_by: holder.to_owned(),
            transitions: vec![Transition {
                from: ServiceState::Asleep,
                to: ServiceState::Asleep,
                holder: holder.to_owned(),
                at_ms: now_ms,
                note: Some("created".to_owned()),
            }],
        };
        let mut registry = self.lock()?;
        if let Some(raced) = registry.records.get(&id) {
            return Ok(raced.clone());
        }
        self.keep(&mut registry, &record)?;
        registry.records.insert(id, record.clone());
        drop(registry);
        Ok(record)
    }

    /// Whether a service serving `subject`, not retired, was placed in Lys at
    /// or after `since_ms`: a pass issued before then cannot carry its
    /// rights, so its offline refusal says nothing and Lys is asked live.
    #[must_use]
    pub fn placed_since(&self, subject: &str, since_ms: u64) -> bool {
        self.lock().is_ok_and(|registry| {
            registry.records.values().any(|record| {
                record.state != ServiceState::Retired
                    && record.subjects.iter().any(|served| served == subject)
                    && record
                        .transitions
                        .first()
                        .is_some_and(|created| created.at_ms >= since_ms)
            })
        })
    }

    fn keep(&self, registry: &mut Registry, record: &ServiceRecord) -> Result<(), LifecycleError> {
        site::storage::keep(self, registry, record)
    }

    /// The service's log file.
    fn log_path(&self, id: &str) -> PathBuf {
        self.dir.join(format!("{id}.log"))
    }
}

/// Takes a running process out of the registry and revokes its credential,
/// marking it asked to stop.
fn revoke(registry: &mut Registry, id: &str) -> Option<run::Running> {
    let process = registry.running.remove(id)?;
    registry.tokens.remove(&process.digest);
    Some(process)
}

fn store(path: &Path, error: &std::io::Error) -> LifecycleError {
    LifecycleError::Store(format!("{}: {}", path.display(), error.kind()))
}

/// 32 hex digits of a domain-separated digest of `parts`.
fn digest(parts: &[&str]) -> String {
    let mut hasher = blake3::Hasher::new();
    for part in parts {
        hasher.update(&u64::try_from(part.len()).unwrap_or(u64::MAX).to_be_bytes());
        hasher.update(part.as_bytes());
    }
    hasher.finalize().to_hex()[..32].to_owned()
}

#[cfg(test)]
#[path = "lifecycle_tests.rs"]
mod tests;

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