liminal_server/services/site/

processes.rs

//! Process state changes share the lifecycle registry.

use super::{
    Lifecycle, LifecycleError, ProcessGrant, ServiceRecord, ServiceState, Transition, revoke, run,
};
use std::sync::Arc;

impl Lifecycle {
    /// 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.template_for(&registry, &record)?;
        let approved = record
            .template
            .as_ref()
            .and_then(|name| registry.construction.templates.get(name));
        let (process, digest) = run::start(self, &record, &template, approved, 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);
            // A site under construction runs nothing: it can be abandoned
            // (retired) but there is nothing to hold.
            if state == ServiceState::Building && to != ServiceState::Retired {
                return Err(LifecycleError::NotNow {
                    verb,
                    state,
                    running,
                });
            }
            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) -> Result<u64, LifecycleError> {
        let registry = self.lock()?;
        let record = registry
            .records
            .get(id)
            .ok_or_else(|| LifecycleError::UnknownService(id.into()))?;
        let name = record
            .template
            .as_ref()
            .ok_or_else(|| LifecycleError::UnknownTemplate("none".into()))?;
        if let Some(template) = registry.construction.templates.get(name) {
            return Ok(template.proposal.stop_timeout_ms);
        }
        let name = name.clone();
        drop(registry);
        self.templates
            .get(&name)
            .map(|template| template.stop_timeout_ms)
            .ok_or_else(|| LifecycleError::UnknownTemplate(name.clone()))
    }

    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()))?;
        if record.purpose.is_some()
            && record.transitions.len()
                >= self
                    .site_bindings()
                    .map_err(|error| LifecycleError::Store(error.to_string()))?
                    .limits
                    .transitions
                    .get()
        {
            return Err(LifecycleError::Store("transitions_bound_exceeded".into()));
        }
        record.state = to;
        record.transitions.push(Transition {
            from,
            to,
            holder: holder.to_owned(),
            at_ms: now_ms,
            note,
        });
        self.keep(&mut registry, &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::Building {
            return Err(LifecycleError::NotNow {
                verb: "mark_live",
                state,
                running: false,
            });
        }
        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
    }

    /// 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 = match self.timeout_of(&id) {
                Ok(timeout) => timeout,
                Err(error) => {
                    tracing::error!(service = %id, %error, "service_stop_policy_unavailable");
                    return;
                }
            };
            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.
    pub(in crate::services) 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
                });
                self.changed.notify_all();
                (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");
        }
    }
}