liminal_server/services/

lifecycle.rs

//! The services registry and its lifecycle (SRV-028 R2 to R5).
//!
//! A service is made from one of the server's own templates
//! (`[[auth.lys.templates]]`), never from a command line a request names. Its
//! record is kept under the persistence path, one JSON file per service,
//! written whole with one full flush per transition and read back by one
//! directory scan at start: nothing is replayed. A held or retired service is
//! kept with its history; nothing here removes a record.
//!
//! Who may act is decided before any method here is called (the lifecycle
//! route asks Lys); this module records the holder it is handed. A create is
//! placed in Lys first, the service under the network and each subject under
//! the network, and is recorded only when every placement is answered.

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

use serde::{Deserialize, Serialize};

use super::lys_http::LysHttp;
use crate::config::templates::is_segment;
use crate::config::types::ServiceTemplateConfig;

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

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

/// Where a service stands.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ServiceState {
    /// Registered, not running: created, or exited without being asked.
    Asleep,
    /// Its worker registered its subjects.
    Live,
    /// Stopped by a holder; woken by one.
    Held,
    /// Stopped for good; kept with its history.
    Retired,
}

/// One change of state, with who made it and when.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct Transition {
    /// The state before.
    pub from: ServiceState,
    /// The state after.
    pub to: ServiceState,
    /// The Lys holder that asked, or `process` for the service's own exit.
    pub holder: String,
    /// Milliseconds since the epoch.
    pub at_ms: u64,
    /// What happened beside the state, such as an exit status.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub note: Option<String>,
}

/// One service, as kept.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ServiceRecord {
    /// Its id, which is also its `liminal.service` id in Lys.
    pub id: String,
    /// The name it was created with, one segment.
    pub name: String,
    /// The template it runs.
    pub template: String,
    /// The subjects it serves, `<root>.<name>` for each template root.
    pub subjects: Vec<String>,
    /// Where it stands.
    pub state: ServiceState,
    /// The holder that created it.
    pub created_by: String,
    /// Every transition, oldest first.
    pub transitions: Vec<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>;
}

/// [`Placer`] over Lys's placement route with this liminal's app credential.
pub struct LysPlacer {
    http: Arc<dyn LysHttp>,
    credential: String,
}

impl LysPlacer {
    /// A placer asking through `http` with `credential`.
    #[must_use]
    pub fn new(http: Arc<dyn LysHttp>, credential: String) -> Self {
        Self { http, credential }
    }
}

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

impl Placer for LysPlacer {
    fn place(
        &self,
        operation: &str,
        child: (&str, &str),
        parent: (&str, &str),
    ) -> Result<(), String> {
        let route = format!("apps/{APP}/placements");
        let body = serde_json::json!({
            "operation": operation,
            "child": {"kind": child.0, "id": child.1},
            "parent": {"kind": parent.0, "id": parent.1},
        });
        let answer = self
            .http
            .post(&self.credential, &route, &body)
            .map_err(|error| error.0)?;
        if answer.succeeded() {
            Ok(())
        } else {
            Err(answer.refusal())
        }
    }
}

/// 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.
#[derive(Debug)]
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>,
}

