crates/jera/src/boot.rs
//! **Boot runtime** — the engine-agnostic lifecycle shared by the container / VM /
//! exe runners, and the job-bound [`BootJobs`] manager: the reusable "start X as a
//! tracked job" engine korp and nornir both drive.
//!
//! A boot is one of three [modes](BootMode):
//! 1. [`Exe`](BootMode::Exe) — spawn a *built binary* as a child process with an
//! interactive console ([`ExeBootRunner`]). (Spawning a built exe is a legit
//! process; the zero-shell rule targets *container/VM control*, not this.)
//! 2. [`Container`](BootMode::Container) — a [`crate::container::ContainerSpec`]
//! brought up over the podman/Docker REST API THROUGH draupnir's ONE OCI engine
//! ([`crate::container::DraupnirContainerBootRunner`]).
//! 3. [`Vm`](BootMode::Vm) — a [`crate::vm::BootSpec`] booted as a KVM guest
//! ([`crate::vm::VmBootRunner`]).
//!
//! The two directions of the lifecycle are wired through [`BootProc`]:
//! **kill the job → kill the boot** ([`BootProc::kill`]) and **the boot dies → the
//! job terminates** (the next [`BootProc::poll`] goes terminal). Side-work sits
//! behind the [`BootRunner`]/[`BootProc`] seams so tests inject a
//! [`FakeBootRunner`] and drive the WHOLE lifecycle with no real process/container/VM.
use std::io::{BufRead, BufReader, Write};
use std::path::PathBuf;
use std::process::{Child, ChildStdin, Command, Stdio};
use std::sync::{Arc, Mutex};
use crate::container::ContainerSpec;
use crate::jobs::{kind, Health, JobHandle, JobSink, ReportRegistry, ReportSink, Reporter};
use crate::vm::BootSpec;
// ---------------------------------------------------------------------------
// Mode + status
// ---------------------------------------------------------------------------
// `BootMode` lives with the ledger's policy (lane B4: the ledger builds without
// the runners); it is re-exported here at its old path.
pub use crate::policy::BootMode;
/// Live status of a boot. The lifecycle is `Booting → Running → (BootedOk | Failed
/// | Killed)`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BootStatus {
/// Spawn issued; the target is coming up.
Booting,
/// The target is up and running.
Running,
/// The target booted and then exited cleanly (exit 0).
BootedOk,
/// The target failed to boot or exited non-zero. Carries a reason.
Failed(String),
/// The boot was killed (via the job) — or died by signal after a kill.
Killed,
}
impl BootStatus {
/// Whether the boot has reached a terminal state.
pub fn is_terminal(&self) -> bool {
matches!(self, BootStatus::BootedOk | BootStatus::Failed(_) | BootStatus::Killed)
}
/// Stable label for the UI / state_json.
pub fn as_str(&self) -> &'static str {
match self {
BootStatus::Booting => "booting",
BootStatus::Running => "running",
BootStatus::BootedOk => "booted_ok",
BootStatus::Failed(_) => "failed",
BootStatus::Killed => "killed",
}
}
}
// ---------------------------------------------------------------------------
// Target + seams
// ---------------------------------------------------------------------------
/// A bootable target — the input to a boot. Its variant fixes the [`BootMode`].
#[derive(Debug, Clone)]
pub enum BootTarget {
/// A built binary: the exe path + args to run it with.
Exe {
/// The target label (the job's `target`).
name: String,
/// The built executable to spawn.
exe: PathBuf,
/// Args passed to the exe.
args: Vec<String>,
},
/// A container target (image + run overrides).
Container(ContainerSpec),
/// A VM target.
Vm(BootSpec),
}
impl BootTarget {
/// The human name of the target (the job's `target` label).
pub fn name(&self) -> &str {
match self {
BootTarget::Exe { name, .. } => name,
BootTarget::Container(c) => &c.name,
BootTarget::Vm(v) => &v.name,
}
}
/// **Mode selection** — the boot mode is fixed by the target's type.
pub fn mode(&self) -> BootMode {
match self {
BootTarget::Exe { .. } => BootMode::Exe,
BootTarget::Container(_) => BootMode::Container,
BootTarget::Vm(_) => BootMode::Vm,
}
}
}
/// A live boot's process handle. [`poll`](Self::poll) reports death,
/// [`kill`](Self::kill) tears the boot down.
pub trait BootProc: Send {
/// The current status. MUST transition to terminal the moment the underlying
/// process/container/VM dies on its own.
fn poll(&mut self) -> BootStatus;
/// Kill the underlying process-group / container / VM. Idempotent.
fn kill(&mut self);
/// **Stop and let it FLUSH**: a graceful power-down with `grace` to comply
/// before the hard kill. Default: [`kill`](BootProc::kill) — right for a
/// process or a container. The KVM proc overrides it with draupnir's
/// `KvmBoot::shutdown` (monitor power-down → ACPI → SIGKILL), because an install
/// guest shot with SIGKILL loses QEMU's writeback cache and the second boot reads
/// a disk that was never fully written (measured 2026-08-16). Idempotent.
fn shutdown(&mut self, _grace: std::time::Duration) {
self.kill();
}
/// New log lines since the last drain.
fn drain_log(&mut self) -> Vec<String>;
/// Send a line to the boot's stdin (the Exe console PROMPT). Default no-op.
fn send_stdin(&mut self, _line: &str) {}
/// **Exec `argv` inside an already-running boot** — the live `podman exec <id>
/// <argv…>` control primitive a Skidbladnir start/stop/exec lifecycle-swap needs:
/// after a detached container boot hands back a long-lived proc, run a command
/// *into* the running container and read back its
/// [`ExecOutcome`](draupnir::container::ExecOutcome) (exit code + captured
/// stdout/stderr). A **non-zero** exec exit is carried in
/// [`ExecOutcome::exit_code`](draupnir::container::ExecOutcome::exit_code), **not**
/// an `Err`; only a guard failure or an engine/connect error is `Err`.
///
/// **Default**: an honest error — a plain Exe/VM boot has no container to exec
/// into. Only [`crate::container::DraupnirContainerBootRunner`]'s detached
/// container proc overrides it, forwarding to draupnir's ONE OCI engine
/// ([`draupnir::container::ContainerControl::exec`]). Additive (a provided method,
/// like [`send_stdin`](Self::send_stdin)) — every existing [`BootProc`] impl keeps
/// compiling unchanged and simply reports "not exec-capable".
fn exec(&self, _argv: &[&str]) -> Result<draupnir::container::ExecOutcome, String> {
Err("this boot proc does not support exec (only a detached container proc routed through draupnir does)".to_string())
}
}
/// The seam that spawns a boot for a [`BootTarget`].
pub trait BootRunner: Send + Sync {
/// Spawn the target; `Ok` carries the live [`BootProc`], `Err` a human reason.
fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String>;
/// Like [`spawn`](Self::spawn), but also hands the runner a [`Reporter`] it MAY
/// use to emit [`crate::jobs::Health`] as the target's lifecycle progresses —
/// e.g. the container engine reports `Starting -> Alive -> Gone`. Default:
/// ignore the reporter and defer to [`spawn`](Self::spawn) — additive, so every
/// existing [`BootRunner`] implementor keeps compiling unchanged; only
/// [`crate::container::DraupnirContainerBootRunner`] overrides it today (Exe/VM
/// boots have no extra health signal beyond [`BootStatus`] itself).
fn spawn_with_reporter(
&self,
target: &BootTarget,
_reporter: Reporter,
) -> Result<Box<dyn BootProc>, String> {
self.spawn(target)
}
}
/// A [`BootRunner`] that always fails with a fixed reason — the honest stand-in when
/// the real engine is unavailable (no container socket / no KVM backend), never a
/// silent shell fallback.
pub struct UnavailableRunner {
reason: String,
}
impl UnavailableRunner {
/// A runner whose every spawn fails with `reason`.
pub fn new(reason: String) -> Self {
UnavailableRunner { reason }
}
}
impl BootRunner for UnavailableRunner {
fn spawn(&self, _target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
Err(self.reason.clone())
}
}
// ---------------------------------------------------------------------------
// Exe mode — spawn the built binary as a child + console (REAL).
// ---------------------------------------------------------------------------
/// A shared, append-only log buffer the reader threads push into and the caller drains.
type LogBuf = Arc<Mutex<Vec<String>>>;
/// A live child process (Exe mode).
struct ExeProc {
child: Child,
stdin: Option<ChildStdin>,
pid: u32,
log: LogBuf,
killed: bool,
terminal: Option<BootStatus>,
}
impl ExeProc {
fn compute(&mut self) -> BootStatus {
if let Some(t) = &self.terminal {
return t.clone();
}
match self.child.try_wait() {
Ok(Some(status)) => {
let t = if self.killed {
BootStatus::Killed
} else if status.success() {
BootStatus::BootedOk
} else {
BootStatus::Failed(format!("exited {status}"))
};
self.terminal = Some(t.clone());
t
}
Ok(None) => BootStatus::Running,
Err(e) => {
let t = BootStatus::Failed(format!("wait failed: {e}"));
self.terminal = Some(t.clone());
t
}
}
}
}
/// **Nothing leaves this type unreaped.** A `std::process::Child` that is merely
/// DROPPED is detached, not waited on: the kernel keeps the exit status around
/// forever, so the pid slot is burned for the life of the parent. A zombie holds no
/// RSS, no CPU and no fds, so it appears in neither `top` nor an OOM dump's memory
/// accounting — the only thing it exhausts is `TasksMax` / `pid_max`, and the failure
/// mode is not a slow box but one that abruptly cannot fork.
///
/// [`kill`](BootProc::kill) already reaps (`child.wait()`), and [`poll`](BootProc::poll)
/// reaps through `try_wait()` once the child has exited. The hole was the handle that
/// is simply released — a `Box<dyn BootProc>` dropped without either — and in the
/// normal [`BootJobs`] flow that cannot happen, because `job.proc = None` runs only in
/// the TERMINAL branch. It happens when the OWNER goes away with a boot still live.
///
/// LAW 5 — this is the ONE place the drop path reaps, not a `try_wait()` sprinkled at
/// every site that might release a proc.
impl Drop for ExeProc {
fn drop(&mut self) {
if self.terminal.is_some() {
return; // compute() or kill() already waited; nothing is outstanding
}
// Already exited, just never polled — this is the common leak and a plain
// wait is enough. No signal: the process is gone, only its status remains.
if matches!(self.child.try_wait(), Ok(Some(_))) {
return;
}
// Still running, and the handle that could poll or kill it is being
// destroyed. Leaving it is not "letting it finish" — it is an orphan in its
// own process group that nothing in this program can ever stop or reap, and
// it becomes a zombie the moment it exits anyway. So take the same route
// `kill` does: SIGKILL the whole GROUP (grandchildren included), then wait.
self.stdin = None;
unsafe {
libc::kill(-(self.pid as i32), libc::SIGKILL);
}
let _ = self.child.wait();
}
}
impl BootProc for ExeProc {
fn poll(&mut self) -> BootStatus {
self.compute()
}
fn kill(&mut self) {
if self.terminal.is_some() {
return;
}
// Drop stdin (child sees EOF), then SIGKILL the whole process GROUP (the
// child was spawned in its own group = its pid), so grandchildren die too.
self.stdin = None;
unsafe {
libc::kill(-(self.pid as i32), libc::SIGKILL);
}
let _ = self.child.wait();
self.killed = true;
self.terminal = Some(BootStatus::Killed);
}
fn drain_log(&mut self) -> Vec<String> {
let mut g = self.log.lock().unwrap();
std::mem::take(&mut *g)
}
fn send_stdin(&mut self, line: &str) {
if let Some(stdin) = self.stdin.as_mut() {
let _ = stdin.write_all(line.as_bytes());
let _ = stdin.write_all(b"\n");
let _ = stdin.flush();
}
}
}
/// The Exe boot runner: spawns the built binary as a child in its OWN process group
/// (so a job-kill SIGKILLs the whole group), pipes stdin for the console PROMPT, and
/// streams stdout/stderr into the job log via reader threads.
pub struct ExeBootRunner;
impl ExeBootRunner {
fn spawn_process(program: &std::ffi::OsStr, args: &[String]) -> Result<Box<dyn BootProc>, String> {
use std::os::unix::process::CommandExt;
let mut cmd = Command::new(program);
cmd.args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped());
cmd.process_group(0);
let mut child = cmd.spawn().map_err(|e| format!("spawn `{}`: {e}", program.to_string_lossy()))?;
let pid = child.id();
let log: LogBuf = Arc::new(Mutex::new(Vec::new()));
if let Some(out) = child.stdout.take() {
spawn_reader(out, "", Arc::clone(&log));
}
if let Some(err) = child.stderr.take() {
spawn_reader(err, "err: ", Arc::clone(&log));
}
let stdin = child.stdin.take();
Ok(Box::new(ExeProc { child, stdin, pid, log, killed: false, terminal: None }))
}
}
impl BootRunner for ExeBootRunner {
fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
match target {
BootTarget::Exe { exe, args, .. } => Self::spawn_process(exe.as_os_str(), args),
other => Err(format!("ExeBootRunner cannot boot a {} target", other.mode().as_str())),
}
}
}
/// Stream one child pipe into the shared log, line by line, with a `prefix`.
fn spawn_reader<R: std::io::Read + Send + 'static>(reader: R, prefix: &'static str, log: LogBuf) {
std::thread::spawn(move || {
let buf = BufReader::new(reader);
for line in buf.lines() {
match line {
Ok(l) => log.lock().unwrap().push(format!("{prefix}{l}")),
Err(_) => break,
}
}
});
}
// ---------------------------------------------------------------------------
// Dispatch runner — routes a BootTarget to its mode's runner.
// ---------------------------------------------------------------------------
/// The production [`BootRunner`]: routes each [`BootTarget`] to its mode's runner —
/// Exe → [`ExeBootRunner`], Container → the zero-shell bollard runner (or an honest
/// [`UnavailableRunner`]), Vm → the [`crate::vm::VmBootRunner`] seam.
pub struct DispatchBootRunner {
exe: ExeBootRunner,
container: Box<dyn BootRunner>,
vm: Box<dyn crate::vm::VmBootRunner>,
}
impl Default for DispatchBootRunner {
fn default() -> Self {
DispatchBootRunner {
exe: ExeBootRunner,
container: crate::container::connect_default_runner(),
vm: crate::vm::default_vm_runner(),
}
}
}
impl BootRunner for DispatchBootRunner {
fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
match target {
BootTarget::Exe { .. } => self.exe.spawn(target),
BootTarget::Container(_) => self.container.spawn(target),
BootTarget::Vm(vm) => self.vm.boot(vm),
}
}
fn spawn_with_reporter(
&self,
target: &BootTarget,
reporter: Reporter,
) -> Result<Box<dyn BootProc>, String> {
match target {
// Only the container path has a real Health signal today; route it
// through so DraupnirContainerBootRunner's override actually fires.
BootTarget::Container(_) => self.container.spawn_with_reporter(target, reporter),
BootTarget::Exe { .. } | BootTarget::Vm(_) => self.spawn(target),
}
}
}
// ---------------------------------------------------------------------------
// The boot-job manager — job list integration + bidirectional lifecycle.
// ---------------------------------------------------------------------------
/// One live boot job: the durable job record's id + the observable status/log, and
/// the live [`BootProc`] handle + the [`JobHandle`] that writes the ledger row.
pub struct BootJob {
/// The job id (the ledger key).
pub job_id: String,
/// The target name.
pub target: String,
/// The boot mode.
pub mode: BootMode,
/// Live status.
pub status: BootStatus,
/// The boot log, newest last.
pub log: Vec<String>,
proc: Option<Box<dyn BootProc>>,
handle: Option<JobHandle>,
}
impl BootJob {
/// The observable JSON for state_json / a robot assert.
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"job_id": self.job_id,
"target": self.target,
"mode": self.mode.as_str(),
"status": self.status.as_str(),
"terminal": self.status.is_terminal(),
"log_lines": self.log.len(),
})
}
}
/// Manages live boot jobs: presses a boot (creating a job), polls each boot's
/// lifecycle both directions, and kills on demand. This is the reusable
/// "start X as a tracked job" engine (nornir's Build-tab test-boot state; korp's
/// container-as-a-job runner).
pub struct BootJobs {
runner: Arc<dyn BootRunner>,
sink: JobSink,
workspace: String,
jobs: Vec<BootJob>,
/// Where each boot job's [`Reporter`] emits — [`ReportSink::noop`] (nowhere)
/// unless [`with_reports`](Self::with_reports) was called.
report_sink: ReportSink,
/// The read side of the same registry `report_sink` writes into — `Some` only
/// once [`with_reports`](Self::with_reports) is called, so [`Self::state_json`]
/// can fold each job's latest reports in.
report_registry: Option<ReportRegistry>,
}
impl BootJobs {
/// A manager wired to the real [`DispatchBootRunner`] and the given ledger `sink`
/// (use [`JobSink::noop`] when no ledger is configured).
pub fn new(sink: JobSink, workspace: impl Into<String>) -> Self {
Self::with_runner(Arc::new(DispatchBootRunner::default()), sink, workspace)
}
/// A manager with an injected [`BootRunner`] — the test seam.
pub fn with_runner(runner: Arc<dyn BootRunner>, sink: JobSink, workspace: impl Into<String>) -> Self {
BootJobs {
runner,
sink,
workspace: workspace.into(),
jobs: Vec::new(),
report_sink: ReportSink::noop(),
report_registry: None,
}
}
/// Wire live [`crate::jobs::JobReport`] emission (write) + reads through
/// `store`'s report registry — OPT-IN, builder style, chain onto
/// [`new`](Self::new)/[`with_runner`](Self::with_runner). Without this call,
/// boot jobs still work exactly as before; they just emit no reports and
/// [`Self::state_json`] carries no `"reports"` key.
pub fn with_reports(mut self, store: &crate::jobs::JobStore) -> Self {
self.report_sink = store.report_sink();
self.report_registry = Some(store.report_registry());
self
}
/// The live boot jobs (newest last).
pub fn jobs(&self) -> &[BootJob] {
&self.jobs
}
/// A clone of the injected [`BootRunner`] `Arc`.
pub fn runner_arc(&self) -> Arc<dyn BootRunner> {
Arc::clone(&self.runner)
}
/// Whether a non-terminal boot job exists.
pub fn any_running(&self) -> bool {
self.jobs.iter().any(|j| !j.status.is_terminal())
}
/// The observable JSON — the boot jobs. Each job's object carries a `"reports"`
/// key (its latest [`crate::jobs::JobReports`]) when [`with_reports`](Self::with_reports)
/// was called; omitted otherwise (unchanged shape for callers that never opt in).
pub fn state_json(&self) -> serde_json::Value {
serde_json::json!({
"boot_jobs": self.jobs.iter().map(|j| self.job_state_json(j)).collect::<Vec<_>>(),
"running": self.jobs.iter().filter(|j| !j.status.is_terminal()).count(),
})
}
/// [`BootJob::state_json`] plus a `"reports"` key folded in from
/// `self.report_registry`, when wired.
fn job_state_json(&self, job: &BootJob) -> serde_json::Value {
let mut v = job.state_json();
if let Some(registry) = &self.report_registry {
if let Some(reports) = registry.lock().unwrap_or_else(|e| e.into_inner()).get(&job.job_id) {
v["reports"] = serde_json::to_value(reports).unwrap_or(serde_json::Value::Null);
}
}
v
}
/// **Press boot.** Spawn `target` through the [`BootRunner`], open a `running`
/// job (kind = the mode's boot kind), and register the live boot job. On a spawn
/// failure the job is opened and immediately failed. Returns the new job id.
/// When [`with_reports`](Self::with_reports) was called, the job's [`Reporter`]
/// is passed to the runner too (see [`BootRunner::spawn_with_reporter`]) — this
/// is how a container boot emits `Starting`/`Alive`/`Gone`/`Failed`.
pub fn boot(&mut self, target: BootTarget) -> String {
let mode = target.mode();
let name = target.name().to_string();
let handle = JobHandle::start(
self.sink.clone(),
mode.job_kind(),
&name,
&self.workspace,
Self::open_detail(&target),
)
.with_reports(self.report_sink.clone());
let job_id = handle.job_id().to_string();
let reporter = handle.reporter();
match self.runner.spawn_with_reporter(&target, reporter.clone()) {
Ok(proc) => {
crate::jobs::selftest::emit(
"jera-boot",
&format!("boot_{}_{name}", mode.as_str()),
true,
&format!("boot '{name}' ({}) started", mode.as_str()),
);
self.jobs.push(BootJob {
job_id: job_id.clone(),
target: name,
mode,
status: BootStatus::Booting,
log: Vec::new(),
proc: Some(proc),
handle: Some(handle),
});
}
Err(reason) => {
crate::jobs::selftest::emit(
"jera-boot",
&format!("boot_{}_{name}", mode.as_str()),
false,
&format!("boot '{name}' ({}) FAILED to start: {reason}", mode.as_str()),
);
reporter.health(Health::Failed);
handle.fail(&anyhow::anyhow!("{reason}"));
self.jobs.push(BootJob {
job_id: job_id.clone(),
target: name,
mode,
status: BootStatus::Failed(reason),
log: Vec::new(),
proc: None,
handle: None,
});
}
}
job_id
}
/// The opening ledger-row detail for a boot job. For a **container** target it
/// folds in the korp-relevant knobs — the `zone` label, the published `ports`, and
/// the resource caps (`cpus`/`mem_mb`, `null` ⇒ unconstrained = all host cores) —
/// so a poller of the [`crate::jobs::JobRecord`] sees *what* zone-infra this job is
/// bringing up. Non-container targets keep the pre-existing `{ "mode": … }` shape.
fn open_detail(target: &BootTarget) -> serde_json::Value {
match target {
BootTarget::Container(spec) => serde_json::json!({
"mode": BootMode::Container.as_str(),
"image": spec.image,
"zone": spec.zone,
"ports": spec.ports,
"cpus": spec.cpus,
"mem_mb": spec.mem_mb,
}),
other => serde_json::json!({ "mode": other.mode().as_str() }),
}
}
/// **The live status of a boot job by id** — the korp-facing read of the
/// submit→poll loop. `None` if no such job. (`state_json` carries the whole
/// roster; this is the single-job convenience a caller polls after [`boot`](Self::boot).)
pub fn status_of(&self, job_id: &str) -> Option<BootStatus> {
self.jobs.iter().find(|j| j.job_id == job_id).map(|j| j.status.clone())
}
/// **Wait until a submitted boot job is RUNNING (or terminal), bounded.** The
/// "wait-for-running helper" korp calls after [`boot`](Self::boot): it [`poll`](Self::poll)s
/// the whole roster on `interval` until the job named by `job_id` reports
/// [`BootStatus::Running`] — meaning the container process is up through
/// draupnir's ONE OCI engine — or reaches a terminal state (`Failed`/`Killed`/
/// `BootedOk`), or `timeout` elapses. Returns the last observed [`BootStatus`]
/// (`Failed` naming the reason if the boot never came up; the caller then surfaces
/// it and does NOT proceed). This confirms the *container* is live; an app-level
/// readiness gate (e.g. "the published port accepts") is the caller's next step.
///
/// Returns `BootStatus::Failed("no such job …")` if `job_id` is unknown.
pub fn wait_until_running(
&mut self,
job_id: &str,
timeout: std::time::Duration,
interval: std::time::Duration,
) -> BootStatus {
let deadline = std::time::Instant::now() + timeout;
loop {
self.poll();
match self.status_of(job_id) {
None => return BootStatus::Failed(format!("no such job {job_id}")),
Some(BootStatus::Running) => return BootStatus::Running,
Some(s) if s.is_terminal() => return s,
Some(_) => {}
}
if std::time::Instant::now() >= deadline {
return self.status_of(job_id).unwrap_or(BootStatus::Booting);
}
std::thread::sleep(interval);
}
}
/// **Poll every boot both directions.** Drain each boot's log, read its status;
/// when a boot has DIED on its own, finish/fail the job.
pub fn poll(&mut self) {
for job in &mut self.jobs {
if job.status.is_terminal() {
continue;
}
let Some(proc) = job.proc.as_mut() else { continue };
let new_lines = proc.drain_log();
if !new_lines.is_empty() {
job.log.extend(new_lines);
}
let status = proc.poll();
if status != job.status {
job.status = status.clone();
}
if status.is_terminal() {
Self::terminate_job(job, &status);
}
}
}
/// **Kill a boot job by id** — the "kill the job → kill the boot" direction.
pub fn kill(&mut self, job_id: &str) -> bool {
let Some(job) = self.jobs.iter_mut().find(|j| j.job_id == job_id) else {
return false;
};
if job.status.is_terminal() {
return false;
}
if let Some(proc) = job.proc.as_mut() {
proc.kill();
let drained = proc.drain_log();
job.log.extend(drained);
}
job.status = BootStatus::Killed;
Self::terminate_job(job, &BootStatus::Killed);
true
}
/// Send a line to an Exe boot's console (the PROMPT).
pub fn send_stdin(&mut self, job_id: &str, line: &str) {
if let Some(job) = self.jobs.iter_mut().find(|j| j.job_id == job_id) {
if let Some(proc) = job.proc.as_mut() {
proc.send_stdin(line);
job.log.push(format!("> {line}"));
}
}
}
/// Write the terminal ledger row for a boot that reached `status`, drop the live
/// handle.
fn terminate_job(job: &mut BootJob, status: &BootStatus) {
let Some(handle) = job.handle.take() else { return };
let detail = serde_json::json!({
"mode": job.mode.as_str(),
"status": status.as_str(),
"log_tail": job.log.iter().rev().take(20).rev().cloned().collect::<Vec<_>>(),
});
match status {
BootStatus::BootedOk => handle.finish(detail, ""),
BootStatus::Killed => handle.fail_with_detail(detail),
BootStatus::Failed(reason) => {
let mut d = detail;
d["reason"] = serde_json::json!(reason);
handle.fail_with_detail(d);
}
_ => job.handle = Some(handle),
}
job.proc = None;
let ok = matches!(status, BootStatus::BootedOk);
crate::jobs::selftest::emit(
"jera-boot",
&format!("boot_{}_{}_terminal", job.mode.as_str(), job.target),
ok,
&format!("boot '{}' → {}", job.target, status.as_str()),
);
}
}
// ---------------------------------------------------------------------------
// Test seam — a fake runner that drives the whole lifecycle with no real boot.
// ---------------------------------------------------------------------------
/// A scriptable fake [`BootProc`]: its status is driven by a shared cell (a caller
/// flips it to simulate the boot dying); [`kill`](BootProc::kill) records the ask.
#[derive(Clone)]
pub struct FakeProc {
/// The status the fake reports on `poll`.
pub status: Arc<Mutex<BootStatus>>,
/// Set true when `kill` was called.
pub killed: Arc<Mutex<bool>>,
/// Pending log lines emitted on the next `drain_log`.
pub pending_log: Arc<Mutex<Vec<String>>>,
}
impl FakeProc {
/// A fake that starts `Running`.
pub fn running() -> Self {
FakeProc {
status: Arc::new(Mutex::new(BootStatus::Running)),
killed: Arc::new(Mutex::new(false)),
pending_log: Arc::new(Mutex::new(Vec::new())),
}
}
}
impl BootProc for FakeProc {
fn poll(&mut self) -> BootStatus {
self.status.lock().unwrap().clone()
}
fn kill(&mut self) {
*self.killed.lock().unwrap() = true;
*self.status.lock().unwrap() = BootStatus::Killed;
}
fn drain_log(&mut self) -> Vec<String> {
std::mem::take(&mut *self.pending_log.lock().unwrap())
}
fn send_stdin(&mut self, line: &str) {
self.pending_log.lock().unwrap().push(format!("echo: {line}"));
}
}
/// A fake [`BootRunner`] that hands back a preset [`FakeProc`] per spawn.
pub struct FakeBootRunner {
/// The proc handed to the NEXT spawn.
pub next: Mutex<Option<FakeProc>>,
/// If set, spawn fails with this reason.
pub fail: Mutex<Option<String>>,
}
impl FakeBootRunner {
/// A runner whose next spawn yields `proc`.
pub fn yielding(proc: FakeProc) -> Self {
FakeBootRunner { next: Mutex::new(Some(proc)), fail: Mutex::new(None) }
}
}
impl BootRunner for FakeBootRunner {
fn spawn(&self, _target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
if let Some(reason) = self.fail.lock().unwrap().clone() {
return Err(reason);
}
let proc = self.next.lock().unwrap().clone().unwrap_or_else(FakeProc::running);
Ok(Box::new(proc))
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::jobs::{status, JobRecord, JobSink};
fn capturing_sink() -> (JobSink, Arc<Mutex<Vec<JobRecord>>>) {
let cap: Arc<Mutex<Vec<JobRecord>>> = Arc::new(Mutex::new(Vec::new()));
let c2 = Arc::clone(&cap);
let sink = JobSink::remote(move |rec: &JobRecord| c2.lock().unwrap().push(rec.clone()));
(sink, cap)
}
fn exe_target() -> BootTarget {
BootTarget::Exe { name: "full".into(), exe: PathBuf::from("/bin/true"), args: vec![] }
}
#[test]
fn mode_selection_maps_target_type_to_mode() {
assert_eq!(exe_target().mode(), BootMode::Exe);
assert_eq!(BootMode::Exe.job_kind(), kind::BOOT_EXE);
let c = BootTarget::Container(ContainerSpec::image("img"));
assert_eq!(c.mode(), BootMode::Container);
assert_eq!(BootMode::Container.job_kind(), kind::BOOT_CONTAINER);
let v = BootTarget::Vm(BootSpec::kernel_rootfs("vm", "/k", "r"));
assert_eq!(v.mode(), BootMode::Vm);
assert_eq!(BootMode::Vm.job_kind(), kind::BOOT_VM);
}
#[test]
fn press_boot_creates_a_running_job() {
let (sink, cap) = capturing_sink();
let runner = Arc::new(FakeBootRunner::yielding(FakeProc::running()));
let mut st = BootJobs::with_runner(runner, sink, "ws");
let id = st.boot(exe_target());
assert_eq!(st.jobs().len(), 1);
assert_eq!(st.jobs()[0].job_id, id);
assert_eq!(st.jobs()[0].status, BootStatus::Booting);
assert!(st.any_running());
let recs = cap.lock().unwrap();
assert_eq!(recs.len(), 1);
assert_eq!(recs[0].kind, kind::BOOT_EXE);
assert_eq!(recs[0].target, "full");
assert_eq!(recs[0].status, status::RUNNING);
}
#[test]
fn poll_tracks_running_then_death_terminates_job() {
let (sink, cap) = capturing_sink();
let proc = FakeProc::running();
let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
let mut st = BootJobs::with_runner(runner, sink, "ws");
let id = st.boot(exe_target());
st.poll();
assert_eq!(st.jobs()[0].status, BootStatus::Running);
*proc.status.lock().unwrap() = BootStatus::BootedOk;
st.poll();
assert_eq!(st.jobs()[0].status, BootStatus::BootedOk);
assert!(!st.any_running());
let recs = cap.lock().unwrap();
let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
assert_eq!(last.status, status::DONE);
}
#[test]
fn killing_the_job_kills_the_boot_and_marks_killed() {
let (sink, cap) = capturing_sink();
let proc = FakeProc::running();
let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
let mut st = BootJobs::with_runner(runner, sink, "ws");
let id = st.boot(exe_target());
st.poll();
assert!(st.kill(&id));
assert!(*proc.killed.lock().unwrap(), "kill reached the boot proc");
assert_eq!(st.jobs()[0].status, BootStatus::Killed);
let recs = cap.lock().unwrap();
let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
assert!(status::is_terminal(&last.status));
assert!(last.detail_json.contains("killed"), "killed detail carried: {}", last.detail_json);
assert!(!st.kill(&id), "already-terminal kill is a no-op");
}
#[test]
fn spawn_failure_opens_and_immediately_fails_the_job() {
let (sink, cap) = capturing_sink();
let runner = FakeBootRunner::yielding(FakeProc::running());
*runner.fail.lock().unwrap() = Some("no such exe".into());
let mut st = BootJobs::with_runner(Arc::new(runner), sink, "ws");
let id = st.boot(exe_target());
assert!(matches!(st.jobs()[0].status, BootStatus::Failed(_)));
let recs = cap.lock().unwrap();
let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
assert_eq!(last.status, status::FAILED);
}
/// A fake [`BootRunner`] whose `spawn_with_reporter` override actually USES the
/// reporter (unlike [`FakeBootRunner`], which relies on the default no-op) —
/// stands in for [`crate::container::DraupnirContainerBootRunner`] so this test exercises
/// [`BootJobs::with_reports`]'s wiring (handle -> reporter -> runner -> registry
/// -> `state_json`) without a real container engine (that path is covered in
/// `container.rs`'s own tests).
struct ReportingFakeRunner {
proc: FakeProc,
}
impl BootRunner for ReportingFakeRunner {
fn spawn(&self, _target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
Ok(Box::new(self.proc.clone()))
}
fn spawn_with_reporter(
&self,
_target: &BootTarget,
reporter: Reporter,
) -> Result<Box<dyn BootProc>, String> {
reporter.health(crate::jobs::Health::Starting);
reporter.health(crate::jobs::Health::Alive);
Ok(Box::new(self.proc.clone()))
}
}
/// **NEW (job reporting):** [`BootJobs::with_reports`] wires a boot job's
/// [`Reporter`] all the way through to the runner AND back out through
/// [`BootJobs::state_json`] — the read path a UI polls. Without `with_reports`,
/// boot jobs behave exactly as before (see the other tests in this module).
#[test]
fn boot_jobs_with_reports_wires_reporter_and_state_json_carries_reports() {
let tmp = tempfile::TempDir::new().unwrap();
let store = crate::jobs::JobStore::open(tmp.path()).unwrap();
let runner = Arc::new(ReportingFakeRunner { proc: FakeProc::running() });
let mut st = BootJobs::with_runner(runner, store.sink(), "ws").with_reports(&store);
let id = st.boot(BootTarget::Container(ContainerSpec::image("falkordb:edge")));
// The read path via the store directly.
let reports = store.job_reports(&id).expect("health was reported through the wired reporter");
assert_eq!(reports.health(), Some(crate::jobs::Health::Alive), "Starting then Alive -> latest is Alive");
// The read path via state_json (what a UI polls).
let json = st.state_json();
let job_json = &json["boot_jobs"][0];
assert_eq!(job_json["job_id"], id);
assert_eq!(job_json["reports"]["by_key"]["health"]["kind"], "health");
assert_eq!(job_json["reports"]["by_key"]["health"]["data"], "alive");
}
/// Without `with_reports`, `state_json` carries no `"reports"` key at all —
/// the OPT-IN contract: nothing changes for a caller that never asks for it.
#[test]
fn boot_jobs_without_reports_state_json_has_no_reports_key() {
let (sink, _cap) = capturing_sink();
let runner = Arc::new(FakeBootRunner::yielding(FakeProc::running()));
let mut st = BootJobs::with_runner(runner, sink, "ws");
st.boot(exe_target());
let json = st.state_json();
assert!(json["boot_jobs"][0].get("reports").is_none(), "no reports key when not opted in");
}
#[test]
fn container_boot_through_manager_uses_boot_container_kind() {
let (sink, cap) = capturing_sink();
let proc = FakeProc::running();
let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
let mut st = BootJobs::with_runner(runner, sink, "ws");
let target = BootTarget::Container(ContainerSpec::image("holger-appliance:0.1"));
assert_eq!(target.mode(), BootMode::Container);
let id = st.boot(target);
st.poll();
assert_eq!(st.jobs()[0].mode, BootMode::Container);
assert!(st.kill(&id));
assert!(*proc.killed.lock().unwrap());
assert_eq!(cap.lock().unwrap()[0].kind, kind::BOOT_CONTAINER);
}
#[test]
fn vm_boot_through_manager_uses_boot_vm_kind_and_terminates() {
let (sink, cap) = capturing_sink();
let proc = FakeProc::running();
let runner = Arc::new(FakeBootRunner::yielding(proc.clone()));
let mut st = BootJobs::with_runner(runner, sink, "ws");
let id = st.boot(BootTarget::Vm(BootSpec::kernel_rootfs("holger", "/boot/vmlinuz", "rootfs.cpio.gz")));
st.poll();
assert_eq!(st.jobs()[0].mode, BootMode::Vm);
*proc.status.lock().unwrap() = BootStatus::Failed("qemu vanished".into());
st.poll();
assert!(matches!(st.jobs()[0].status, BootStatus::Failed(_)));
let recs = cap.lock().unwrap();
assert_eq!(recs[0].kind, kind::BOOT_VM);
let last = recs.iter().rev().find(|r| r.job_id == id).unwrap();
assert_eq!(last.status, status::FAILED);
}
/// REAL Exe boot (no fake): spawn a sleeper, watch it run, kill it — proves the
/// process-group kill path compiles + runs on this host.
#[test]
fn real_exe_boot_spawns_and_reaches_terminal() {
let runner = Arc::new(DispatchBootRunner::default());
let mut st = BootJobs::with_runner(runner, JobSink::noop(), "ws");
let target = BootTarget::Exe {
name: "sleeper".into(),
exe: PathBuf::from("/bin/sh"),
args: vec!["-c".into(), "sleep 30".into()],
};
let id = st.boot(target);
st.poll();
assert_eq!(st.jobs()[0].status, BootStatus::Running, "sleeper is running");
assert!(st.kill(&id), "kill the running sleeper");
assert_eq!(st.jobs()[0].status, BootStatus::Killed);
}
/// Every pid whose parent is THIS process, with its state letter — read out of
/// `/proc` rather than inferred. `comm` can contain spaces and parentheses, so
/// the fields are taken after the LAST `)`, which is the only parse of
/// `/proc/<pid>/stat` that survives a process called `sh -c (weird name)`.
fn my_children() -> Vec<(i32, char)> {
let me = std::process::id();
let mut out = Vec::new();
let Ok(rd) = std::fs::read_dir("/proc") else { return out };
for e in rd.flatten() {
let Some(pid) = e.file_name().to_str().and_then(|s| s.parse::<i32>().ok()) else {
continue;
};
let Ok(stat) = std::fs::read_to_string(format!("/proc/{pid}/stat")) else { continue };
let Some((_, rest)) = stat.rsplit_once(')') else { continue };
let mut it = rest.split_whitespace();
let Some(state) = it.next().and_then(|s| s.chars().next()) else { continue };
let Some(ppid) = it.next().and_then(|s| s.parse::<u32>().ok()) else { continue };
if ppid == me {
out.push((pid, state));
}
}
out
}
/// The state letter of one pid, or `None` once it is gone (reaped).
fn state_of(pid: i32) -> Option<char> {
my_children().into_iter().find(|(p, _)| *p == pid).map(|(_, s)| s)
}
/// Find OUR child by a marker planted in its argv.
///
/// `cargo test` runs these in THREADS of one process, so a set-difference over
/// "children of me" can pick up a sibling test's sleeper — which is exactly how
/// the running-boot case failed in the suite while passing alone. Matching on
/// `/proc/<pid>/cmdline` names the right process regardless of who else spawns.
/// Polls briefly because `spawn` returning and the pid appearing are not the same
/// instant.
fn child_with_marker(marker: &str) -> Option<i32> {
for _ in 0..100 {
for (pid, _) in my_children() {
if let Ok(cmd) = std::fs::read(format!("/proc/{pid}/cmdline")) {
if String::from_utf8_lossy(&cmd).contains(marker) {
return Some(pid);
}
}
}
std::thread::sleep(std::time::Duration::from_millis(10));
}
None
}
/// **A dropped boot leaves NO zombie — the leak is closed by construction.**
///
/// `ExeProc` owns a `std::process::Child`, and dropping a `Child` in Rust
/// DETACHES it: the OS keeps the exit status until someone `wait()`s, so a proc
/// dropped without being polled to a terminal state leaks a zombie that holds a
/// pid slot forever. A zombie has no RSS, no CPU and no fds, so it shows up in
/// neither `top` nor an OOM dump — the only thing it exhausts is `TasksMax` /
/// `pid_max`, and the failure mode is not slowness but a box that abruptly
/// cannot fork.
///
/// This is the drop path specifically: `kill()` already reaps (`child.wait()`),
/// and `poll()` reaps via `try_wait()` once the child has exited. The hole is the
/// proc that is simply DROPPED — a `Box<dyn BootProc>` released without either.
///
/// RED before the `Drop` impl: `/bin/true` exits at once, is never polled, and
/// its pid is still in state `Z` after the drop.
#[test]
fn a_dropped_boot_proc_reaps_its_child_instead_of_leaving_a_zombie() {
// A short life, not an instant one: the child must still be alive long
// enough to be identified by its marker, then exit on its own so that
// NOTHING has waited on it when the drop happens.
let marker = "jera-drop-probe-exits-on-its-own";
let target = BootTarget::Exe {
name: "quick".into(),
exe: PathBuf::from("/bin/sh"),
args: vec!["-c".into(), format!("sleep 0.3 # {marker}")],
};
let proc = ExeBootRunner.spawn(&target).expect("spawn the short-lived child");
let pid = child_with_marker(marker).expect("the child must appear under /proc");
// Let it exit. Deliberately do NOT poll() — polling is the path that reaps,
// and skipping it is exactly the leak being tested.
std::thread::sleep(std::time::Duration::from_millis(600));
assert_eq!(state_of(pid), Some('Z'), "unpolled and exited, it should be a zombie HERE");
drop(proc);
std::thread::sleep(std::time::Duration::from_millis(50));
assert_ne!(
state_of(pid),
Some('Z'),
"pid {pid} is still a zombie after its BootProc was dropped — a dropped \
`Child` is detached, not reaped, so every boot released without kill() or \
a terminal poll() burns a pid slot permanently"
);
}
/// **The other half: dropping a proc that is STILL RUNNING.**
///
/// `try_wait()` alone does not close this one — a running child returns
/// `Ok(None)`, the drop returns having reaped nothing, and the process becomes a
/// zombie later with no handle left to wait on it. Worse than the zombie: it is
/// an orphan in its own process group that nothing in this program can stop.
///
/// So the drop must SIGKILL the group and wait. This test holds a real sleeper —
/// which would outlive the test binary if the drop merely detached.
#[test]
fn dropping_a_still_running_boot_kills_its_group_rather_than_orphaning_it() {
let marker = "jera-drop-probe-still-running";
let target = BootTarget::Exe {
name: "sleeper".into(),
exe: PathBuf::from("/bin/sh"),
args: vec!["-c".into(), format!("sleep 300 # {marker}")],
};
let mut proc = ExeBootRunner.spawn(&target).expect("spawn the sleeper");
let pid = child_with_marker(marker).expect("the sleeper must appear under /proc");
assert_eq!(proc.poll(), BootStatus::Running, "the sleeper is genuinely running");
drop(proc);
std::thread::sleep(std::time::Duration::from_millis(150));
let state = state_of(pid);
assert!(
state.is_none(),
"pid {pid} survives its dropped BootProc as {state:?} — a running boot whose \
handle is gone can never be polled or killed again, so the drop must kill \
the process GROUP and wait, not detach it"
);
}
}