draupnir 0.1.9

//! **Exe backend** — run a plain executable as a child process on this host.
//!
//! This is the non-VM, non-container shape: the `exe` build-thing kind, which is
//! the single largest bucket in the appliance chain matrix (23 of 44 rows measured
//! 2026-08-15, `edda/crates/nornir-testmatrix/tests/chain_matrix.rs:181`). Stage 1
//! builds a binary; stage 2 has to actually *run* it, and until now draupnir had no
//! backend that could, so every one of those rows was a named skip.
//!
//! # A thing that keeps running is the NORMAL case
//!
//! The mistake this module is built to make impossible: waiting for the process to
//! **exit**. A server exits when it fails. `wg-appliance-core`'s container rung used
//! `Command::output()`, which blocks until termination — gunnar's `serve` printed
//! `MINTED new keypair` and then served, correctly, for 927 seconds while the rung
//! hung (`edda/crates/wg-appliance-core/src/robot.rs::run_once`, fixed 2026-08-15).
//! Its sibling rung "passed" only because that binary happened to be broken enough
//! to exit.
//!
//! So [`ExeBoot`] never waits for exit. It:
//!
//! 1. spawns the binary with both pipes captured, and returns **immediately** — the
//!    process is left RUNNING and its handle recorded, exactly what the chain
//!    driver's `ChainCtx::instance` wants;
//! 2. streams stdout and stderr on two background readers, so a process that fills
//!    a 64 KiB pipe buffer cannot deadlock against a parent that is not reading;
//! 3. [`await_marker`](ExeBoot::await_marker)s a **named string** with a deadline —
//!    and if the deadline wins, returns [`Seen`](Seen) describing *what was actually
//!    observed*, never a verdict. Deciding is the caller's job; this reports.
//!
//! Both of those readers come from [`gatling::background::Job`], the constellation's
//! sanctioned home for a raw thread (ROOT-LAW #0) — no bare `std::thread::spawn`.
//!
//! # Zero-shell
//!
//! The binary is executed directly through `std::process::Command` with an argv
//! vector. There is no shell, no `sh -c`, no string to quote and therefore nothing
//! to quote wrong — running a compiled artifact *is* the purpose of this backend,
//! and it is the only way a built binary can be run at all.

use crate::{Boot, BootSpec, Error, ImageSource, Lifecycle, Machine, PowerState, Result, Seen};

use std::collections::HashMap;
use std::io::{BufRead, BufReader};
use std::process::{Child, Command, Stdio};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// One live child process plus its streamed output.
struct Running {
    child: Child,
    /// stdout+stderr, interleaved in arrival order, appended by the two reader jobs.
    output: Arc<Mutex<String>>,
    started: Instant,
}

/// The **exe** backend: run a built binary, stream its output, wait on a marker.
#[derive(Default, Clone)]
pub struct ExeBoot {
    /// Live children keyed by [`Machine::id`], so [`Lifecycle`] and
    /// [`await_marker`](ExeBoot::await_marker) address the process `boot` started.
    live: Arc<Mutex<HashMap<String, Running>>>,
}

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

impl ExeBoot {
    /// Construct the exe backend.
    pub fn new() -> Self {
        Self::default()
    }

