liminal-server 0.25.4

use std::io::Read;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

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

/// Records every placement and refuses those whose child id is listed.
#[derive(Debug, Default)]
struct FakeLys {
    placed: Mutex<Vec<(String, String, String)>>,
    refuse: Option<&'static str>,
}

impl Placer for FakeLys {
    fn place(
        &self,
        operation: &str,
        child: (&str, &str),
        parent: (&str, &str),
    ) -> Result<(), String> {
        if self.refuse.is_some_and(|kind| kind == child.0) {
            return Err("WithheldFromAgents".to_owned());
        }
        assert_eq!(parent, ("liminal.network", "net"));
        self.placed
            .lock()
            .map_err(|_| "poisoned".to_owned())?
            .push((operation.to_owned(), child.0.to_owned(), child.1.to_owned()));
        Ok(())
    }
}

/// A fresh directory under the target's temporary space.
fn scratch(name: &str) -> PathBuf {
    let dir = std::env::temp_dir().join(format!("liminal-lifecycle-{name}-{}", std::process::id()));
    let _gone = std::fs::remove_dir_all(&dir);
    dir
}

/// A template running `script` under `/bin/sh -c`.
fn template(script: &str) -> ServiceTemplateConfig {
    ServiceTemplateConfig {
        name: "echo".to_owned(),
        command: PathBuf::from("/bin/sh"),
        args: vec!["-c".to_owned(), script.to_owned()],
        subjects: vec!["echo".to_owned()],
        env: Vec::new(),
        stop_timeout_ms: 2_000,
    }
}

fn pipe(dir: &Path, name: &str) -> Result<PathBuf, LifecycleError> {
    std::fs::create_dir_all(dir).map_err(|error| LifecycleError::Store(error.to_string()))?;
    let path = dir.join(name);
    if let Ok(metadata) = std::fs::symlink_metadata(&path) {
        use std::os::unix::fs::FileTypeExt;
        if metadata.file_type().is_fifo() {
            return Ok(path);
        }
        return Err(LifecycleError::Process(
            "test FIFO path contains another shape".into(),
        ));
    }
    let status = std::process::Command::new("/usr/bin/mkfifo")
        .arg(&path)
        .status()
        .map_err(|error| LifecycleError::Store(error.to_string()))?;
    if !status.success() {
        return Err(LifecycleError::Process("test FIFO creation failed".into()));
    }
    Ok(path)
}
fn quoted(path: &Path) -> String {
    format!("'{}'", path.display().to_string().replace('\'', "'\\''"))
}
fn parked(dir: &Path) -> Result<String, LifecycleError> {
    Ok(format!("read value < {}", quoted(&pipe(dir, "parked")?)))
}
fn ready(dir: &Path) -> Result<(), LifecycleError> {
    let mut byte = [0];
    std::fs::File::open(dir.join("ready"))
        .and_then(|mut file| file.read_exact(&mut byte))
        .map_err(|error| LifecycleError::Store(error.to_string()))
}

fn open(dir: &Path, script: &str, lys: Arc<FakeLys>) -> Arc<Lifecycle> {
    let command = (|| {
        let signal = pipe(dir, "ready")?;
        let body = if matches!(script, "park" | "ignore-term") {
            parked(dir)?
        } else {
            script.into()
        };
        Ok::<_, LifecycleError>(format!(
            "{}printf . > {}; {body}",
            if script == "ignore-term" {
                "trap '' TERM; "
            } else {
                ""
            },
            quoted(&signal)
        ))
    })()
    .unwrap_or_else(|error| unreachable!("fixture signal setup failed: {error}"));
    let lifecycle = Lifecycle::open(
        &[template(&command)],
        dir.to_path_buf(),
        "net".to_owned(),
        Some(lys),
    )
    .unwrap_or_else(|error| unreachable!("the registry opens: {error}"));
    lifecycle.listening("127.0.0.1:1".to_owned());
    lifecycle
}

