liminal-server 0.21.1

//! Starting and stopping a service's template process (SRV-028 R4).
//!
//! The process gets only the template's args, the environment names the
//! template passes through, and four variables of its own: `LIMINAL_SERVER`,
//! `LIMINAL_SERVICE`, `LIMINAL_SUBJECTS` (comma-separated) and
//! `LIMINAL_TOKEN`, a credential made for this process alone, kept only as
//! its digest in memory, that admits registering its subjects and nothing
//! else. Its output is appended to the service's log beside its record. A
//! watcher thread waits on it and reports its exit to the registry.

use std::io::Read;
use std::process::{Command, Stdio};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};

use super::{Lifecycle, LifecycleError, ServiceRecord};
use crate::config::types::ServiceTemplateConfig;

/// The system's kill program; std signals a child only with SIGKILL, and the
/// crate forbids `unsafe`.
const KILL: &str = "/bin/kill";

/// How long a stop waits for the exit after SIGKILL.
const KILL_WAIT: Duration = Duration::from_secs(5);

/// A started process.
#[derive(Debug)]
pub(super) struct Running {
    /// Its process id.
    pid: u32,
    /// The digest of its credential.
    pub(super) digest: [u8; 32],
    /// The holder that started it.
    pub(super) starter: String,
    /// Set by the watcher when the process has exited.
    exited: Arc<(Mutex<bool>, Condvar)>,
}

/// Starts `record`'s `template` for `starter`; answers the process and its
/// credential's digest.
pub(super) fn start(
    lifecycle: &Arc<Lifecycle>,
    record: &ServiceRecord,
    template: &ServiceTemplateConfig,
    starter: &str,
) -> Result<(Running, [u8; 32]), LifecycleError> {
    let server = lifecycle
        .server_address()
        .ok_or_else(|| LifecycleError::Process("the server is not listening yet".to_owned()))?;
    let token = credential()?;
    let log = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(lifecycle.log_path(&record.id))
        .map_err(|error| LifecycleError::Process(format!("log: {}", error.kind())))?;
    let log_err = log
        .try_clone()
        .map_err(|error| LifecycleError::Process(format!("log: {}", error.kind())))?;
    let mut command = Command::new(&template.command);
    command
        .args(&template.args)
        .env_clear()
        .stdin(Stdio::null())
        .stdout(log)
        .stderr(log_err);
    for name in &template.env {
        if let Some(value) = std::env::var_os(name) {
            command.env(name, value);
        }
    }
    command
        .env("LIMINAL_SERVER", server)
        .env("LIMINAL_SERVICE", &record.id)
        .env("LIMINAL_SUBJECTS", record.subjects.join(","))
        .env("LIMINAL_TOKEN", &token);
    let mut child = command.spawn().map_err(|error| {
        LifecycleError::Process(format!(
            "{} could not start: {}",
            template.command.display(),
            error.kind()
        ))
    })?;
    let pid = child.id();
    let exited = Arc::new((Mutex::new(false), Condvar::new()));
    let watcher = Arc::clone(lifecycle);
    let seen = Arc::clone(&exited);
    let id = record.id.clone();
    std::thread::Builder::new()
        .name(format!("service-{id}"))
        .spawn(move || {
            let status = child.wait().map_or_else(
                |error| format!("wait failed: {}", error.kind()),
                |s| s.to_string(),
            );
            let (flag, signal) = &*seen;
            if let Ok(mut done) = flag.lock() {
                *done = true;
            }
            signal.notify_all();
            watcher.exited(&id, &status, now_ms());
        })
        .map_err(|error| LifecycleError::Process(format!("watcher: {}", error.kind())))?;
    let digest = *blake3::hash(token.as_bytes()).as_bytes();
    tracing::info!(service = %record.id, pid, "service process started");
    Ok((
        Running {
            pid,
            digest,
            starter: starter.to_owned(),
            exited,
        },
        digest,
    ))
}

/// Sends SIGTERM, waits `timeout_ms`, then SIGKILL and waits again; a
/// process that already exited is never signalled.
pub(super) fn stop(process: &Running, timeout_ms: u64) -> Result<(), String> {
    if signal(process, "TERM")? {
        return Ok(());
    }
    if wait(process, Duration::from_millis(timeout_ms)) {
        return Ok(());
    }
    if signal(process, "KILL")? {
        return Ok(());
    }
    if wait(process, KILL_WAIT) {
        Ok(())
    } else {
        Err(format!(
            "process {} did not exit after SIGKILL",
            process.pid
        ))
    }
}

/// Signals the process with `kill -<name>` unless it has exited; `Ok(true)`
/// when it had. The exit flag is held across the signal, so the pid is still
/// our unreaped child and cannot have been reused.
fn signal(process: &Running, name: &str) -> Result<bool, String> {
    let (flag, _) = &*process.exited;
    let done = flag
        .lock()
        .map_err(|_| "the exit flag is poisoned".to_owned())?;
    if *done {
        return Ok(true);
    }
    let sent = Command::new(KILL)
        .arg(format!("-{name}"))
        .arg(process.pid.to_string())
        .stdin(Stdio::null())
        .stdout(Stdio::null())
        .stderr(Stdio::null())
        .status();
    drop(done);
    match sent {
        Ok(status) if status.success() => Ok(false),
        Ok(status) => Err(format!("kill -{name} {} answered {status}", process.pid)),
        Err(error) => Err(format!(
            "kill -{name} {} could not run: {}",
            process.pid,
            error.kind()
        )),
    }
}

/// Waits up to `limit` for the exit; whether it came.
fn wait(process: &Running, limit: Duration) -> bool {
    let (flag, signal) = &*process.exited;
    let deadline = Instant::now() + limit;
    let Ok(mut done) = flag.lock() else {
        return false;
    };
    while !*done {
        let left = deadline.saturating_duration_since(Instant::now());
        if left.is_zero() {
            return false;
        }
        match signal.wait_timeout(done, left) {
            Ok((guard, _)) => done = guard,
            Err(_) => return false,
        }
    }
    true
}

/// 32 random bytes as 64 hex digits.
fn credential() -> Result<String, LifecycleError> {
    let mut bytes = [0_u8; 32];
    std::fs::File::open("/dev/urandom")
        .and_then(|mut source| source.read_exact(&mut bytes))
        .map_err(|error| {
            LifecycleError::Process(format!("no random credential: {}", error.kind()))
        })?;
    Ok(hex::encode(bytes))
}

/// Milliseconds since the epoch.
pub(super) fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |since| {
            u64::try_from(since.as_millis()).unwrap_or(u64::MAX)
        })
}