    /// The binary + argv this spec will run.
    pub fn program<'a>(&self, spec: &'a BootSpec) -> Result<(&'a str, &'a [String])> {
        match &spec.image {
            ImageSource::Executable { path, args } => Ok((path.as_str(), args.as_slice())),
            other => Err(Error::Spec(format!(
                "exe backend needs an ImageSource::Executable, got {other:?}"
            ))),
        }
    }

    /// **Everything captured so far** from a live child — stdout and stderr
    /// interleaved in arrival order. `None` when no live process matches `machine`
    /// (it was never started here, or has already been reaped by
    /// [`Lifecycle::power_off`]).
    ///
    /// The KVM twin of this is `KvmBoot::serial_log`; both exist so a caller holding
    /// a [`Machine`] can read the applied output without re-implementing capture.
    pub fn output(&self, machine: &Machine) -> Option<String> {
        self.live
            .lock()
            .unwrap()
            .get(&machine.id)
            .map(|r| r.output.lock().unwrap().clone())
    }

    /// **Wait for a NAMED MARKER or a deadline — never for termination.**
    ///
    /// Polls the streamed output for `marker` every `poll`, and reports what it
    /// [`Seen`]: the marker (with the line it was on), the process having exited
    /// first, or the budget elapsing with the thing still running. It returns
    /// `Ok(Seen::…)` in all three cases — a deadline is an observation, not an error,
    /// and turning it into one would be this function inventing the verdict.
    ///
    /// `Err` is reserved for "there is no such process here".
    pub fn await_marker(
        &self,
        machine: &Machine,
        marker: &str,
        budget: Duration,
        poll: Duration,
    ) -> Result<Seen> {
        let started = Instant::now();
        let deadline = started + budget;
        loop {
            // Scope the lock: never hold it across a sleep.
            let observed = {
                let mut guard = self.live.lock().unwrap();
                let running = guard.get_mut(&machine.id).ok_or_else(|| {
                    Error::Backend(format!("no live exe process for {}", machine.id))
                })?;
                let captured = running.output.lock().unwrap().clone();
                let hit = captured
                    .lines()
                    .find(|l| l.contains(marker))
                    .map(str::to_string);
                // `try_wait` reaps without blocking — the whole point: we look at
                // whether it exited, we never wait for it to.
                let exited = running
                    .child
                    .try_wait()
                    .map_err(|e| Error::Backend(format!("try_wait on {}: {e}", machine.id)))?;
                (hit, exited, captured)
            };
            let (hit, exited, captured) = observed;

            if let Some(line) = hit {
                let seen = Seen::Marker {
                    line,
                    after: started.elapsed(),
                };
                self.record("await_marker", true, machine, &seen);
                return Ok(seen);
            }
            if let Some(status) = exited {
                // Give the readers a beat to drain the pipe: a process that prints
                // its marker and immediately exits must not be reported as having
                // exited WITHOUT it. Re-check once after the drain.
                std::thread::sleep(poll);
                let captured = self
                    .live
                    .lock()
                    .unwrap()
                    .get(&machine.id)
                    .map(|r| r.output.lock().unwrap().clone())
                    .unwrap_or(captured);
                if let Some(line) = captured.lines().find(|l| l.contains(marker)) {
                    let seen = Seen::Marker {
                        line: line.to_string(),
                        after: started.elapsed(),
                    };
                    self.record("await_marker", true, machine, &seen);
                    return Ok(seen);
                }
                let seen = Seen::Exited {
                    code: status.code(),
                    tail: captured,
                    after: started.elapsed(),
                };
                self.record("await_marker", false, machine, &seen);
                return Ok(seen);
            }
            let now = Instant::now();
            if now >= deadline {
                let seen = Seen::StillRunning {
                    tail: captured,
                    waited: started.elapsed(),
                };
                self.record("await_marker", false, machine, &seen);
                return Ok(seen);
            }
            std::thread::sleep(poll.min(deadline.saturating_duration_since(now)));
        }
    }

    /// Emit the observation as a functional-status row, so a marker that never
    /// appeared is visible in nornir's matrix rather than swallowed.
    fn record(&self, check: &str, ok: bool, machine: &Machine, seen: &Seen) {
        crate::functional_status(
            "draupnir/exe",
            check,
            ok,
            &format!("`{}`: {}", machine.id, seen.detail()),
        );
    }
}

impl Boot for ExeBoot {
    /// Spawn the binary and **return while it is still running**.
    ///
    /// Both pipes are captured and drained by background readers from the moment of
    /// spawn. Draining is not a nicety: a child that writes more than the pipe
    /// buffer (64 KiB on Linux) blocks in `write` forever if the parent is not
    /// reading, which looks precisely like a hung appliance.
    fn boot(&self, spec: &BootSpec) -> Result<Machine> {
        spec.validate()?;
        let (path, args) = self.program(spec)?;
        if !std::path::Path::new(path).is_file() {
            return Err(Error::Spec(format!(
                "executable `{path}` does not exist (spec `{}`): stage 1 must produce \
                 the binary before stage 2 can run it",
                spec.name
            )));
        }

        let mut cmd = Command::new(path);
        cmd.args(args)
            .envs(spec.env.iter())
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        let mut child = cmd
            .spawn()
            .map_err(|e| Error::Backend(format!("spawning `{path}`: {e}")))?;

        let output = Arc::new(Mutex::new(String::new()));
        // Two background readers, one per pipe, interleaving into the one buffer in
        // arrival order. `gatling::background::Job` is the sanctioned home for the
        // raw thread (ROOT-LAW #0); these are blocking-IO drains, not CPU work, so
        // one dedicated thread each is the correct shape — a fork-join pool would
        // have to run to completion, and these must live as long as the process.
        // The jobs are deliberately detached (dropped unjoined): they end when the
        // pipe closes, which is when the child exits, which we do not wait for.
        for pipe in [
            child.stdout.take().map(PipeEnd::Out),
            child.stderr.take().map(PipeEnd::Err),
        ]
        .into_iter()
        .flatten()
        {
            let sink = Arc::clone(&output);
            drop(gatling::background::Job::spawn(move || {
                let reader: Box<dyn std::io::Read + Send> = match pipe {
                    PipeEnd::Out(o) => Box::new(o),
                    PipeEnd::Err(e) => Box::new(e),
                };
                for line in BufReader::new(reader).lines().map_while(std::result::Result::ok) {
                    let mut buf = sink.lock().unwrap();
                    buf.push_str(&line);
                    buf.push('\n');
                }
            }));
        }

        let id = format!("exe-{}-{}", spec.name, child.id());
        let machine = Machine::started(&id, spec);
        self.live.lock().unwrap().insert(
            id,
            Running {
                child,
                output,
                started: Instant::now(),
            },
        );
        Ok(machine)
    }
}