/// Waits for the registry's own durable commit or process-close signal.
fn wait_record(
    lifecycle: &Lifecycle,
    id: &str,
    done: impl Fn(&super::ServiceRecord, bool) -> bool,
) -> Result<(), LifecycleError> {
    let mut registry = lifecycle.lock()?;
    loop {
        let record = registry
            .records
            .get(id)
            .ok_or_else(|| LifecycleError::UnknownService(id.into()))?;
        if done(record, registry.running.contains_key(id)) {
            return Ok(());
        }
        registry = lifecycle
            .changed
            .wait(registry)
            .map_err(|_| LifecycleError::Store("test registry signal poisoned".into()))?;
    }
}

#[test]
fn create_places_the_service_and_subjects_then_records_asleep() -> Result<(), LifecycleError> {
    let dir = scratch("create");
    let lys = Arc::new(FakeLys::default());
    let lifecycle = open(&dir, "true", Arc::clone(&lys));
    let record = lifecycle.create("echo", "one", "agent-a", "op-1", 10)?;
    assert_eq!(record.state, ServiceState::Asleep);
    assert_eq!(record.subjects, vec!["echo.one".to_owned()]);
    assert_eq!(record.created_by, "agent-a");
    let placed = lys.placed.lock().map(|p| p.clone()).unwrap_or_default();
    let kinds: Vec<(&str, &str)> = placed
        .iter()
        .map(|(_, kind, id)| (kind.as_str(), id.as_str()))
        .collect();
    assert_eq!(
        kinds,
        vec![
            ("liminal.service", record.id.as_str()),
            ("liminal.subject", "echo.one")
        ]
    );
    // The same operation from the same holder is the same service, placed once.
    assert_eq!(
        lifecycle.create("echo", "one", "agent-a", "op-1", 11)?,
        record
    );
    assert_eq!(lys.placed.lock().map(|p| p.len()).unwrap_or_default(), 2);
    Ok(())
}

#[test]
fn create_refused_by_lys_records_nothing() {
    let dir = scratch("refused");
    let lys = Arc::new(FakeLys {
        refuse: Some("liminal.subject"),
        ..FakeLys::default()
    });
    let lifecycle = open(&dir, "true", lys);
    let refused = lifecycle.create("echo", "one", "agent-a", "op-1", 10);
    assert_eq!(
        refused,
        Err(LifecycleError::Placement("WithheldFromAgents".to_owned()))
    );
    let listed = lifecycle.list(|_| true).unwrap_or_default();
    assert_eq!(listed["services"], serde_json::json!([]));
}

#[test]
fn create_refuses_an_unknown_template_a_bad_name_and_a_taken_name() -> Result<(), LifecycleError> {
    let dir = scratch("names");
    let lifecycle = open(&dir, "true", Arc::new(FakeLys::default()));
    assert_eq!(
        lifecycle.create("nope", "one", "a", "op-1", 1),
        Err(LifecycleError::UnknownTemplate("nope".to_owned()))
    );
    assert_eq!(
        lifecycle.create("echo", "o.ne", "a", "op-1", 1),
        Err(LifecycleError::NameInvalid)
    );
    lifecycle.create("echo", "one", "a", "op-1", 1)?;
    assert_eq!(
        lifecycle.create("echo", "one", "b", "op-2", 1),
        Err(LifecycleError::NameTaken("one".to_owned()))
    );
    Ok(())
}

#[test]
fn registry_survives_a_restart_with_every_transition() -> Result<(), LifecycleError> {
    let dir = scratch("restart");
    let lys = Arc::new(FakeLys::default());
    let first = open(&dir, "true", Arc::clone(&lys));
    let made = first.create("echo", "one", "agent-a", "op-1", 10)?;
    let held = first.retire(&made.id, "agent-a", 20)?;
    drop(first);
    let second = open(&dir, "true", lys);
    let read = second.record(&made.id)?;
    assert_eq!(read, held);
    assert_eq!(read.state, ServiceState::Retired);
    assert_eq!(read.transitions.len(), 2);
    Ok(())
}