#[derive(Debug, Default)]
struct Registry {
    records: BTreeMap<String, ServiceRecord>,
    running: HashMap<String, run::Running>,
    tokens: HashMap<[u8; 32], ProcessGrant>,
    stopping_all: 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> {
        std::fs::create_dir_all(&dir).map_err(|error| store(&dir, &error))?;
        let mut records = BTreeMap::new();
        for entry in std::fs::read_dir(&dir).map_err(|error| store(&dir, &error))? {
            let path = entry.map_err(|error| store(&dir, &error))?.path();
            if path.extension().is_none_or(|ext| ext != "json") {
                continue;
            }
            let bytes = std::fs::read(&path).map_err(|error| store(&path, &error))?;
            let record: ServiceRecord = serde_json::from_slice(&bytes).map_err(|error| {
                LifecycleError::Store(format!(
                    "{} is not a service record: {error}",
                    path.display()
                ))
            })?;
            records.insert(record.id.clone(), record);
        }
        Ok(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,
            registry: Mutex::new(Registry {
                records,
                ..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> {
        self.registry
            .lock()
            .map_err(|_| LifecycleError::Store("the registry lock is poisoned".to_owned()))
    }

    /// 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).unwrap_or_default();
                if let Some(object) = value.as_object_mut() {
                    object.insert(
                        "running".to_owned(),
                        registry.running.contains_key(&record.id).into(),
                    );
                }
                value
            })
            .collect();
        drop(registry);
        let templates: Vec<serde_json::Value> = self
            .templates
            .values()
            .map(|template| {
                serde_json::json!({"name": template.name, "subjects": template.subjects})
            })
            .collect();
        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 == 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: template.to_owned(),
            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(&record)?;
        registry.records.insert(id, record.clone());
        drop(registry);
        Ok(record)
    }

    /// Starts an asleep service's template for `holder`; it is recorded
    /// live when its worker registers.
    ///
    /// # Errors
    /// By name: the service, its state, or the start.
    pub fn deploy(
        self: &Arc<Self>,
        id: &str,
        holder: &str,
    ) -> Result<ServiceRecord, LifecycleError> {
        self.start(id, holder, "deploy", ServiceState::Asleep)
    }

    /// Starts a held service's template again for `holder`.
    ///
    /// # Errors
    /// As [`Self::deploy`].
    pub fn wake(self: &Arc<Self>, id: &str, holder: &str) -> Result<ServiceRecord, LifecycleError> {
        self.start(id, holder, "wake", ServiceState::Held)
    }

    fn start(
        self: &Arc<Self>,
        id: &str,
        holder: &str,
        verb: &'static str,
        from: ServiceState,
    ) -> Result<ServiceRecord, LifecycleError> {
        let mut registry = self.lock()?;
        let record = registry
            .records
            .get(id)
            .cloned()
            .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
        let running = registry.running.contains_key(id);
        if record.state != from || running || registry.stopping_all {
            return Err(LifecycleError::NotNow {
                verb,
                state: record.state,
                running,
            });
        }
        let template = self
            .templates
            .get(&record.template)
            .ok_or_else(|| LifecycleError::UnknownTemplate(record.template.clone()))?;
        let (process, digest) = run::start(self, &record, template, holder)?;
        registry.tokens.insert(
            digest,
            ProcessGrant {
                service: record.id.clone(),
                subjects: record.subjects.clone(),
            },
        );
        registry.running.insert(record.id.clone(), process);
        drop(registry);
        Ok(record)
    }

    /// Stops a running or live service for `holder` and records it held.
    ///
    /// # Errors
    /// By name: the service, its state, the stop or the store.
    pub fn hold(
        &self,
        id: &str,
        holder: &str,
        now_ms: u64,
    ) -> Result<ServiceRecord, LifecycleError> {
        self.stop(id, holder, now_ms, "hold", ServiceState::Held)
    }

    /// Stops a service if it runs and records it retired, for good.
    ///
    /// # Errors
    /// As [`Self::hold`].
    pub fn retire(
        &self,
        id: &str,
        holder: &str,
        now_ms: u64,
    ) -> Result<ServiceRecord, LifecycleError> {
        self.stop(id, holder, now_ms, "retire", ServiceState::Retired)
    }

    fn stop(
        &self,
        id: &str,
        holder: &str,
        now_ms: u64,
        verb: &'static str,
        to: ServiceState,
    ) -> Result<ServiceRecord, LifecycleError> {
        let (state, stopping) = {
            let mut registry = self.lock()?;
            let state = registry
                .records
                .get(id)
                .map(|record| record.state)
                .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
            let running = registry.running.contains_key(id);
            let applies = match to {
                ServiceState::Held => state == ServiceState::Live || running,
                _ => state != ServiceState::Retired,
            };
            if !applies {
                return Err(LifecycleError::NotNow {
                    verb,
                    state,
                    running,
                });
            }
            let process = revoke(&mut registry, id);
            drop(registry);
            (state, process)
        };
        if let Some(process) = stopping {
            run::stop(&process, self.timeout_of(id)).map_err(LifecycleError::Process)?;
        }
        self.transition(id, state, to, holder, now_ms, None)
    }

    fn timeout_of(&self, id: &str) -> u64 {
        self.lock()
            .ok()
            .and_then(|registry| {
                registry
                    .records
                    .get(id)
                    .map(|record| record.template.clone())
            })
            .and_then(|template| self.templates.get(&template).map(|t| t.stop_timeout_ms))
            .unwrap_or(1)
    }

    fn transition(
        &self,
        id: &str,
        from: ServiceState,
        to: ServiceState,
        holder: &str,
        now_ms: u64,
        note: Option<String>,
    ) -> Result<ServiceRecord, LifecycleError> {
        let mut registry = self.lock()?;
        let mut record = registry
            .records
            .get(id)
            .cloned()
            .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
        record.state = to;
        record.transitions.push(Transition {
            from,
            to,
            holder: holder.to_owned(),
            at_ms: now_ms,
            note,
        });
        self.keep(&record)?;
        registry.records.insert(id.to_owned(), record.clone());
        drop(registry);
        Ok(record)
    }

    /// What `token` admits, when it is a running service process's own.
    #[must_use]
    pub fn admit_process(&self, token: &[u8]) -> Option<ProcessGrant> {
        let digest = *blake3::hash(token).as_bytes();
        self.lock().ok()?.tokens.get(&digest).cloned()
    }

    /// Records `id` live when its worker registered, for the holder that
    /// started it; nothing when it already is.
    ///
    /// # Errors
    /// The service is unknown or the store failed.
    pub fn mark_live(&self, id: &str, now_ms: u64) -> Result<(), LifecycleError> {
        let (state, starter) = {
            let registry = self.lock()?;
            let state = registry
                .records
                .get(id)
                .map(|record| record.state)
                .ok_or_else(|| LifecycleError::UnknownService(id.to_owned()))?;
            let starter = registry
                .running
                .get(id)
                .map_or_else(|| "process".to_owned(), |process| process.starter.clone());
            drop(registry);
            (state, starter)
        };
        if state == ServiceState::Live {
            return Ok(());
        }
        self.transition(id, state, ServiceState::Live, &starter, now_ms, None)
            .map(|_| ())
    }

    /// Starts every service last recorded live, as the holder that last
    /// made it so, after a restart. A start that fails is recorded on the
    /// service and the rest still start.
    pub fn restart_live(self: &Arc<Self>, now_ms: u64) -> Vec<(String, LifecycleError)> {
        let live: Vec<(String, String)> = self.lock().map_or_else(
            |_| Vec::new(),
            |registry| {
                registry
                    .records
                    .values()
                    .filter(|record| record.state == ServiceState::Live)
                    .map(|record| {
                        let starter = record
                            .transitions
                            .last()
                            .map_or_else(|| record.created_by.clone(), |t| t.holder.clone());
                        (record.id.clone(), starter)
                    })
                    .collect()
            },
        );
        let mut failed = Vec::new();
        for (id, starter) in live {
            if let Err(error) = self.start(&id, &starter, "restart", ServiceState::Live) {
                let note = Some(format!("restart failed: {error}"));
                let _kept = self.transition(
                    &id,
                    ServiceState::Live,
                    ServiceState::Asleep,
                    "process",
                    now_ms,
                    note,
                );
                failed.push((id, error));
            }
        }
        failed
    }

    /// 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)
            })
        })
    }

    /// The server's shutdown has begun: from here no service starts, and a
    /// process that exits (a worker leaves when its connection closes) keeps
    /// its service's recorded state, so a live service is live at the next
    /// start. Called when the stop is asked, before any connection drains.
    pub fn shutdown_begun(&self) {
        if let Ok(mut registry) = self.lock() {
            registry.stopping_all = true;
        }
    }

    /// Stops every running process at server shutdown without changing a
    /// state, so a live service is started again by the next start.
    pub fn shutdown(&self) {
        self.shutdown_begun();
        let ids: Vec<String> = match self.lock() {
            Ok(registry) => registry.running.keys().cloned().collect(),
            Err(_) => return,
        };
        for id in ids {
            let timeout = self.timeout_of(&id);
            let process = match self.lock() {
                Ok(mut registry) => revoke(&mut registry, &id),
                Err(_) => return,
            };
            if let Some(process) = process
                && let Err(error) = run::stop(&process, timeout)
            {
                tracing::warn!(service = %id, error, "a service process did not stop at shutdown");
            }
        }
    }

    /// A process exited: unasked, it is recorded on its service; a stop that
    /// asked records its own state, and so does the server's shutdown once it
    /// has begun.
    fn exited(&self, id: &str, status: &str, now_ms: u64) {
        let (asked, stopping) = match self.lock() {
            Ok(mut registry) => {
                let asked = registry.running.remove(id).is_none_or(|process| {
                    registry.tokens.remove(&process.digest);
                    false
                });
                (asked, registry.stopping_all)
            }
            Err(_) => return,
        };
        if asked {
            return;
        }
        if stopping {
            tracing::info!(
                service = id,
                status,
                "a service process exited during shutdown; its state stays"
            );
            return;
        }
        let Ok(record) = self.record(id) else {
            return;
        };
        let note = Some(format!("exited unasked: {status}"));
        if let Err(error) = self.transition(
            id,
            record.state,
            ServiceState::Asleep,
            "process",
            now_ms,
            note,
        ) {
            tracing::warn!(service = id, %error, "a service exit could not be recorded");
        }
    }

    /// Writes `record` whole: a temporary file, one full flush, a rename and
    /// a flush of the directory, so a crash leaves the old record or the new.
    fn keep(&self, record: &ServiceRecord) -> Result<(), LifecycleError> {
        let path = self.dir.join(format!("{}.json", record.id));
        let temporary = self.dir.join(format!("{}.json.new", record.id));
        let bytes = serde_json::to_vec_pretty(record)
            .map_err(|error| LifecycleError::Store(error.to_string()))?;
        let mut file =
            std::fs::File::create(&temporary).map_err(|error| store(&temporary, &error))?;
        file.write_all(&bytes)
            .and_then(|()| file.sync_all())
            .map_err(|error| store(&temporary, &error))?;
        std::fs::rename(&temporary, &path).map_err(|error| store(&path, &error))?;
        std::fs::File::open(&self.dir)
            .and_then(|dir| dir.sync_all())
            .map_err(|error| store(&self.dir, &error))
    }

    /// 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;