/// Which pipe a reader job was handed (the two have different concrete types but
/// identical handling, so they share one drain loop rather than a copied twin).
enum PipeEnd {
    Out(std::process::ChildStdout),
    Err(std::process::ChildStderr),
}

impl Lifecycle for ExeBoot {
    /// A process cannot be re-started in place — the honest answer is "boot again",
    /// the same answer the KVM backend gives.
    fn power_on(&self, machine: &Machine) -> Result<()> {
        let _ = machine;
        Err(Error::Unsupported(
            "an exe instance is fire-and-forget: re-run by calling draupnir::boot() again".into(),
        ))
    }

    /// Kill the child and drop it from the registry. Idempotent-ish: a process that
    /// has already exited is reaped rather than reported as an error.
    fn power_off(&self, machine: &Machine) -> Result<()> {
        let mut running = self
            .live
            .lock()
            .unwrap()
            .remove(&machine.id)
            .ok_or_else(|| Error::Backend(format!("no live exe process for {}", machine.id)))?;
        // `kill` on an already-exited child is an error on some platforms; we only
        // care that it is not running afterwards.
        let _ = running.child.kill();
        let _ = running.child.wait();
        Ok(())
    }

    /// `On` while the child is running, `Off` once it has exited, `Unknown` for an
    /// id this backend never started.
    fn status(&self, machine: &Machine) -> Result<PowerState> {
        let mut guard = self.live.lock().unwrap();
        let Some(running) = guard.get_mut(&machine.id) else {
            return Ok(PowerState::Unknown);
        };
        match running.child.try_wait() {
            Ok(None) => Ok(PowerState::On),
            Ok(Some(_)) => Ok(PowerState::Off),
            Err(e) => Err(Error::Backend(format!("try_wait on {}: {e}", machine.id))),
        }
    }
}

impl ExeBoot {
    /// How long the child named by `machine` has been running, or `None` if it is
    /// not live here.
    pub fn uptime(&self, machine: &Machine) -> Option<Duration> {
        self.live
            .lock()
            .unwrap()
            .get(&machine.id)
            .map(|r| r.started.elapsed())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Backend, BootOrder};

    /// A binary that exists on every box this ever runs on, and prints predictably.
    const TRUE_BIN: &str = "/bin/sh";

    fn sh(name: &str, script: &str) -> BootSpec {
        BootSpec::exe(name, TRUE_BIN, ["-c", script])
    }

    #[test]
    fn an_exe_spec_validates_and_carries_its_argv_in_the_image() {
        let spec = BootSpec::exe("hello", "/bin/echo", ["hi", "there"]);
        spec.validate().unwrap();
        assert_eq!(spec.backend, Backend::Exe);
        let (path, args) = ExeBoot::new().program(&spec).unwrap();
        assert_eq!(path, "/bin/echo");
        assert_eq!(args, ["hi".to_string(), "there".to_string()]);
        // The exe's arguments are NOT the container `cmd` knob, which stays
        // container-only and must remain empty here or validate() would reject it.
        assert!(spec.cmd.is_empty());
    }

    #[test]
    fn an_empty_executable_path_is_rejected_by_name() {
        // RED-when-broken: drop the `require("executable path", ..)` arm in
        // BootSpec::validate and this passes.
        let spec = BootSpec::exe("blank", "   ", Vec::<String>::new());
        let err = spec.validate().unwrap_err();
        assert!(
            format!("{err}").contains("executable path"),
            "the failure NAMES the field: {err}"
        );
    }