#[test]
fn deploy_starts_the_template_and_goes_live_on_registration() -> Result<(), LifecycleError> {
    let dir = scratch("deploy");
    let lifecycle = open(&dir, "park", Arc::new(FakeLys::default()));
    let made = lifecycle.create("echo", "one", "agent-a", "op-1", 10)?;
    lifecycle.deploy(&made.id, "agent-b")?;
    ready(&dir)?;
    // A second deploy while it runs does nothing.
    assert!(matches!(
        lifecycle.deploy(&made.id, "agent-b"),
        Err(LifecycleError::NotNow { running: true, .. })
    ));
    lifecycle.mark_live(&made.id, 30)?;
    let live = lifecycle.record(&made.id)?;
    assert_eq!(live.state, ServiceState::Live);
    assert_eq!(
        live.transitions.last().map(|t| t.holder.as_str()),
        Some("agent-b")
    );
    let held = lifecycle.hold(&made.id, "agent-b", 40)?;
    assert_eq!(held.state, ServiceState::Held);
    assert_eq!(lifecycle.list(|_| true)?["services"][0]["running"], false);
    lifecycle.wake(&made.id, "agent-b")?;
    lifecycle.retire(&made.id, "agent-b", 50)?;
    assert_eq!(lifecycle.record(&made.id)?.state, ServiceState::Retired);
    Ok(())
}

#[test]
fn hold_stops_within_the_timeout_even_when_sigterm_is_ignored() -> Result<(), LifecycleError> {
    let dir = scratch("stubborn");
    let lifecycle = open(&dir, "ignore-term", Arc::new(FakeLys::default()));
    let made = lifecycle.create("echo", "one", "a", "op-1", 1)?;
    lifecycle.deploy(&made.id, "a")?;
    ready(&dir)?;
    lifecycle.hold(&made.id, "a", 2)?;
    assert_eq!(lifecycle.list(|_| true)?["services"][0]["running"], false);
    assert_eq!(lifecycle.record(&made.id)?.state, ServiceState::Held);
    Ok(())
}

#[test]
fn a_process_credential_admits_only_its_own_subjects() -> Result<(), LifecycleError> {
    let dir = scratch("token");
    let file = dir.join("token");
    std::fs::create_dir_all(&dir).map_err(|error| LifecycleError::Store(error.to_string()))?;
    let status = std::process::Command::new("/usr/bin/mkfifo")
        .arg(&file)
        .status()
        .map_err(|error| LifecycleError::Store(error.to_string()))?;
    assert!(status.success(), "credential signal FIFO was not created");
    let mut written = template(&format!(
        "printf %s \"$LIMINAL_TOKEN\" > \"$0\"; {}",
        parked(&dir)?
    ));
    written.args.push(file.display().to_string());
    let lifecycle = Lifecycle::open(
        &[written],
        dir,
        "net".to_owned(),
        Some(Arc::new(FakeLys::default())),
    )?;
    lifecycle.listening("127.0.0.1:1".to_owned());
    let made = lifecycle.create("echo", "one", "a", "op-1", 1)?;
    lifecycle.deploy(&made.id, "a")?;
    let mut token = [0_u8; 64];
    std::fs::File::open(&file)
        .and_then(|mut signal| signal.read_exact(&mut token))
        .map_err(|error| LifecycleError::Store(error.to_string()))?;
    let grant = lifecycle.admit_process(&token);
    assert_eq!(
        grant.as_ref().map(|g| g.service.as_str()),
        Some(made.id.as_str())
    );
    assert!(grant.as_ref().is_some_and(|g| g.serves("echo.one")));
    assert!(grant.as_ref().is_some_and(|g| !g.serves("echo.two")));
    assert_eq!(lifecycle.admit_process(b"not-a-token"), None);
    lifecycle.hold(&made.id, "a", 2)?;
    assert_eq!(lifecycle.admit_process(&token), None, "a stop revokes it");
    Ok(())
}

