crates/jera/src/container.rs
//! **No-shell container runtime — routed THROUGH draupnir.** jera keeps only the
//! *job policy* for a container; the ONE bollard/podman REST engine for the whole
//! constellation lives in **draupnir** (`draupnir::container`). This mirrors how the
//! KVM leg already delegates (`crate::vm::DraupnirVmBootRunner` → `draupnir::Boot`):
//! there is no second engine here.
//!
//! Two shapes, both delegating to draupnir:
//!
//! - **long-lived boot** — [`DraupnirContainerBootRunner`] implements [`BootRunner`]:
//! map a [`ContainerSpec`] onto a `draupnir::BootSpec` ([`ContainerSpec::to_draupnir`]),
//! boot it via `draupnir::Boot`, and drive its lifecycle both ways through the
//! returned [`BootProc`] over draupnir's [`ContainerControl`](draupnir::container::ContainerControl)
//! seam (poll reports a container that died on its own; kill stops+removes it).
//! This is what the job-bound [`crate::boot::BootJobs`] manager drives, and the API
//! korp uses to "start a container as a job".
//! - **run to completion** — [`run_container`] opens a `running` job row, then routes
//! through [`draupnir::container::run_to_completion`] (the one engine): bring the
//! container up, wait for it to exit, capture its logs + exit code, remove it, then
//! write the terminal row (`done` on exit 0, else `failed`).
//!
//! draupnir's always-compiled `Boot` + `ContainerControl` traits are the test seam:
//! unit tests inject a mock draupnir backend and exercise the whole boot → logs →
//! state → kill (and run-to-completion) wiring with **no daemon and no feature**. The
//! live engine is enabled by jera's `container-oci` feature (`draupnir/backend-oci`);
//! without it [`StubContainerBootRunner`] fails honestly (it never fakes a boot).
use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use crate::boot::{BootProc, BootRunner, BootStatus, BootTarget};
use crate::jobs::{Health, Reporter};
/// The outcome of an [`exec`](BootProc::exec) into a running container, re-exported
/// from draupnir's ONE OCI engine so a jera consumer (a Skidbladnir lifecycle-swap)
/// reads exit code + captured stdout/stderr without naming draupnir directly. A
/// non-zero `exit_code` is carried here, never as an `Err`.
pub use draupnir::container::ExecOutcome;
// ---------------------------------------------------------------------------
// NetMode — the container network attachment (airgap isolation knob).
// ---------------------------------------------------------------------------
/// Container **network mode** — how a routed container attaches to the host
/// network. This is the **load-bearing airgap knob**: [`NetMode::None`] renders
/// `--network none`, so a routed job runs with **no network namespace and no
/// egress** — the isolation guarantee a Skidbladnir airgap container route must not
/// lose. The default ([`NetMode::Default`]) emits **no** `--network` flag, so an
/// existing NAT/bridge spec is byte-identical.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum NetMode {
/// The engine's default network (rootful bridge / rootless slirp NAT). No
/// `--network` flag is emitted — the pre-existing behaviour for every spec.
#[default]
Default,
/// `--network none` — **no network namespace attached**: full airgap egress
/// isolation (Skidbladnir's reason to exist).
None,
/// `--network host` — share the host's network namespace.
Host,
/// `--network bridge` — the default bridge network, named explicitly.
Bridge,
}
impl NetMode {
/// Whether this is the engine default (no `--network` emitted). The
/// `skip_serializing_if` guard so an existing spec serializes byte-identically.
pub fn is_default(&self) -> bool {
matches!(self, NetMode::Default)
}
/// The `--network` **value** for the OCI/podman run, or `None` for the engine
/// default (no flag): `None`→`"none"`, `Host`→`"host"`, `Bridge`→`"bridge"`.
/// This is the value draupnir's OCI engine would set on the container's
/// `HostConfig.network_mode` once it carries a net hook (see the engine-hook
/// note on [`ContainerSpec::network_arg`]).
pub fn oci_value(&self) -> Option<&'static str> {
match self {
NetMode::Default => None,
NetMode::None => Some("none"),
NetMode::Host => Some("host"),
NetMode::Bridge => Some("bridge"),
}
}
}
// ---------------------------------------------------------------------------
// BindMount — a host path mounted into the container (`-v` semantics).
// ---------------------------------------------------------------------------
/// A host **bind mount** applied to the container: the [`host`](Self::host) path is
/// mounted at [`container`](Self::container) inside it, **read-only** when
/// [`ro`](Self::ro). Renders to podman `-v host:container[:ro]` and, at the boundary,
/// to draupnir's bind-string form ([`ContainerSpec::bind_strings`]) fed straight into
/// draupnir's mount-aware run
/// ([`draupnir::container::ContainerBoot::run_to_completion_with_binds`]).
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct BindMount {
/// Host path (the mount source).
pub host: String,
/// Container path (the mount point inside the container).
pub container: String,
/// Mount read-only (`:ro`) when true. Defaults to false (read-write).
#[serde(default, skip_serializing_if = "is_false")]
pub ro: bool,
}
/// `skip_serializing_if` helper for a plain `bool` field (serde needs `fn(&bool)`).
fn is_false(b: &bool) -> bool {
!*b
}
impl BindMount {
/// A **read-write** bind of `host` → `container`.
pub fn rw(host: impl Into<String>, container: impl Into<String>) -> Self {
BindMount { host: host.into(), container: container.into(), ro: false }
}
/// A **read-only** bind of `host` → `container`.
pub fn ro(host: impl Into<String>, container: impl Into<String>) -> Self {
BindMount { host: host.into(), container: container.into(), ro: true }
}
/// The `host:container[:ro]` bind string — the podman `-v` form and the exact
/// `&str` draupnir's `run_to_completion_with_binds` consumes.
pub fn bind_string(&self) -> String {
if self.ro {
format!("{}:{}:ro", self.host, self.container)
} else {
format!("{}:{}", self.host, self.container)
}
}
}
// ---------------------------------------------------------------------------
// PortMap — a distinct host:container published-port mapping (the zone-offset knob).
// ---------------------------------------------------------------------------
/// A container **published-port mapping** — a distinct `host:container` pair
/// (podman `-p host:container`). This is the **per-zone port knob**: a service
/// listening on a **fixed port inside** the container (FalkorDB always binds
/// `6379`, Spark-Connect `15002`) is published on a **possibly-different host
/// port** so isolated zones coexist on one host. The `Demo` zone publishes
/// `6379:6379`; `Test` publishes host `6380` → container `6379`
/// (`PortMap::new(6380, 6379)`); `Prod` host `6381` → container `6379`. Threaded
/// at the boundary onto `draupnir::PortMap` in [`ContainerSpec::to_draupnir`].
///
/// The single-port [`ContainerSpec::ports`] field (a bare `u16`) is the
/// `host == container` special case and stays byte-identical; use this only when
/// the host port must differ from the in-container port. Both are published.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct PortMap {
/// The **host** port the publish binds on the host (what a host client hits).
pub host: u16,
/// The **container** port the service listens on *inside* the container.
pub container: u16,
}
impl PortMap {
/// A distinct `host:container` publish (the host port may differ, e.g. a
/// per-zone offset onto a fixed in-container port).
pub fn new(host: u16, container: u16) -> Self {
PortMap { host, container }
}
/// The `host == container` publish — the single-port special case (matches a
/// bare-`u16` [`ContainerSpec::ports`] entry).
pub fn same(port: u16) -> Self {
PortMap { host: port, container: port }
}
}
impl From<u16> for PortMap {
/// A bare port maps `host == container` (the single-port form).
fn from(port: u16) -> Self {
PortMap::same(port)
}
}
// ---------------------------------------------------------------------------
// ContainerSpec — the descriptor a boot/run brings up.
// ---------------------------------------------------------------------------
/// The container a boot/run brings up: an `image` reference plus the **run
/// overrides** applied when it comes up (an entrypoint/`cmd` override, extra
/// `env`, and published `ports`). `name` is an optional explicit container name;
/// when empty the runner mints a fresh `jera-boot-<uuid>` so a kill can address the
/// exact container.
///
/// Plain data (pure serde) so a caller (korp, a server RPC) builds it without
/// linking draupnir's backend.
// `PartialEq` only (not `Eq`): the `cpus` cap is an `Option<f64>` and `f64` is not
// `Eq`. Every `==`/`assert_eq!` on a `ContainerSpec` needs only `PartialEq`, and it
// is never a hash/btree key — so dropping `Eq` is additive.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct ContainerSpec {
/// Image reference to run, e.g. `docker.io/library/nginx:alpine`.
pub image: String,
/// Explicit container name, or empty to mint `jera-boot-<uuid>`.
#[serde(default, skip_serializing_if = "String::is_empty")]
pub name: String,
/// Command / entrypoint override (empty = the image's own default).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub cmd: Vec<String>,
/// Extra environment (sorted, stable), carried as `KEY=VALUE` at the boundary.
#[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
pub env: BTreeMap<String, String>,
/// Container ports to publish, each bound to the **same host port**
/// (`host == container`) — the single-port form. For a **distinct** host port
/// (a per-zone offset onto a fixed in-container port) use
/// [`port_maps`](Self::port_maps); both sets are published.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub ports: Vec<u16>,
/// Container **distinct-mapping** published ports — each a [`PortMap`]
/// `{ host, container }` (podman `-p host:container`). This is the form that
/// publishes a **different host port than the in-container port**: a FalkorDB
/// container listens on `6379` inside every zone, but the `Test` zone
/// publishes it on host `6380` and `Prod` on `6381`, so the zones don't
/// collide on the host yet each reaches the fixed in-container `6379`. Compose
/// it from [`nornir_zone::Zone`] with [`with_zone_port`](Self::with_zone_port)
/// (`host = zone.port_on(base)`, `container = base`). Empty (the default) =
/// serializes byte-identically (skipped) and leaves the boot unchanged;
/// threaded onto `draupnir::BootSpec.port_maps` in [`to_draupnir`](Self::to_draupnir).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub port_maps: Vec<PortMap>,
/// Network mode applied when the container comes up. [`NetMode::Default`] (the
/// default) keeps the engine's default network — no `--network` flag, so an
/// existing spec is unchanged. [`NetMode::None`] = `--network none` = full
/// airgap egress isolation (the load-bearing knob for a routed airgap job).
#[serde(default, skip_serializing_if = "NetMode::is_default")]
pub net: NetMode,
/// Host **bind mounts** applied when the container comes up (`-v
/// host:container[:ro]`). Empty (the default) = no mounts, so an existing spec
/// is unchanged; a non-empty list is threaded through
/// [`bind_strings`](Self::bind_strings) into draupnir's mount-aware run
/// ([`draupnir::container::ContainerBoot::run_to_completion_with_binds`]).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub mounts: Vec<BindMount>,
/// The **zone** this container belongs to (`demo` / `test` / `prod`, the
/// [`nornir_zone::Zone::key`] — a platform-level label carried on every job). Empty
/// (the default) = unscoped, so an existing spec serializes byte-identically. A
/// caller (korp) sets it so the submitted job is tagged with its zone; per-zone
/// isolation of the *container name* + *published port* is the caller's to compose
/// from [`nornir_zone::Zone`]'s helpers (`container_name` / `port_on`), which this
/// spec's [`name`](Self::name) + [`ports`](Self::ports) then carry.
#[serde(default, skip_serializing_if = "String::is_empty")]
pub zone: String,
/// **CPU quota** for this container (podman `--cpus`) — how many host cores it
/// may use. `None` (the default) leaves it **unconstrained**, so the container
/// sees **all** host cores: the hot-infra default (FalkorDB's OpenMP pool, a Spark
/// executor must never be throttled to one core). `Some(n)` caps it at `n` cores.
/// Threaded onto `draupnir::BootSpec.cpus` → `HostConfig.nano_cpus`. `None`
/// serializes byte-identically (skipped), so an existing spec is unchanged.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cpus: Option<f64>,
/// **Memory limit** in MiB for this container (podman `--memory`). `None` (the
/// default) = unconstrained (may use the box's memory) — the hot-infra default;
/// `Some(m)` caps it. Threaded onto `draupnir::BootSpec.mem_limit_mb` →
/// `HostConfig.memory`. `None` serializes byte-identically (skipped).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub mem_mb: Option<u32>,
}
impl ContainerSpec {
/// A spec that runs `image` with the image's own defaults.
pub fn image(image: impl Into<String>) -> Self {
ContainerSpec { image: image.into(), ..Default::default() }
}
/// Tag this container with its [`zone`](Self::zone) (builder style) — the
/// platform-level environment (`demo` / `test` / `prod`) the job belongs to,
/// stored as [`nornir_zone::Zone::key`]. Compose the zone's *container name* +
/// *published port* from the same [`nornir_zone::Zone`] into [`name`](Self::name)
/// / [`ports`](Self::ports) so two zones never collide.
pub fn with_zone(mut self, zone: nornir_zone::Zone) -> Self {
self.zone = zone.key().to_string();
self
}
/// Set the explicit container **name** (builder style) — e.g. a per-zone name
/// [`nornir_zone::Zone::container_name`] (`korp-falkordb-test`) so two zones'
/// containers never collide. Empty leaves the runner to mint `jera-boot-<uuid>`.
pub fn with_name(mut self, name: impl Into<String>) -> Self {
self.name = name.into();
self
}
/// Publish a **distinct** `host:container` mapping (builder style) — the host
/// port may differ from the in-container port (see [`port_maps`](Self::port_maps)
/// and [`PortMap`]). Use when the service's in-container port is fixed but the
/// host port must vary (per-zone offsets).
pub fn with_port_map(mut self, host: u16, container: u16) -> Self {
self.port_maps.push(PortMap::new(host, container));
self
}
/// **THE korp-facing per-zone publish** (builder style): publish the fixed
/// in-container `container` port on the zone's **offset host port**
/// ([`nornir_zone::Zone::port_on`]) — `Demo` → `container:container` (the
/// pre-zone byte-identical case), `Test` → `(container+1):container`, `Prod` →
/// `(container+2):container`. This is the fix for a per-zone-offset service:
/// `.with_zone_port(Zone::Test, 6379)` publishes host `6380` → container
/// `6379`, so FalkorDB (always `6379` INSIDE) is reachable on the test zone's
/// host port while demo keeps `6379`. Pair with [`with_zone`](Self::with_zone)
/// (which sets the job's zone *label*) — this sets the *port*.
///
/// Note `Zone::Demo` yields `PortMap { host: container, container }`, i.e. the
/// `host == container` form — functionally the same publish as
/// `ports = [container]`, so an existing demo caller is unaffected.
pub fn with_zone_port(mut self, zone: nornir_zone::Zone, container: u16) -> Self {
self.port_maps.push(PortMap::new(zone.port_on(container), container));
self
}
/// Give this container **all host cores** (builder style) — set
/// [`cpus`](Self::cpus) to `std::thread::available_parallelism()`. This is the
/// explicit "run hot" knob for infra that must saturate the box; note the
/// *default* (unset `cpus`) is ALSO all cores (podman leaves an unconstrained
/// container on every core), so this is for a caller that wants the cap stated
/// explicitly. Falls back to leaving it unset if the core count can't be read.
pub fn with_all_host_cores(mut self) -> Self {
if let Ok(n) = std::thread::available_parallelism() {
self.cpus = Some(n.get() as f64);
}
self
}
/// Cap this container at `cpus` host cores (builder style) — podman `--cpus`.
/// Prefer leaving it unset (all cores) for hot infra; this is the explicit cap.
pub fn with_cpus(mut self, cpus: f64) -> Self {
self.cpus = Some(cpus);
self
}
/// Cap this container's memory at `mem_mb` MiB (builder style) — podman
/// `--memory`. Unset (the default) = unconstrained (the hot-infra default).
pub fn with_mem_mb(mut self, mem_mb: u32) -> Self {
self.mem_mb = Some(mem_mb);
self
}
/// The per-boot container name: the explicit `name`, else a fresh
/// `jera-boot-<uuid>` so a kill addresses the exact container.
pub fn container_name(&self) -> String {
if self.name.is_empty() {
format!("jera-boot-{}", uuid::Uuid::new_v4())
} else {
self.name.clone()
}
}
/// The `--network <mode>` argument this spec's [`net`](Self::net) renders to for
/// the OCI/podman run, or `None` for the engine default (no flag — an existing
/// NAT spec is unchanged). [`NetMode::None`] → `Some("--network none")`, the
/// airgap-isolation form.
///
/// **Live-engine wire (landed).** `net` now reaches the live run: draupnir's OCI
/// engine sets the container `HostConfig.network_mode` from
/// [`NetMode::oci_value`], threaded across in [`to_draupnir`](Self::to_draupnir)
/// via `draupnir::NetMode::from_oci_value(self.net.oci_value())`. draupnir reads
/// `spec.net` in BOTH legs jera drives (the [`DraupnirContainerBootRunner`] boot
/// and [`run_container`]'s `run_to_completion_with_binds`), so a routed
/// [`NetMode::None`] spec produces a real `--network none` airgap container. This
/// `network_arg` string form remains for any direct/diagnostic caller.
pub fn network_arg(&self) -> Option<String> {
self.net.oci_value().map(|v| format!("--network {v}"))
}
/// This spec's [`mounts`](Self::mounts) rendered as draupnir/`-v` bind strings
/// (`host:container[:ro]`), the exact `&[String]` shape draupnir's
/// [`run_to_completion_with_binds`](draupnir::container::ContainerBoot::run_to_completion_with_binds)
/// consumes. Empty when there are no mounts (byte-identical to the mount-less
/// run — draupnir guarantees empty binds ⇒ identical `HostConfig`).
pub fn bind_strings(&self) -> Vec<String> {
self.mounts.iter().map(BindMount::bind_string).collect()
}
/// Map this jera container spec onto a `draupnir::BootSpec` (the OCI container
/// path) — the shape handed to draupnir's container backend, mirroring
/// [`crate::vm::BootSpec::to_draupnir`] for the KVM path. The image, the
/// per-boot [`container_name`](Self::container_name) (so a fresh
/// `jera-boot-<uuid>` still gives a unique, kill-addressable instance), and the
/// run overrides (cmd / env / published ports) all carry across, so a boot
/// through draupnir brings up the same container jera's own engine did.
///
/// draupnir is always a dependency (its container types compile with no
/// feature); the live engine is enabled by jera's `container-oci` feature.
pub fn to_draupnir(&self) -> draupnir::BootSpec {
let mut d = draupnir::BootSpec::container(self.container_name(), self.image.clone())
.with_cmd(self.cmd.clone());
d.ports = self.ports.clone();
// Distinct `host:container` maps (the per-zone offset knob): map each jera
// PortMap onto a draupnir::PortMap so a `Test` zone's host 6380 → container
// 6379 reaches the live podman `-p 6380:6379`. Empty (the default) leaves
// `d.port_maps` empty ⇒ byte-identical create body (demo path unchanged).
d.port_maps = self
.port_maps
.iter()
.map(|p| draupnir::PortMap::new(p.host, p.container))
.collect();
d.env = self.env.clone();
// Net wire (airgap): thread this spec's network mode onto the draupnir
// BootSpec so draupnir's OCI engine sets `HostConfig.network_mode` on the
// live podman run. `NetMode::Default` ⇒ `from_oci_value(None)` ⇒ draupnir's
// net stays `Default` ⇒ no `--network` flag ⇒ byte-identical create body.
// `NetMode::None` ⇒ `"none"` ⇒ live `--network none` (full airgap egress
// isolation — the Skidbladnir airgap route's load-bearing guarantee).
// draupnir reads `spec.net` in BOTH legs jera drives (`Boot::boot` and
// `ContainerBoot::run_to_completion_with_binds`), so this one line flows
// through the long-lived BootRunner boot AND the run-to-completion job.
d.net = draupnir::NetMode::from_oci_value(self.net.oci_value());
// Resource caps (the hot-infra law): thread the CPU/memory knobs onto the
// draupnir BootSpec so they reach the live podman `--cpus`/`--memory` on
// `HostConfig.nano_cpus`/`memory`. Both `None` (the default) ⇒ draupnir mints
// no cap ⇒ the container sees ALL host cores (never throttled to one) and the
// create body is byte-identical. `zone` is a jera-level job label (draupnir
// has no zone concept — it provisions the container the spec's name+ports
// already scope to the zone), so it is carried on the job, not the BootSpec.
d.cpus = self.cpus;
d.mem_limit_mb = self.mem_mb;
// Mounts. This line did not exist, and the comment on `net` above — "draupnir
// reads `spec.net` in BOTH legs jera drives" — was precisely what was NOT true
// of `mounts`: `bind_strings()` reached draupnir only on the run-to-completion
// leg (`run_container`, a separate argument), while the long-lived
// [`DraupnirContainerBootRunner`] boots through `Boot::boot(&spec.to_draupnir())`
// and this BootSpec carried none. MEASURED on oden 2026-09-10: korp's FalkorDB
// ContainerSpec declared a named volume at `/data`; the live container had
// `Mounts: []` and answered `DBSIZE 0` after a recreate. draupnir f6c83ef put
// `binds` ON the spec; from here both legs are one line.
d.binds = self.bind_strings();
d
}
}
// ---------------------------------------------------------------------------
// submit_container_job — THE korp-facing "container up as a tracked job" entry.
// ---------------------------------------------------------------------------
/// **Submit a container as a tracked jera job** — THE entry point korp calls to
/// bring a zone's default-infra container (FalkorDB, Spark) up **as a job**, routed
/// through draupnir's ONE OCI engine with its **published ports**, **env**, **zone**
/// label and **full-machine resource** knobs.
///
/// It opens a durable job row (kind [`crate::jobs::kind::BOOT_CONTAINER`], detail
/// carrying `image`/`zone`/`ports`/`cpus`/`mem_mb`) and returns the **job id**. The
/// container is started detached through draupnir; poll readiness with
/// [`crate::boot::BootJobs::wait_until_running`] (or [`crate::boot::BootJobs::status_of`] /
/// [`crate::boot::BootJobs::state_json`]) on the same `jobs` handle. `Running` means
/// the container process is up — the caller then runs its own app-level gate (e.g.
/// "the published port accepts") before using the service.
///
/// The `jobs` manager must be wired to draupnir's live engine (jera's
/// `container-oci` feature ⇒ `draupnir/backend-oci`); without it the job opens and
/// immediately fails honestly (never a fake boot).
///
/// ## The API shape korp consumes
/// FalkorDB always listens on `6379` **inside** the container, so a per-zone
/// offset must publish a *distinct* `host:container` map — NOT `host == container`
/// (which would bind `6380:6380` and never reach the container). Use
/// [`ContainerSpec::with_zone_port`] — it publishes `zone.port_on(6379)` on the
/// host → `6379` in the container, so `Demo`→`6379:6379`, `Test`→`6380:6379`,
/// `Prod`→`6381:6379` all reach FalkorDB:
/// ```ignore
/// use jera::boot::BootJobs;
/// use jera::container::{submit_container_job, ContainerSpec};
/// use nornir_zone::Zone;
///
/// let zone = Zone::Test; // demo/test/prod
/// let mut jobs = BootJobs::new(sink, "korp"); // draupnir-backed runner
/// let spec = ContainerSpec::image("localhost/nordisk-falkordb-valkey:9-20260724")
/// .with_zone(zone) // zone label on the job
/// .with_name(zone.container_name("korp-falkordb")) // per-zone name (no collide)
/// .with_zone_port(zone, 6379) // host 6380 -> container 6379
/// .with_all_host_cores() // hot: all cores (never throttled)
/// .with_mem_mb(16384); // sized memory
/// let id = submit_container_job(&mut jobs, spec);
/// let status = jobs.wait_until_running(&id, timeout, interval); // wait-for-running
/// ```
pub fn submit_container_job(jobs: &mut crate::boot::BootJobs, spec: ContainerSpec) -> String {
jobs.boot(crate::boot::BootTarget::Container(spec))
}
// ---------------------------------------------------------------------------
// connect_default_runner — the container BootRunner the job manager drives.
// ---------------------------------------------------------------------------
/// Connect the default container [`BootRunner`]: the [`DraupnirContainerBootRunner`]
/// (routing THROUGH draupnir's ONE OCI engine) when the `container-oci` feature is
/// on, else the honest [`StubContainerBootRunner`] (a clear not-wired failure — never
/// a silent shell fallback).
///
// TODO(jera-container): route the long-lived container/systemd leg through
// Skidbladnir's service/container bring-up (a job that IS a managed service) rather
// than a bare boot. Skidbladnir is a release-recipe repo and is intentionally NOT
// edited from jera. Until that seam lands, this connects draupnir's OCI engine
// (the one zero-shell REST engine) under `container-oci`, else the honest stub.
pub fn connect_default_runner() -> Box<dyn BootRunner> {
// Preferred (feature `container-oci`): route the container boot THROUGH
// draupnir's OCI backend — the ONE bollard engine — exactly as the VM boot goes
// through draupnir's KVM backend. No second engine.
#[cfg(feature = "container-oci")]
{
Box::new(DraupnirContainerBootRunner::<draupnir::container::ContainerBoot>::default())
}
// No `container-oci`: honest not-wired stub (mirror of the VM leg's stub).
#[cfg(not(feature = "container-oci"))]
{
Box::new(StubContainerBootRunner)
}
}
// ---------------------------------------------------------------------------
// DraupnirContainerBootRunner — route a container boot THROUGH draupnir's OCI
// backend (the ONE bollard engine), mirroring `vm.rs`'s DraupnirVmBootRunner.
// ---------------------------------------------------------------------------
/// The **zero-shell** container boot runner that provisions THROUGH **draupnir** —
/// the shared boot/provisioning library that owns the single bollard/podman REST
/// engine. It maps a [`ContainerSpec`] onto a `draupnir::BootSpec`
/// ([`ContainerSpec::to_draupnir`]), boots it via `draupnir::Boot`, and adapts the
/// returned `draupnir::Machine` into a [`BootProc`] using draupnir's
/// [`ContainerControl`](draupnir::container::ContainerControl) seam — so the job
/// lifecycle drives the container through draupnir: job-kill → `stop`; poll →
/// `container_state` (exit-code aware, so a container that dies on its own flips the
/// job, a clean exit is `BootedOk`, a crash is `Failed`).
///
/// This is the container mirror of [`crate::vm::DraupnirVmBootRunner`]: jera keeps
/// only the job policy; instance-firing lives in exactly one place, draupnir. It is
/// generic over the draupnir backend `B` so a test injects a mock and proves the
/// jera→draupnir routing with no daemon; the production `B` is
/// [`draupnir::container::ContainerBoot`].
#[derive(Default)]
pub struct DraupnirContainerBootRunner<B = draupnir::container::ContainerBoot> {
backend: B,
}
impl<B> DraupnirContainerBootRunner<B>
where
B: draupnir::Boot + draupnir::container::ContainerControl + Clone + Send + Sync + 'static,
{
/// A runner over an explicit draupnir backend — the test seam (a mock backend
/// drives the whole lifecycle with no daemon).
pub fn with_backend(backend: B) -> Self {
DraupnirContainerBootRunner { backend }
}
/// Spawn a [`ContainerSpec`] directly (the korp-facing entry point that does not
/// go through a [`BootTarget`]): boot it through draupnir, returning the live
/// [`BootProc`]. No [`Health`] reporting — see
/// [`spawn_spec_with_reporter`](Self::spawn_spec_with_reporter).
pub fn spawn_spec(&self, spec: &ContainerSpec) -> Result<Box<dyn BootProc>, String> {
self.spawn_spec_with_reporter(spec, Reporter::noop())
}
/// Like [`spawn_spec`](Self::spawn_spec) but emits [`Health`] as the container's
/// lifecycle progresses: [`Health::Starting`] before the boot, [`Health::Failed`]
/// if it errors, [`Health::Alive`] once up — then the returned proc keeps emitting
/// on every poll/kill. This is the korp "is the FalkorDB container alive or gone"
/// wiring.
pub fn spawn_spec_with_reporter(
&self,
spec: &ContainerSpec,
reporter: Reporter,
) -> Result<Box<dyn BootProc>, String> {
reporter.health(Health::Starting);
let machine = match self.backend.boot(&spec.to_draupnir()) {
Ok(m) => m,
Err(e) => {
reporter.health(Health::Failed);
return Err(format!("draupnir container boot: {e}"));
}
};
reporter.health(Health::Alive);
Ok(Box::new(DraupnirContainerProc {
backend: self.backend.clone(),
machine,
killed: false,
terminal: None,
reporter,
}))
}
}
impl<B> BootRunner for DraupnirContainerBootRunner<B>
where
B: draupnir::Boot + draupnir::container::ContainerControl + Clone + Send + Sync + 'static,
{
fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
let BootTarget::Container(spec) = target else {
return Err(format!("DraupnirContainerBootRunner cannot boot a {} target", target.mode().as_str()));
};
self.spawn_spec(spec)
}
fn spawn_with_reporter(&self, target: &BootTarget, reporter: Reporter) -> Result<Box<dyn BootProc>, String> {
let BootTarget::Container(spec) = target else {
return Err(format!("DraupnirContainerBootRunner cannot boot a {} target", target.mode().as_str()));
};
self.spawn_spec_with_reporter(spec, reporter)
}
}
/// Adapts a draupnir container `draupnir::Machine` to jera's [`BootProc`], driving
/// its lifecycle over draupnir's [`ContainerControl`](draupnir::container::ContainerControl)
/// seam. `poll` maps draupnir's exit-code-aware
/// [`ContainerState`](draupnir::container::ContainerState) onto [`BootStatus`]
/// (clean exit → `BootedOk`, crash → `Failed(code)`, gone → `Failed`/`Killed`).
struct DraupnirContainerProc<B: draupnir::container::ContainerControl> {
backend: B,
machine: draupnir::Machine,
killed: bool,
terminal: Option<BootStatus>,
reporter: Reporter,
}
impl<B: draupnir::container::ContainerControl + Send> BootProc for DraupnirContainerProc<B> {
fn poll(&mut self) -> BootStatus {
use draupnir::container::ContainerState;
if let Some(t) = &self.terminal {
return t.clone();
}
match self.backend.container_state(&self.machine) {
ContainerState::Running => {
self.reporter.health(Health::Alive);
BootStatus::Running
}
ContainerState::Exited(code) => {
let t = if self.killed {
BootStatus::Killed
} else if code == 0 {
BootStatus::BootedOk
} else {
BootStatus::Failed(format!("container exited {code}"))
};
self.reporter.health(if self.killed || code == 0 { Health::Gone } else { Health::Failed });
self.terminal = Some(t.clone());
t
}
ContainerState::Gone => {
let t = if self.killed { BootStatus::Killed } else { BootStatus::Failed("container gone".into()) };
self.reporter.health(if self.killed { Health::Gone } else { Health::Failed });
self.terminal = Some(t.clone());
t
}
}
}
fn kill(&mut self) {
if self.terminal.is_some() {
return;
}
self.backend.stop(&self.machine);
self.killed = true;
self.terminal = Some(BootStatus::Killed);
self.reporter.health(Health::Gone);
}
fn drain_log(&mut self) -> Vec<String> {
self.backend.drain_logs(&self.machine)
}
/// **Exec forward — the detached-container control primitive (jera "knob 3").**
/// Run `argv` *inside* this proc's already-running container by forwarding to
/// draupnir's ONE OCI engine over the exact `(backend, machine)` handle this proc
/// already holds: `self.backend.exec(&self.machine, argv)`. `machine.id`
/// (`draupnir-<name>`) IS the container id draupnir's `podman exec` targets, so
/// nothing else needs plumbing — the surviving handle is the whole story.
///
/// Returns the exec'd process's [`ExecOutcome`] (exit code + captured
/// stdout/stderr); a **non-zero** exec exit is in `exit_code`, not an `Err`. Only
/// a guard failure (container-only / non-empty argv, enforced by draupnir) or an
/// engine/connect error maps through to `Err(String)`. Always compiled + fully
/// mock-tested via draupnir's provided [`ContainerControl::exec`](draupnir::container::ContainerControl::exec)
/// (no daemon, no feature); the LIVE `podman exec` needs jera's `container-oci`
/// (⇒ `draupnir/backend-oci`) — without it draupnir's `exec_command` honestly
/// reports `Unsupported` rather than faking a run.
fn exec(&self, argv: &[&str]) -> Result<ExecOutcome, String> {
self.backend
.exec(&self.machine, argv)
.map_err(|e| format!("draupnir container exec: {e}"))
}
}
/// The default container runner when draupnir's OCI backend is not compiled in
/// (`container-oci` off): it reports a clear, honest failure rather than pretending
/// to boot — the container mirror of [`crate::vm::StubVmBootRunner`].
pub struct StubContainerBootRunner;
impl BootRunner for StubContainerBootRunner {
fn spawn(&self, target: &BootTarget) -> Result<Box<dyn BootProc>, String> {
let BootTarget::Container(spec) = target else {
return Err(format!("StubContainerBootRunner cannot boot a {} target", target.mode().as_str()));
};
Err(format!(
"container boot '{}' not wired in this build: rebuild with `--features container-oci` to boot \
THROUGH draupnir's OCI backend (draupnir::container::ContainerBoot → the shared bollard engine)",
spec.image,
))
}
}
// ---------------------------------------------------------------------------
// run_container — bring a container up, run to completion, AS a tracked job.
// ---------------------------------------------------------------------------
/// Outcome of a [`run_container`] job: the exit code (None = no code / gone before
/// one was read) plus the captured stdout/stderr (truncated to keep `detail_json`
/// bounded).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ContainerOutcome {
pub exit_code: Option<i64>,
pub stdout: String,
pub stderr: String,
}
/// Per-stream capture cap so a chatty container can't bloat the redb value. The
/// head is kept (where errors usually are).
const LOG_CAP_BYTES: usize = 16 * 1024;
fn truncate_log(mut s: String) -> String {
if s.len() > LOG_CAP_BYTES {
s.truncate(LOG_CAP_BYTES);
s.push_str("\n…[truncated]");
}
s
}
/// Run a container **as a tracked job**, zero-shell: open a `running` row, then route
/// through [`draupnir::container::run_to_completion`] (the ONE OCI engine) to bring
/// `spec` up, wait for it to exit, capture its logs + exit code, remove it, and write
/// the terminal row (`done` on exit 0, else `failed`). Returns the [`ContainerOutcome`]
/// even on a non-zero exit (the *job* failed but the call succeeded — inspect
/// `exit_code`); only a boot/connect error (or, if the caller ever bounds it, a
/// timeout) returns `Err` (and the [`crate::jobs::JobHandle`] records `failed`).
///
/// Without the `container-oci` feature draupnir's engine is not linked, so this
/// fails honestly (an `Err` naming the missing backend) rather than faking a run.
///
/// `target` is a human label for the row (e.g. the image); `workspace` scopes it.
/// No [`Health`] reporting — see [`run_container_with_reports`] for that (this just
/// forwards to it with a [`crate::jobs::ReportSink::noop`]; kept as its own function
/// so every existing caller compiles unchanged).
pub fn run_container(
sink: crate::jobs::JobSink,
spec: &ContainerSpec,
target: &str,
workspace: &str,
) -> anyhow::Result<ContainerOutcome> {
run_container_with_reports(sink, crate::jobs::ReportSink::noop(), spec, target, workspace)
}
/// Like [`run_container`], but also wires the job's [`Reporter`] (built over
/// `report_sink`) to emit [`Health`] as the container runs to completion:
/// [`Health::Starting`] before the run, then [`Health::Gone`] on a clean exit or
/// [`Health::Failed`] on a non-zero exit / any boot error.
pub fn run_container_with_reports(
sink: crate::jobs::JobSink,
report_sink: crate::jobs::ReportSink,
spec: &ContainerSpec,
target: &str,
workspace: &str,
) -> anyhow::Result<ContainerOutcome> {
// The production engine (`draupnir::container::ContainerBoot`) via its
// mount-aware run: the spec's bind mounts ([`ContainerSpec::bind_strings`]) are
// threaded through so a routed job keeps its `-v` mounts. Empty binds ⇒
// byte-identical to the mount-less run (draupnir guarantees identical
// `HostConfig`), so an existing mount-less caller is unchanged.
let backend = draupnir::container::ContainerBoot::new();
let binds = spec.bind_strings();
run_recorded(sink, report_sink, spec, target, workspace, || {
backend.run_to_completion_with_binds(
&spec.to_draupnir(),
&draupnir::container::RunOptions::default(),
&binds,
)
})
}
/// The generic run-to-completion job over any draupnir backend — the mock-testable
/// seam (a scripted `Boot`+`ContainerControl` drives the whole ledger path with no
/// daemon). Binds are **not** carried here (draupnir's `Boot` trait has no
/// mount-aware boot; only the concrete `ContainerBoot::run_to_completion_with_binds`
/// used by [`run_container_with_reports`] does), so this is the mount-less generic
/// path used by the mock tests.
#[cfg(test)]
fn run_container_on<B>(
backend: &B,
sink: crate::jobs::JobSink,
report_sink: crate::jobs::ReportSink,
spec: &ContainerSpec,
target: &str,
workspace: &str,
) -> anyhow::Result<ContainerOutcome>
where
B: draupnir::Boot + draupnir::container::ContainerControl,
{
run_recorded(sink, report_sink, spec, target, workspace, || {
// `RunOptions::default()` waits indefinitely (parity with the old blocking
// `wait_container`).
draupnir::container::run_to_completion(
backend,
&spec.to_draupnir(),
&draupnir::container::RunOptions::default(),
)
})
}
/// Open a `running` ledger row for a container job, drive the supplied engine `run`
/// (the ONE draupnir engine: boot → wait for exit → capture split logs + exit code →
/// remove), and write the terminal row (`done` on exit 0, else `failed`) — the
/// single ledger wrap both the generic ([`run_container_on`]) and the production
/// mount-aware ([`run_container_with_reports`]) paths reuse (L5: one ledger flow, two
/// engine calls). Emits `Health` as it goes.
fn run_recorded(
sink: crate::jobs::JobSink,
report_sink: crate::jobs::ReportSink,
spec: &ContainerSpec,
target: &str,
workspace: &str,
run: impl FnOnce() -> draupnir::Result<draupnir::container::RunOutcome>,
) -> anyhow::Result<ContainerOutcome> {
use crate::jobs::{kind, JobHandle};
let job = JobHandle::start(
sink,
kind::CONTAINER,
target,
workspace,
serde_json::json!({ "image": spec.image, "cmd": spec.cmd }),
)
.with_reports(report_sink);
let reporter = job.reporter();
reporter.health(Health::Starting);
match run() {
Ok(outcome) => {
let code = outcome.exit_code;
let clean = code == Some(0);
reporter.health(if clean { Health::Gone } else { Health::Failed });
let result = ContainerOutcome {
exit_code: code,
stdout: truncate_log(outcome.stdout.join("\n")),
stderr: truncate_log(outcome.stderr.join("\n")),
};
let detail = serde_json::json!({
"image": spec.image,
"exit_code": code,
"stdout": result.stdout,
"stderr": result.stderr,
});
if clean {
job.finish(detail, "");
} else {
job.fail_with_detail(detail);
}
Ok(result)
}
Err(e) => {
reporter.health(Health::Failed);
let err = anyhow::anyhow!("{e}");
job.fail(&err);
Err(err)
}
}
}
// ---------------------------------------------------------------------------
// Tests — a mock draupnir backend drives the whole lifecycle with no daemon.
// ---------------------------------------------------------------------------
#[cfg(test)]
mod tests {
use super::*;
use std::sync::{Arc, Mutex};
// -----------------------------------------------------------------------
// DraupnirContainerBootRunner + run_container — the jera→draupnir delegation
// path. A mock draupnir backend proves the routing with NO daemon and NO
// feature: it exercises only draupnir's always-compiled `Boot` +
// `ContainerControl` traits.
// -----------------------------------------------------------------------
#[derive(Default)]
struct MockDraupnirInner {
booted: Vec<draupnir::BootSpec>,
stopped: Vec<String>,
state: Option<draupnir::container::ContainerState>,
logs: Vec<String>,
fail_boot: Option<String>,
/// The assembled `["exec", id, argv…]` commands draupnir's provided
/// `ContainerControl::exec` handed to this backend's `exec_command` hook —
/// records that the proc's machine id + the given argv actually reached the
/// engine (the knob-3 forward assertion).
exec_commands: Vec<Vec<String>>,
/// The scripted [`ExecOutcome`] the overridden `exec_command` returns (e.g.
/// `PONG`, exit 0). `None` ⇒ a default clean empty outcome.
exec_outcome: Option<ExecOutcome>,
}
/// A fake draupnir OCI backend: records the specs it was handed to boot + the
/// machines it was asked to stop, and reports a scriptable container state / logs
/// — so the jera→draupnir wiring is driven with no real podman.
#[derive(Clone, Default)]
struct MockDraupnirBackend {
inner: Arc<Mutex<MockDraupnirInner>>,
}
impl MockDraupnirBackend {
fn set_state(&self, s: draupnir::container::ContainerState) {
self.inner.lock().unwrap().state = Some(s);
}
fn set_logs(&self, lines: &[&str]) {
self.inner.lock().unwrap().logs = lines.iter().map(|s| s.to_string()).collect();
}
fn booted(&self) -> Vec<draupnir::BootSpec> {
self.inner.lock().unwrap().booted.clone()
}
fn stopped(&self) -> Vec<String> {
self.inner.lock().unwrap().stopped.clone()
}
/// Script the [`ExecOutcome`] an `exec` forward returns.
fn set_exec_outcome(&self, o: ExecOutcome) {
self.inner.lock().unwrap().exec_outcome = Some(o);
}
/// The assembled exec commands (`["exec", id, argv…]`) the forward delivered.
fn exec_commands(&self) -> Vec<Vec<String>> {
self.inner.lock().unwrap().exec_commands.clone()
}
}
impl draupnir::Boot for MockDraupnirBackend {
fn boot(&self, spec: &draupnir::BootSpec) -> draupnir::Result<draupnir::Machine> {
let mut g = self.inner.lock().unwrap();
if let Some(reason) = g.fail_boot.clone() {
return Err(draupnir::Error::Backend(reason));
}
g.booted.push(spec.clone());
Ok(draupnir::Machine::started(format!("draupnir-{}", spec.name), spec))
}
}
impl draupnir::container::ContainerControl for MockDraupnirBackend {
fn container_state(&self, _m: &draupnir::Machine) -> draupnir::container::ContainerState {
self.inner
.lock()
.unwrap()
.state
.clone()
.unwrap_or(draupnir::container::ContainerState::Running)
}
fn drain_logs(&self, _m: &draupnir::Machine) -> Vec<String> {
std::mem::take(&mut self.inner.lock().unwrap().logs)
}
fn stop(&self, m: &draupnir::Machine) {
self.inner.lock().unwrap().stopped.push(m.id.clone());
}
/// Record the assembled `["exec", id, argv…]` command draupnir's provided
/// `exec` handed down (after its container-only + non-empty-argv guards), and
/// hand back the scripted [`ExecOutcome`] — the daemon-free stand-in for the
/// live bollard `/exec` op, so the jera→draupnir exec forward is proven with
/// no podman.
fn exec_command(&self, command: &[String]) -> draupnir::Result<ExecOutcome> {
let mut g = self.inner.lock().unwrap();
g.exec_commands.push(command.to_vec());
Ok(g.exec_outcome.clone().unwrap_or_default())
}
}
/// The spec mapping jera→draupnir carries image, a unique per-boot name, and the
/// run overrides (cmd / env / ports) — the container analogue of vm.rs's
/// `to_draupnir`.
#[test]
fn container_spec_maps_onto_a_draupnir_container_spec() {
let spec = ContainerSpec {
image: "docker.io/library/nginx:alpine".into(),
cmd: vec!["nginx".into(), "-g".into(), "daemon off;".into()],
env: BTreeMap::from([("TZ".to_string(), "UTC".to_string())]),
ports: vec![8080, 8443],
..Default::default()
};
let d = spec.to_draupnir();
assert_eq!(d.kind, draupnir::Kind::Container);
assert_eq!(d.image, draupnir::ImageSource::OciImage("docker.io/library/nginx:alpine".into()));
assert!(d.name.starts_with("jera-boot-"), "empty name mints a unique instance: {}", d.name);
assert_eq!(d.cmd, vec!["nginx", "-g", "daemon off;"]);
assert_eq!(d.ports, vec![8080, 8443]);
assert_eq!(d.env.get("TZ").map(String::as_str), Some("UTC"));
d.validate().expect("mapped container spec is valid");
}
/// **Per-zone distinct `host:container` port map reaches draupnir (the deferred
/// bug fix).** FalkorDB listens on `6379` INSIDE every zone's container; a
/// per-zone host offset must publish `host:6379` (a DISTINCT map), not
/// `host:host` (which never reaches the container). `with_zone_port(Zone::Test,
/// 6379)` must thread onto `draupnir::BootSpec.port_maps` as
/// `PortMap { host: 6380, container: 6379 }`; `Demo` stays `6379:6379`; and the
/// single-port `ports` path still yields `6379:6379`. **RED-when-broken:**
/// neutralize the `d.port_maps = …` line in `to_draupnir` and the Test/Prod
/// arms fail; if `with_zone_port` bound `host:host` the container arm flips.
/// **A jera mount reaches the draupnir BootSpec the long-lived boot runs on.**
///
/// RED direction: delete the `d.binds = self.bind_strings()` line in
/// [`ContainerSpec::to_draupnir`] and the first assertion fails with an empty
/// vector — which is exactly the state that lost korp's FalkorDB `/data` volume on
/// oden 2026-09-10 (`podman inspect … Mounts: []`, `DBSIZE 0` after a recreate).
/// The run-to-completion leg had the binds all along; this pins the leg that did
/// not.
#[test]
fn a_mount_reaches_the_boot_spec_the_long_lived_runner_boots() {
let spec = ContainerSpec {
mounts: vec![BindMount::rw("korp-falkordb-data", "/data"), BindMount::ro("/etc/certs", "/certs")],
..ContainerSpec::image("localhost/falkordb:9")
};
let d = spec.to_draupnir();
assert_eq!(
d.binds,
vec!["korp-falkordb-data:/data".to_string(), "/etc/certs:/certs:ro".to_string()],
"the BootSpec `Boot::boot` receives must carry every mount, as -v strings"
);
assert_eq!(d.binds, spec.bind_strings(), "one renderer for both legs");
assert!(
ContainerSpec::image("img").to_draupnir().binds.is_empty(),
"no mounts ⇒ no binds ⇒ the create body stays byte-identical"
);
}
#[test]
fn zone_offset_port_map_threads_distinct_host_container_onto_draupnir() {
use nornir_zone::Zone;
// Test zone: publish host 6380 -> container 6379 (the fix). d.ports stays
// empty; the distinct map carries the pair.
let test = ContainerSpec::image("localhost/nordisk-falkordb-valkey:9-20260724")
.with_zone(Zone::Test)
.with_name(Zone::Test.container_name("korp-falkordb"))
.with_zone_port(Zone::Test, 6379);
assert_eq!(test.port_maps, vec![PortMap::new(6380, 6379)], "test: host 6380 -> container 6379");
assert_eq!(test.name, "korp-falkordb-test", "per-zone container name");
let dt = test.to_draupnir();
assert!(dt.ports.is_empty(), "distinct map does not populate the same-port field");
assert_eq!(
dt.port_maps,
vec![draupnir::PortMap::new(6380, 6379)],
"the (host 6380, container 6379) map reaches draupnir -> podman -p 6380:6379",
);
dt.validate().expect("test-zone FalkorDB spec is valid");
// Prod zone: host 6381 -> container 6379.
let prod = ContainerSpec::image("localhost/nordisk-falkordb-valkey:9-20260724").with_zone_port(Zone::Prod, 6379);
assert_eq!(
prod.to_draupnir().port_maps,
vec![draupnir::PortMap::new(6381, 6379)],
"prod: host 6381 -> container 6379",
);
// Demo zone: host == container (6379:6379) — byte-identical to the pre-zone
// publish; the demo path that already worked is unchanged.
let demo = ContainerSpec::image("localhost/nordisk-falkordb-valkey:9-20260724").with_zone_port(Zone::Demo, 6379);
assert_eq!(demo.port_maps, vec![PortMap::new(6379, 6379)], "demo: host == container");
demo.to_draupnir().validate().expect("demo-zone spec is valid");
// The single-port `ports` path still yields 6379:6379 (no port_maps) — the
// pre-existing same-host-port form is untouched.
let single = ContainerSpec { ports: vec![6379], ..ContainerSpec::image("redis:7") };
let ds = single.to_draupnir();
assert_eq!(ds.ports, vec![6379], "single-port form carried as host==container");
assert!(ds.port_maps.is_empty(), "single-port form uses no distinct maps");
// No maps + no ports = serializes byte-identically (no `port_maps` key).
let plain = ContainerSpec::image("redis:7");
assert!(plain.to_draupnir().port_maps.is_empty(), "default: no port maps");
assert!(
!serde_json::to_string(&plain).unwrap().contains("port_maps"),
"unset port_maps skipped (byte-identical)",
);
// A distinct map round-trips through serde (the wire form korp hands jera).
let rt: ContainerSpec = serde_json::from_str(&serde_json::to_string(&test).unwrap()).unwrap();
assert_eq!(rt.port_maps, test.port_maps, "port_maps round-trip through serde");
crate::assert_emit!(
"jera-container",
"zone_offset_port_map_reaches_draupnir",
dt.port_maps == vec![draupnir::PortMap::new(6380, 6379)] && ds.port_maps.is_empty(),
"with_zone_port(Test,6379) threads a distinct host:container map (6380:6379) onto draupnir -> per-zone FalkorDB reachable; single-port stays 6379:6379"
);
}
/// **Resource knobs + zone reach draupnir (the hot-infra + zones wire).**
/// `ContainerSpec`'s `cpus`/`mem_mb` thread onto `draupnir::BootSpec.cpus`/
/// `mem_limit_mb` via `to_draupnir()` (so they reach the live podman
/// `--cpus`/`--memory`), and the default (unset) maps to `None` = all host cores /
/// unconstrained (the container is never throttled to one core) with a
/// byte-identical create body downstream. The `zone` label rides on the jera spec
/// (draupnir has no zone) and serializes only when set. **RED-when-broken:**
/// neutralize the `d.cpus`/`d.mem_limit_mb` lines in `to_draupnir` and the hot arm
/// fails; make `zone` always serialize and the byte-identical arm fails.
#[test]
fn resources_and_zone_reach_draupnir_and_default_is_all_cores() {
// Hot FalkorDB: all 12 Loki cores + 16 GiB, published :6379, zone=demo.
let hot = ContainerSpec {
ports: vec![6379],
cpus: Some(12.0),
mem_mb: Some(16384),
..ContainerSpec::image("localhost/nordisk-falkordb-valkey:9-20260724")
}
.with_zone(nornir_zone::Zone::Demo);
let d = hot.to_draupnir();
assert_eq!(d.cpus, Some(12.0), "cpus reaches draupnir → --cpus 12");
assert_eq!(d.mem_limit_mb, Some(16384), "mem reaches draupnir → --memory 16384MiB");
assert_eq!(d.ports, vec![6379], "published port carried");
assert_eq!(hot.zone, "demo", "zone label set from nornir_zone::Zone");
d.validate().expect("hot FalkorDB spec is valid");
// Default (no caps) → draupnir None = ALL host cores, unconstrained memory.
let plain = ContainerSpec::image("redis:7");
let dp = plain.to_draupnir();
assert_eq!(dp.cpus, None, "default = all host cores (never throttled to one)");
assert_eq!(dp.mem_limit_mb, None, "default = unconstrained memory");
// Byte-identical serialize when no zone/caps are set (existing spec unchanged).
let js = serde_json::to_string(&plain).unwrap();
assert!(!js.contains("zone") && !js.contains("cpus") && !js.contains("mem_mb"), "unset knobs skipped: {js}");
// `with_all_host_cores` states the cap explicitly (== available_parallelism).
let explicit = ContainerSpec::image("redis:7").with_all_host_cores();
assert!(explicit.cpus.unwrap() >= 1.0, "all-host-cores sets a positive cap");
crate::assert_emit!(
"jera-container",
"resources_and_zone_reach_draupnir",
d.cpus == Some(12.0) && d.mem_limit_mb == Some(16384) && dp.cpus.is_none(),
"ContainerSpec cpus/mem thread onto draupnir::BootSpec (→ podman --cpus/--memory); default = all host cores (hot infra never throttled)"
);
}
/// **NetMode (airgap knob).** `NetMode::None` renders `--network none`; the
/// default renders no flag AND serializes byte-identically (no `net` key), so an
/// existing NAT spec is unchanged. RED-when-broken: a wrong mapping (e.g. `None`
/// → the wrong string, or the default emitting a flag) fails here.
#[test]
fn net_mode_renders_network_arg_and_default_is_byte_identical() {
// The value mapping onto the OCI `--network` arg.
assert_eq!(NetMode::default(), NetMode::Default, "Default is the derived default");
assert_eq!(NetMode::Default.oci_value(), None, "engine default emits no --network");
assert_eq!(NetMode::None.oci_value(), Some("none"), "airgap -> none");
assert_eq!(NetMode::Host.oci_value(), Some("host"));
assert_eq!(NetMode::Bridge.oci_value(), Some("bridge"));
// Default spec: no arg, and no `net` key in the serialized form (unchanged).
let nat = ContainerSpec::image("nginx:alpine");
assert_eq!(nat.network_arg(), None, "default spec renders no --network flag");
let nat_json = serde_json::to_string(&nat).unwrap();
assert!(!nat_json.contains("net"), "default net is skipped -> byte-identical: {nat_json}");
// Airgap spec: --network none renders + round-trips through serde.
let airgap = ContainerSpec { net: NetMode::None, ..ContainerSpec::image("nginx:alpine") };
assert_eq!(airgap.network_arg().as_deref(), Some("--network none"), "airgap -> --network none");
let rt: ContainerSpec = serde_json::from_str(&serde_json::to_string(&airgap).unwrap()).unwrap();
assert_eq!(rt.net, NetMode::None, "net round-trips through serde");
crate::assert_emit!(
"jera-container",
"container_spec_net_mode_renders_network_none",
airgap.network_arg().as_deref() == Some("--network none") && nat.network_arg().is_none(),
"NetMode::None -> `--network none`; default emits no flag (airgap knob available to routers)"
);
}
/// **Net wire → draupnir (the airgap unblocker).** `to_draupnir()` threads this
/// spec's [`NetMode`] onto the `draupnir::BootSpec` that BOTH jera legs consume
/// (the [`DraupnirContainerBootRunner`] boot AND [`run_container`]'s
/// `run_to_completion_with_binds`), so a routed [`NetMode::None`] spec produces a
/// live `--network none` airgap container. Asserts the mapped
/// `draupnir::BootSpec.net` for every mode, and that the default maps to
/// `draupnir::NetMode::Default` (byte-identical downstream create body). Because
/// both legs build the draupnir spec via this one `to_draupnir()`, this single
/// assertion covers both legs. **RED-when-broken:** neutralize the `d.net = …`
/// line in `to_draupnir` (so `net` always stays `Default`) and the None/Host/
/// Bridge arms fail here.
#[test]
fn container_net_reaches_draupnir() {
// Default net maps to draupnir's Default (no --network flag downstream).
let nat = ContainerSpec::image("nginx:alpine");
assert_eq!(nat.to_draupnir().net, draupnir::NetMode::Default, "default net -> draupnir Default (byte-identical)");
// Airgap: NetMode::None reaches draupnir as NetMode::None (live --network none).
let airgap = ContainerSpec { net: NetMode::None, ..ContainerSpec::image("nginx:alpine") };
assert_eq!(airgap.to_draupnir().net, draupnir::NetMode::None, "None -> draupnir None (live --network none)");
// Host / Bridge carry across too (full-surface parity).
let host = ContainerSpec { net: NetMode::Host, ..ContainerSpec::image("x:1") };
assert_eq!(host.to_draupnir().net, draupnir::NetMode::Host, "Host -> draupnir Host");
let bridge = ContainerSpec { net: NetMode::Bridge, ..ContainerSpec::image("x:1") };
assert_eq!(bridge.to_draupnir().net, draupnir::NetMode::Bridge, "Bridge -> draupnir Bridge");
crate::assert_emit!(
"jera-container",
"container_net_reaches_draupnir",
airgap.to_draupnir().net == draupnir::NetMode::None
&& nat.to_draupnir().net == draupnir::NetMode::Default,
"ContainerSpec.net threads onto draupnir::BootSpec.net via to_draupnir -> live --network none for a routed airgap job"
);
}
/// **Bind mounts.** A `BindMount` renders `host:container[:ro]`; a spec's
/// `bind_strings()` is the exact `&[String]` draupnir's
/// `run_to_completion_with_binds` consumes, and empty mounts render `[]` (the
/// mount-less run is byte-identical). RED-when-broken: a wrong `:ro` placement or
/// a leaked separator fails here.
#[test]
fn bind_mounts_render_bind_strings_and_empty_is_unchanged() {
assert_eq!(BindMount::rw("/data", "/data").bind_string(), "/data:/data", "rw = host:container");
assert_eq!(BindMount::ro("/etc/certs", "/certs").bind_string(), "/etc/certs:/certs:ro", "ro appends :ro");
// No mounts -> empty binds -> byte-identical mount-less run + no `mounts` key.
let plain = ContainerSpec::image("busybox:latest");
assert!(plain.bind_strings().is_empty(), "no mounts -> no binds");
assert!(!serde_json::to_string(&plain).unwrap().contains("mounts"), "mounts skipped when empty");
// A mounted spec renders the exact draupnir/-v bind list, ro flag preserved.
let mounted = ContainerSpec {
mounts: vec![BindMount::rw("/host/in", "/in"), BindMount::ro("/host/ro", "/ro")],
..ContainerSpec::image("packer:1")
};
assert_eq!(
mounted.bind_strings(),
vec!["/host/in:/in".to_string(), "/host/ro:/ro:ro".to_string()],
"mounts render in order with ro preserved"
);
// Round-trips through serde (the wire form a router hands jera).
let rt: ContainerSpec = serde_json::from_str(&serde_json::to_string(&mounted).unwrap()).unwrap();
assert_eq!(rt.mounts, mounted.mounts, "mounts round-trip through serde");
crate::assert_emit!(
"jera-container",
"container_spec_bind_mounts_render_v_binds",
mounted.bind_strings() == vec!["/host/in:/in".to_string(), "/host/ro:/ro:ro".to_string()]
&& plain.bind_strings().is_empty(),
"BindMount -> `host:container[:ro]`; empty = mount-less run unchanged (mounts knob available to routers)"
);
}
#[test]
fn draupnir_runner_boots_through_draupnir_streams_logs_and_kills() {
let backend = MockDraupnirBackend::default();
backend.set_logs(&["nginx up", "listening on :8080"]);
let runner = DraupnirContainerBootRunner::with_backend(backend.clone());
let target = BootTarget::Container(ContainerSpec {
image: "nginx:alpine".into(),
cmd: vec!["nginx".into()],
env: BTreeMap::from([("TZ".to_string(), "UTC".to_string())]),
ports: vec![8080],
..Default::default()
});
let mut proc = runner.spawn(&target).expect("draupnir spawn");
// The boot was routed to draupnir with the mapped spec.
let booted = backend.booted();
assert_eq!(booted.len(), 1, "one boot routed through draupnir");
assert_eq!(booted[0].image, draupnir::ImageSource::OciImage("nginx:alpine".into()));
assert_eq!(booted[0].cmd, vec!["nginx"]);
assert_eq!(booted[0].ports, vec![8080]);
assert_eq!(proc.poll(), BootStatus::Running, "container is up (draupnir state)");
assert_eq!(proc.drain_log(), vec!["nginx up", "listening on :8080"], "draupnir-streamed logs drained");
proc.kill();
assert_eq!(proc.poll(), BootStatus::Killed, "kill is terminal");
assert_eq!(backend.stopped().len(), 1, "kill routed to draupnir stop");
proc.kill();
assert_eq!(backend.stopped().len(), 1, "kill is idempotent");
}
#[test]
fn draupnir_runner_maps_container_state_to_terminal_status() {
let make = || {
let backend = MockDraupnirBackend::default();
let runner = DraupnirContainerBootRunner::with_backend(backend.clone());
(backend, runner)
};
let ctr = || BootTarget::Container(ContainerSpec::image("redis:7"));
let (backend, runner) = make();
backend.set_state(draupnir::container::ContainerState::Exited(0));
let mut p = runner.spawn(&ctr()).unwrap();
assert_eq!(p.poll(), BootStatus::BootedOk, "clean exit -> BootedOk");
let (backend, runner) = make();
backend.set_state(draupnir::container::ContainerState::Exited(137));
let mut p = runner.spawn(&ctr()).unwrap();
assert!(matches!(p.poll(), BootStatus::Failed(r) if r.contains("137")), "crash -> Failed(code)");
let (backend, runner) = make();
backend.set_state(draupnir::container::ContainerState::Gone);
let mut p = runner.spawn(&ctr()).unwrap();
assert!(matches!(p.poll(), BootStatus::Failed(_)), "gone -> Failed");
}
#[test]
fn draupnir_runner_boot_failure_is_reported() {
let backend = MockDraupnirBackend::default();
backend.inner.lock().unwrap().fail_boot = Some("no such image".into());
let runner = DraupnirContainerBootRunner::with_backend(backend);
let err = runner
.spawn(&BootTarget::Container(ContainerSpec::image("ghost:0")))
.err()
.expect("boot failure surfaces");
assert!(err.contains("no such image"), "reason carried: {err}");
}
/// The draupnir-routed proc emits `Health` through its lifecycle: `Starting`
/// before boot, `Alive` once up, `Gone` on clean exit.
#[test]
fn draupnir_runner_emits_health_starting_alive_then_gone() {
let backend = MockDraupnirBackend::default();
let runner = DraupnirContainerBootRunner::with_backend(backend.clone());
let registry = crate::jobs::report::new_registry();
let reporter = Reporter::new("svc-d", crate::jobs::ReportSink::registry(registry.clone()));
let latest = |reg: &crate::jobs::ReportRegistry| reg.lock().unwrap().get("svc-d").and_then(|r| r.health());
let mut proc = runner
.spawn_spec_with_reporter(&ContainerSpec::image("falkordb:edge"), reporter)
.expect("draupnir spawn with reporter");
assert_eq!(latest(®istry), Some(Health::Alive), "up right after draupnir boot");
assert_eq!(proc.poll(), BootStatus::Running);
backend.set_state(draupnir::container::ContainerState::Exited(0));
assert_eq!(proc.poll(), BootStatus::BootedOk);
assert_eq!(latest(®istry), Some(Health::Gone), "clean exit -> Gone, not Failed");
}
/// **Exec forward — jera "knob 3" (the detached-container control primitive).**
/// A detached container proc's [`BootProc::exec`] forwards `argv` to draupnir's
/// ONE OCI engine over the `(backend, machine)` handle it already holds:
/// `self.backend.exec(&self.machine, argv)`. Asserts the proc's own machine id
/// (`draupnir-cache`) AND the given argv reach the backend as the assembled
/// `["exec", id, argv…]` command, and that the engine's [`ExecOutcome`] (exit code
/// plus stdout) maps straight back. Driven by the mock's `exec_command` hook: the
/// full guard, assembly and forward path, NO daemon and NO feature. The live
/// `podman exec` is a Loki integration concern, deferred.
///
/// **RED-when-broken:** neutralize the forward (e.g. pass a wrong/empty argv or a
/// different machine than `self.machine`) and the recorded command no longer
/// equals `["exec", "draupnir-cache", "redis-cli", "ping"]` — this test fails.
#[test]
fn proc_exec_forwards_machine_and_argv_to_draupnir_and_returns_outcome() {
let backend = MockDraupnirBackend::default();
backend.set_exec_outcome(ExecOutcome {
exit_code: Some(0),
stdout: vec!["PONG".into()],
stderr: vec![],
});
let runner = DraupnirContainerBootRunner::with_backend(backend.clone());
// Spawn detached (stays up), get the proc, exec INTO it — the exact shape a
// Skidbladnir lifecycle-swap uses.
let proc = runner
.spawn_spec(&ContainerSpec { name: "cache".into(), ..ContainerSpec::image("redis:7") })
.expect("detached container boot");
let out = proc.exec(&["redis-cli", "ping"]).expect("exec forwards to draupnir");
// The engine outcome maps straight back (non-zero would be in exit_code, not Err).
assert_eq!(out.exit_code, Some(0), "exec exit code carried back");
assert_eq!(out.stdout, vec!["PONG".to_string()], "exec stdout carried back");
// The proc's OWN machine id + the given argv reached draupnir as the canonical
// `podman exec` command — proves the forward passed `self.machine` and `argv`.
assert_eq!(
backend.exec_commands(),
vec![vec![
"exec".to_string(),
"draupnir-cache".to_string(),
"redis-cli".to_string(),
"ping".to_string(),
]],
"the proc's machine id + argv reach draupnir's exec"
);
crate::assert_emit!(
"jera-container",
"proc_exec_forwards_machine_and_argv",
backend.exec_commands()
== vec![vec![
"exec".to_string(),
"draupnir-cache".to_string(),
"redis-cli".to_string(),
"ping".to_string(),
]]
&& out.exit_code == Some(0)
&& out.stdout == vec!["PONG".to_string()],
"DraupnirContainerProc::exec forwards (self.machine, argv) to draupnir's ONE OCI engine and returns its ExecOutcome -> Skidbladnir live start/stop/exec swap unblocked"
);
}
/// A non-container proc has no container to exec into: the default
/// [`BootProc::exec`] reports an honest error rather than pretending. Guards the
/// additive provided-method default (every non-container `BootProc` keeps working).
#[test]
fn default_boot_proc_exec_is_honest_not_wired() {
// The honest stub proc from the boot module has no exec capability.
struct PlainProc;
impl BootProc for PlainProc {
fn poll(&mut self) -> BootStatus {
BootStatus::Running
}
fn kill(&mut self) {}
fn drain_log(&mut self) -> Vec<String> {
Vec::new()
}
}
let err = PlainProc.exec(&["true"]).expect_err("a plain proc cannot exec");
assert!(err.contains("does not support exec"), "honest not-wired: {err}");
}
/// Without `container-oci`, `StubContainerBootRunner` reports an honest,
/// feature-naming failure — it never fakes a boot (mirror of `StubVmBootRunner`).
#[test]
fn stub_container_runner_reports_honest_not_wired() {
let err = StubContainerBootRunner
.spawn(&BootTarget::Container(ContainerSpec::image("redis:7")))
.err()
.expect("stub must not pretend to boot");
assert!(err.contains("draupnir"), "names the boot library it routes through: {err}");
assert!(err.contains("container-oci"), "names the feature: {err}");
}
// -----------------------------------------------------------------------
// run_container — the run-to-completion job, routed through draupnir's
// run_to_completion. Driven by the mock backend (no daemon, no feature).
// -----------------------------------------------------------------------
/// Collect every ledger record a `JobSink` is handed (the thin-client seam).
fn capturing_sink() -> (crate::jobs::JobSink, Arc<Mutex<Vec<crate::jobs::JobRecord>>>) {
let recs: Arc<Mutex<Vec<crate::jobs::JobRecord>>> = Arc::new(Mutex::new(Vec::new()));
let cap = recs.clone();
let sink = crate::jobs::JobSink::remote(move |r: &crate::jobs::JobRecord| {
cap.lock().unwrap().push(r.clone());
});
(sink, recs)
}
/// A clean run routes through draupnir's `run_to_completion`: the spec is booted
/// through draupnir, its logs are captured, exit 0 → `ContainerOutcome` +
/// a terminal `done` ledger row.
#[test]
fn run_container_routes_through_draupnir_and_records_a_done_row() {
let backend = MockDraupnirBackend::default();
backend.set_logs(&["hello from the container"]);
backend.set_state(draupnir::container::ContainerState::Exited(0));
let (sink, recs) = capturing_sink();
let registry = crate::jobs::report::new_registry();
let outcome = run_container_on(
&backend,
sink,
crate::jobs::ReportSink::registry(registry.clone()),
&ContainerSpec::image("busybox:latest"),
"busybox:latest",
"ws",
)
.expect("run to completion");
assert_eq!(outcome.exit_code, Some(0), "clean exit code carried");
assert_eq!(outcome.stdout, "hello from the container", "draupnir logs captured");
assert_eq!(backend.booted().len(), 1, "one boot routed through draupnir");
assert_eq!(backend.stopped().len(), 1, "container removed after run");
let recs = recs.lock().unwrap();
assert_eq!(recs.first().unwrap().status, crate::jobs::status::RUNNING, "opens a running row");
assert_eq!(recs.last().unwrap().status, crate::jobs::status::DONE, "clean exit -> done");
}
/// A non-zero exit is `Ok` (the *call* succeeded) carrying the code, but the
/// ledger row is terminal `failed` — parity with the old bollard `run_container`.
#[test]
fn run_container_nonzero_exit_is_ok_but_records_a_failed_row() {
let backend = MockDraupnirBackend::default();
backend.set_state(draupnir::container::ContainerState::Exited(2));
let (sink, recs) = capturing_sink();
let outcome = run_container_on(
&backend,
sink,
crate::jobs::ReportSink::noop(),
&ContainerSpec::image("failer:1"),
"failer:1",
"ws",
)
.expect("non-zero exit is Ok, not Err");
assert_eq!(outcome.exit_code, Some(2), "non-zero code carried");
assert_eq!(recs.lock().unwrap().last().unwrap().status, crate::jobs::status::FAILED, "non-zero -> failed row");
}
/// **The korp submit-container-job API end-to-end (no daemon).**
/// [`submit_container_job`] opens a tracked job whose detail carries the korp
/// knobs (`image`/`zone`/`ports`/`cpus`/`mem_mb`), and
/// [`crate::boot::BootJobs::wait_until_running`] reports `Running` once the
/// container process is up. Driven by a [`crate::boot::FakeBootRunner`] so the
/// whole submit→poll→ready flow is proven with no real engine (the live
/// jera→draupnir→podman path is covered by the `to_draupnir` + draupnir engine
/// tests). **RED-when-broken:** drop the `zone`/`ports`/`cpus` from `open_detail`
/// and the detail asserts fail; break `wait_until_running`'s `Running` arm and the
/// ready assert fails.
#[test]
fn submit_container_job_records_knobs_and_waits_until_running() {
use crate::boot::{BootJobs, BootStatus, FakeBootRunner, FakeProc};
let (sink, recs) = capturing_sink();
let runner = std::sync::Arc::new(FakeBootRunner::yielding(FakeProc::running()));
let mut jobs = BootJobs::with_runner(runner, sink, "korp");
let spec = ContainerSpec {
ports: vec![6379],
cpus: Some(12.0),
mem_mb: Some(16384),
..ContainerSpec::image("localhost/nordisk-falkordb-valkey:9-20260724")
}
.with_zone(nornir_zone::Zone::Demo);
let id = super::submit_container_job(&mut jobs, spec);
// The opening job row carries the korp-facing knobs for a poller to read.
let opened = recs.lock().unwrap().first().unwrap().clone();
assert_eq!(opened.kind, crate::jobs::kind::BOOT_CONTAINER, "a container boot job");
let detail: serde_json::Value = serde_json::from_str(&opened.detail_json).unwrap();
assert_eq!(detail["zone"], "demo", "zone label recorded on the job");
assert_eq!(detail["ports"], serde_json::json!([6379]), "published ports recorded");
assert_eq!(detail["cpus"], 12.0, "cpu cap recorded");
assert_eq!(detail["mem_mb"], 16384, "memory cap recorded");
// The wait-for-running helper reports the container process is up.
let status = jobs.wait_until_running(
&id,
std::time::Duration::from_secs(2),
std::time::Duration::from_millis(1),
);
assert_eq!(status, BootStatus::Running, "submitted container job reaches Running");
assert_eq!(jobs.status_of(&id), Some(BootStatus::Running), "status_of agrees");
crate::assert_emit!(
"jera-container",
"submit_container_job_ready",
status == BootStatus::Running && detail["zone"] == "demo",
"submit_container_job tags the job with zone/ports/resources and wait_until_running reports the container is up"
);
}
/// A boot failure is an `Err` and records a `failed` row (no fake success).
#[test]
fn run_container_boot_failure_is_err_and_records_a_failed_row() {
let backend = MockDraupnirBackend::default();
backend.inner.lock().unwrap().fail_boot = Some("no such image".into());
let (sink, recs) = capturing_sink();
let err = run_container_on(
&backend,
sink,
crate::jobs::ReportSink::noop(),
&ContainerSpec::image("ghost:0"),
"ghost:0",
"ws",
)
.err()
.expect("boot failure surfaces as Err");
assert!(format!("{err:#}").contains("no such image"), "reason carried: {err:#}");
assert_eq!(recs.lock().unwrap().last().unwrap().status, crate::jobs::status::FAILED, "boot error -> failed row");
}
}