    #[test]
    fn a_missing_binary_fails_by_name_before_anything_is_spawned() {
        // The stage-1-did-not-produce-it case must not read as "it did not come up".
        // RED-when-broken: remove the is_file() guard in Boot::boot and the error
        // becomes an opaque OS "No such file or directory" from spawn.
        let spec = BootSpec::exe("ghost", "/nonexistent/never/built", Vec::<String>::new());
        let err = ExeBoot::new().boot(&spec).unwrap_err();
        let msg = format!("{err}");
        assert!(matches!(err, Error::Spec(_)), "a missing artifact is a SPEC fault: {msg}");
        assert!(msg.contains("/nonexistent/never/built"), "names the path: {msg}");
        assert!(msg.contains("stage 1"), "names whose job it was: {msg}");
    }

    #[test]
    fn a_medium_or_boot_order_on_an_exe_spec_is_rejected() {
        // An exe has no drive tray and no firmware; both would vanish silently.
        let m = BootSpec::exe("x", "/bin/true", Vec::<String>::new()).with_medium("/x.iso");
        assert!(matches!(m.validate(), Err(Error::Spec(_))), "medium on exe rejected");
        let b = BootSpec::exe("x", "/bin/true", Vec::<String>::new())
            .with_boot_order(BootOrder::Disk);
        assert!(matches!(b.validate(), Err(Error::Spec(_))), "boot order on exe rejected");
    }

    #[test]
    fn a_marker_printed_by_a_process_that_then_serves_forever_is_seen_and_the_process_survives() {
        // THE headline case, and the one that hung a rung for 927 s: a thing that
        // announces itself and then keeps running. The marker must be observed while
        // it is still alive, and it must STILL be alive afterwards.
        //
        // RED-when-broken: make await_marker wait for exit (e.g. `child.wait()`
        // instead of `try_wait`) and this test hangs until the harness kills it —
        // which is exactly the 927 s failure, reproduced.
        let backend = ExeBoot::new();
        let spec = sh("server", "echo APPLIANCE-READY; sleep 30");
        let m = backend.boot(&spec).unwrap();

        let seen = backend
            .await_marker(&m, "APPLIANCE-READY", Duration::from_secs(10), Duration::from_millis(20))
            .unwrap();
        assert!(seen.saw_marker(), "marker seen, got {seen:?}");
        match &seen {
            Seen::Marker { line, .. } => assert_eq!(line, "APPLIANCE-READY"),
            other => panic!("expected the matched LINE to be carried, got {other:?}"),
        }
        // It is still serving — the whole point.
        assert_eq!(backend.status(&m).unwrap(), PowerState::On, "still running");
        assert!(backend.uptime(&m).is_some());
        backend.power_off(&m).unwrap();
        assert_eq!(backend.status(&m).unwrap(), PowerState::Unknown, "reaped");
    }

    #[test]
    fn a_deadline_returns_what_was_seen_rather_than_a_verdict() {
        // The thing runs, prints something else, and never says the magic word. The
        // budget elapsing must report StillRunning WITH the output — not an error,
        // and not a fabricated "not ready".
        let backend = ExeBoot::new();
        let spec = sh("quiet", "echo SOMETHING-ELSE; sleep 30");
        let m = backend.boot(&spec).unwrap();
        let seen = backend
            .await_marker(&m, "NEVER-PRINTED", Duration::from_millis(600), Duration::from_millis(20))
            .unwrap();
        match &seen {
            Seen::StillRunning { tail, .. } => {
                assert!(tail.contains("SOMETHING-ELSE"), "carries real output: {tail:?}");
            }
            other => panic!("expected StillRunning, got {other:?}"),
        }
        assert!(!seen.saw_marker());
        assert!(seen.detail().contains("SOMETHING-ELSE"), "detail quotes it: {}", seen.detail());
        backend.power_off(&m).unwrap();
    }