#[test]
fn an_unasked_exit_is_recorded_and_not_restarted() -> Result<(), LifecycleError> {
    let dir = scratch("exit");
    let lifecycle = open(&dir, "exit 3", Arc::new(FakeLys::default()));
    let made = lifecycle.create("echo", "one", "a", "op-1", 1)?;
    lifecycle.deploy(&made.id, "a")?;
    ready(&dir)?;
    wait_record(&lifecycle, &made.id, |record, running| {
        record.transitions.len() == 2 && !running
    })?;
    let record = lifecycle.record(&made.id)?;
    assert_eq!(record.state, ServiceState::Asleep);
    let note = record
        .transitions
        .last()
        .and_then(|t| t.note.clone())
        .unwrap_or_default();
    assert!(
        note.contains("exited unasked") && note.contains('3'),
        "{note}"
    );
    assert_eq!(lifecycle.list(|_| true)?["services"][0]["running"], false);
    Ok(())
}

#[test]
fn restart_live_starts_what_was_live() -> Result<(), LifecycleError> {
    let dir = scratch("boot");
    let lys = Arc::new(FakeLys::default());
    let first = open(&dir, "park", Arc::clone(&lys));
    let made = first.create("echo", "one", "a", "op-1", 1)?;
    first.deploy(&made.id, "a")?;
    ready(&dir)?;
    first.mark_live(&made.id, 2)?;
    first.shutdown();
    assert_eq!(
        first.record(&made.id)?.state,
        ServiceState::Live,
        "shutdown keeps state"
    );
    drop(first);
    let second = open(&dir, "park", lys);
    assert_eq!(second.restart_live(3), Vec::new());
    ready(&dir)?;
    assert_eq!(second.list(|_| true)?["services"][0]["running"], true);
    second.shutdown();
    Ok(())
}

#[test]
fn an_exit_after_shutdown_began_keeps_live_and_the_next_start_restarts_it()
-> Result<(), LifecycleError> {
    let dir = scratch("disconnect");
    std::fs::create_dir_all(&dir).map_err(|error| LifecycleError::Store(error.to_string()))?;
    // the worker's connection: the process leaves when its writer closes
    let connection = dir.join("connection");
    let made_fifo = std::process::Command::new("/usr/bin/mkfifo")
        .arg(&connection)
        .status()
        .map_err(|error| LifecycleError::Store(error.to_string()))?;
    assert!(made_fifo.success(), "mkfifo {}", connection.display());
    let script = format!("read _ < '{}'; exit 1", connection.display());
    let lys = Arc::new(FakeLys::default());
    let first = open(&dir, &script, Arc::clone(&lys));
    let made = first.create("echo", "one", "a", "op-1", 1)?;
    first.deploy(&made.id, "a")?;
    ready(&dir)?;
    first.mark_live(&made.id, 2)?;
    let before = first.record(&made.id)?.transitions.len();
    first.shutdown_begun();
    // the server drains: the connection closes and the process exits 1
    drop(
        std::fs::OpenOptions::new()
            .write(true)
            .open(&connection)
            .map_err(|error| LifecycleError::Store(error.to_string()))?,
    );
    wait_record(&first, &made.id, |_, running| !running)?;
    let record = first.record(&made.id)?;
    assert_eq!(record.state, ServiceState::Live, "the exit kept live");
    assert_eq!(record.transitions.len(), before, "nothing recorded");
    first.shutdown();
    drop(first);
    let second = open(&dir, "park", lys);
    assert_eq!(second.restart_live(3), Vec::new());
    ready(&dir)?;
    assert_eq!(second.list(|_| true)?["services"][0]["running"], true);
    second.shutdown();
    Ok(())
}

#[test]
fn no_placer_refuses_every_create_by_name() {
    let dir = scratch("noplacer");
    let lifecycle = Lifecycle::open(&[template("true")], dir, "net".to_owned(), None)
        .unwrap_or_else(|error| unreachable!("the registry opens: {error}"));
    assert_eq!(
        lifecycle.create("echo", "one", "a", "op-1", 1),
        Err(LifecycleError::NoPlacer)
    );
}