    #[test]
    fn a_process_that_dies_before_the_marker_reports_the_exit_and_its_stderr() {
        // The real failure for a server. The code and the output must both survive,
        // because "it exited 1 saying 'address in use'" and "it never came up" are
        // different bugs.
        let backend = ExeBoot::new();
        let spec = sh("dies", "echo 'bind: address already in use' >&2; exit 3");
        let m = backend.boot(&spec).unwrap();
        let seen = backend
            .await_marker(&m, "APPLIANCE-READY", Duration::from_secs(10), Duration::from_millis(20))
            .unwrap();
        match &seen {
            Seen::Exited { code, tail, .. } => {
                assert_eq!(*code, Some(3), "the exit code survives");
                assert!(tail.contains("address already in use"), "STDERR is captured: {tail:?}");
            }
            other => panic!("expected Exited, got {other:?}"),
        }
        assert!(seen.detail().contains("EXITED"), "{}", seen.detail());
        backend.power_off(&m).unwrap();
    }

    #[test]
    fn a_marker_printed_immediately_before_exit_is_still_seen() {
        // The drain race: print the marker and exit in the same breath. Reporting
        // "exited without the marker" here would be a false RED.
        //
        // NOT RED-WHEN-BROKEN, and said out loud rather than left implied: deleting
        // the post-exit drain-and-recheck in `await_marker`'s `exited` arm does NOT
        // fail this test. Measured 2026-08-15 — the sabotage was applied and the test
        // stayed green across 5 runs, both with this one-line script and with a
        // 3000-line variant written to force the reader to lag. The reason is
        // structural: `await_marker` reads the captured buffer BEFORE it calls
        // `try_wait` in the same iteration, and on this box the reader job has always
        // drained the pipe by the time the exit is observed, so the fast path wins
        // every time and the re-check never fires.
        //
        // What this test therefore DOES prove: a marker printed immediately before
        // exit is reported as `Seen::Marker`, not `Seen::Exited`. What it does NOT
        // prove: that the drain re-check works, or is needed. That branch is
        // defensive code with no proof behind it — treat it as unproven, not as
        // covered.
        let backend = ExeBoot::new();
        let spec = sh("flash", "echo APPLIANCE-READY");
        let m = backend.boot(&spec).unwrap();
        let seen = backend
            .await_marker(&m, "APPLIANCE-READY", Duration::from_secs(10), Duration::from_millis(50))
            .unwrap();
        assert!(seen.saw_marker(), "marker printed just before exit is still seen: {seen:?}");
        backend.power_off(&m).unwrap();
    }

    #[test]
    fn env_from_the_spec_reaches_the_process() {
        // Applied output, not a round-trip: the child prints what it actually got.
        let backend = ExeBoot::new();
        let spec = BootSpec::exe("envy", TRUE_BIN, ["-c", "echo GOT=$DRAUPNIR_PROBE"])
            .with_env("DRAUPNIR_PROBE", "42");
        let m = backend.boot(&spec).unwrap();
        let seen = backend
            .await_marker(&m, "GOT=42", Duration::from_secs(10), Duration::from_millis(20))
            .unwrap();
        assert!(seen.saw_marker(), "env reached the child: {seen:?}");
        backend.power_off(&m).unwrap();
    }

    #[test]
    fn a_chatty_process_does_not_deadlock_on_a_full_pipe() {
        // > 64 KiB of output before the marker. Without the streaming readers the
        // child blocks in write() forever and this never returns.
        let backend = ExeBoot::new();
        let spec = sh(
            "chatty",
            "i=0; while [ $i -lt 4000 ]; do echo \"filler line $i 0123456789012345678901234567890123456789\"; i=$((i+1)); done; echo APPLIANCE-READY; sleep 5",
        );
        let m = backend.boot(&spec).unwrap();
        let seen = backend
            .await_marker(&m, "APPLIANCE-READY", Duration::from_secs(20), Duration::from_millis(20))
            .unwrap();
        assert!(seen.saw_marker(), "marker after >64KiB of output: {seen:?}");
        let captured = backend.output(&m).unwrap();
        assert!(captured.len() > 64 * 1024, "really did exceed a pipe buffer: {}", captured.len());
        backend.power_off(&m).unwrap();
    }

    #[test]
    fn awaiting_a_machine_this_backend_never_started_is_an_error_not_a_verdict() {
        let backend = ExeBoot::new();
        let m = Machine {
            id: "exe-not-live-1".into(),
            spec_name: "x".into(),
            backend: Backend::Exe,
            power: PowerState::Unknown,
        };
        assert!(matches!(
            backend.await_marker(&m, "x", Duration::from_millis(10), Duration::from_millis(5)),
            Err(Error::Backend(_))
        ));
        assert_eq!(backend.status(&m).unwrap(), PowerState::Unknown);
        assert!(backend.output(&m).is_none());
    }
}