mikronova/draupnir
src/container.rs
//! **Container backend** — fire up an OCI container instance (e.g. a redis
//! service) over a container runtime.
//!
//! Draupnir does not reimplement a runtime. This backend is the thin adapter that
//! maps a Draupnir [`BootSpec`] onto **`bollard`** — the async podman/Docker REST
//! client `jera` already drives for its zero-shell container path — and runs the
//! container lifecycle over it. Reusing bollard keeps one container engine across
//! the constellation rather than a second bespoke one.
//!
//! It sits behind the `backend-oci` feature so the default build stays pure-std;
//! the trait wiring compiles unconditionally. **Zero-shell**: every operation is a
//! Rust API call over the podman/Docker socket — never a `podman`/`docker`
//! subprocess. If no daemon is reachable the backend **degrades with a clear
//! error** (there is deliberately no CLI fallback), it never fakes a boot.
use crate::{Boot, BootSpec, Error, ImageSource, Lifecycle, Machine, PowerState, Result};
// `PortMap` is referenced only inside the bollard-gated engine (create-body /
// create-and-start), so importing it unconditionally warns when `backend-oci` is
// off; the gated sites qualify it as `crate::PortMap` instead.
use std::path::Path;
use std::time::{Duration, Instant};
#[cfg(feature = "backend-oci")]
use std::sync::{Arc, Mutex};
/// **How draupnir decides a container is *app-ready*** — one level above the bare
/// `running` power state. [`ContainerBoot::wait_ready`] blocks on this until it
/// holds or the timeout elapses.
///
/// A freshly `create_and_start`ed container reports [`PowerState::On`] the instant
/// its main process is spawned, which is *not* the same as the app inside having
/// come up (a redis still opening its listen socket, a service still reading its
/// config). [`Readiness::Running`] is the base state poll (parity with the old
/// bare-`running` path); [`Readiness::LogMatch`] waits for the app to *announce*
/// itself on its own logs — the readiness signal the spec can opt into.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum Readiness {
/// Ready as soon as the container's main process reports `running`
/// ([`PowerState::On`]) — the base state poll, equivalent to the pre-existing
/// bare-`running` behaviour.
#[default]
Running,
/// Ready once `needle` appears anywhere in the container's stdout/stderr logs
/// (e.g. `"Ready to accept connections"` for redis) — a readiness probe the
/// spec supplies.
LogMatch(String),
}
/// **Poll `check` until it reports ready, or `timeout` elapses.** The pure,
/// backend-independent core of [`ContainerBoot::wait_ready`]: it owns the deadline
/// arithmetic + sleep cadence and turns a timeout into a clear [`Error::Backend`]
/// naming the instance and the elapsed budget. `check` returns `Ok(true)` when
/// ready, `Ok(false)` to keep polling, and `Err(..)` to fail fast (a backend error
/// is never swallowed as "not ready yet"). Unit-tested with no live daemon.
#[cfg_attr(not(feature = "backend-oci"), allow(dead_code))]
fn poll_until_ready(
label: &str,
timeout: Duration,
interval: Duration,
mut check: impl FnMut() -> Result<bool>,
) -> Result<()> {
let deadline = Instant::now() + timeout;
loop {
if check()? {
return Ok(());
}
let now = Instant::now();
if now >= deadline {
return Err(Error::Backend(format!(
"container `{label}` not ready within {timeout:?}"
)));
}
// Never sleep past the deadline.
let remaining = deadline.saturating_duration_since(now);
std::thread::sleep(interval.min(remaining));
}
}
/// **Who the container's process is and where it starts** — the create body's
/// `User` + `WorkingDir` pair, carried as one value.
///
/// One struct rather than two more positional arguments on each `create_body*` /
/// `create_and_start*` entry: those chains are already at the length that earns a
/// `#[allow(clippy::too_many_arguments)]`, and two `Option<&str>` in a row is the
/// shape where a caller silently swaps them. [`Default`] is both-`None`, which
/// renders a create body **byte-identical** to one written before this type
/// existed — that is what keeps every existing caller unchanged.
///
/// Read [`BootSpec::user`] and [`BootSpec::working_dir`] for *why* a container
/// needs to be able to say either; this is only the transport.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct ProcessIdentity {
/// podman `--user`: `uid`, `uid:gid`, or a name the image knows. `None` =
/// the image's own `USER`.
pub user: Option<String>,
/// podman `--workdir`. `None` = the image's own `WORKDIR`.
pub working_dir: Option<String>,
/// podman `--cap-add`, verbatim (see [`BootSpec::cap_add`]).
pub cap_add: Vec<String>,
/// podman `--device`, host paths (see [`BootSpec::devices`]).
pub devices: Vec<String>,
/// podman `--cap-drop` (see [`BootSpec::cap_drop`]).
pub cap_drop: Vec<String>,
/// podman `--read-only` (see [`BootSpec::read_only_rootfs`]).
pub read_only_rootfs: bool,
/// podman `--tmpfs` (see [`BootSpec::tmpfs`]).
pub tmpfs: Vec<(String, String)>,
/// `--pull=never` (see [`BootSpec::never_pull`]).
pub never_pull: bool,
}
impl ProcessIdentity {
/// The pair a [`BootSpec`] carries, for the boot path to hand to the engine.
pub fn of(spec: &BootSpec) -> Self {
Self {
user: spec.user.clone(),
working_dir: spec.working_dir.clone(),
cap_add: spec.cap_add.clone(),
devices: spec.devices.clone(),
cap_drop: spec.cap_drop.clone(),
read_only_rootfs: spec.read_only_rootfs,
tmpfs: spec.tmpfs.clone(),
never_pull: spec.never_pull,
}
}
/// True when nothing is named — the byte-identical no-op.
pub fn is_default(&self) -> bool {
self.user.is_none()
&& self.working_dir.is_none()
&& self.cap_add.is_empty()
&& self.devices.is_empty()
&& self.cap_drop.is_empty()
&& !self.read_only_rootfs
&& self.tmpfs.is_empty()
&& !self.never_pull
}
}
/// **Which of a container's two streams a line arrived on** — the tag a live log
/// sink ([`drive_to_completion_streaming`]) gets with each line, so a consumer
/// can keep stdout and stderr apart in flight the way [`RunOutcome`] keeps them
/// apart at the end.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogStream {
/// The container's stdout.
Stdout,
/// The container's stderr.
Stderr,
}
/// `spec` with `extra` appended to its [`BootSpec::binds`] — pure, so the merge the
/// mount-aware boot performs is testable without an engine. Order is kept: the spec's
/// own binds first, the caller's after, the way podman receives repeated `-v`.
pub fn with_extra_binds(spec: &BootSpec, extra: &[String]) -> BootSpec {
let mut merged = spec.clone();
merged.binds.extend(extra.iter().cloned());
merged
}
/// The OCI container boot backend.
#[derive(Default, Clone)]
pub struct ContainerBoot {
/// The connected engine (a bollard `Docker` + its own tokio runtime), built
/// lazily on first use and shared across [`Boot`]/[`Lifecycle`] calls.
#[cfg(feature = "backend-oci")]
engine: Arc<Mutex<Option<Arc<Engine>>>>,
}
impl std::fmt::Debug for ContainerBoot {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ContainerBoot").finish_non_exhaustive()
}
}
impl ContainerBoot {
/// Construct the container backend.
pub fn new() -> Self {
Self::default()
}
/// The OCI image reference this spec will pull + run.
pub fn image_ref<'a>(&self, spec: &'a BootSpec) -> Result<&'a str> {
match &spec.image {
ImageSource::OciImage(r) => Ok(r.as_str()),
other => Err(Error::Spec(format!(
"container backend needs an OCI image, got {other:?}"
))),
}
}
/// The per-instance container name derived from the spec name.
#[cfg_attr(not(feature = "backend-oci"), allow(dead_code))]
fn container_name(spec: &BootSpec) -> String {
Self::container_name_for(&spec.name)
}
/// **Every leftover container this backend REAPED since the last drain.**
///
/// The other half of the crash-safety preflight ([`ensure_name_free_over`]):
/// draupnir destroyed somebody's container, and a library that prints nothing
/// still must not do that silently. A build runner is expected to drain this
/// after a boot and put each [`Reap`] on its own output — that is how a human
/// learns that this run began by clearing the wreckage of the last one.
///
/// Empty when nothing was reaped, and when no engine has been connected yet
/// (asking does not connect one).
#[must_use]
pub fn drain_reaps(&self) -> Vec<Reap> {
#[cfg(feature = "backend-oci")]
{
let guard = self.engine.lock().unwrap();
match guard.as_ref() {
Some(engine) => engine.drain_reaps(),
None => Vec::new(),
}
}
#[cfg(not(feature = "backend-oci"))]
{
Vec::new()
}
}
/// **Block until the container is app-ready**, or return a clear timeout error.
///
/// Folds in `ContainerController`'s `wait_ready` readiness model: it polls the
/// container state (and, for [`Readiness::LogMatch`], scans its logs) on a fixed
/// cadence until `readiness` holds or `timeout` elapses. On timeout it returns an
/// [`Error::Backend`] naming the instance and the budget — never a fake "ready".
/// The bare-`running` boot path is unchanged; this is an additive step a caller
/// runs *after* [`Boot::boot`] when it needs the app up, not just the process.
pub fn wait_ready(
&self,
machine: &Machine,
readiness: &Readiness,
timeout: Duration,
) -> Result<()> {
#[cfg(feature = "backend-oci")]
{
let engine = self.engine()?;
let name = machine.id.clone();
let outcome =
poll_until_ready(
&name,
timeout,
Duration::from_millis(200),
|| match readiness {
Readiness::Running => {
Ok(matches!(engine.power_state(&name), PowerState::On))
}
Readiness::LogMatch(needle) => engine.log_contains(&name, needle),
},
);
// Record the readiness verdict: GREEN once the app is up, RED when the
// probe timed out or the backend errored — the wait_ready surface nornir's
// matrix reads back (a container that never became ready must be visible).
crate::functional_status(
"draupnir/container",
"wait_ready",
outcome.is_ok(),
&match &outcome {
Ok(()) => format!("container `{name}` reached {readiness:?}"),
Err(e) => format!("container `{name}` never became ready: {e}"),
},
);
outcome
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (machine, readiness, timeout);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature".into(),
))
}
}
/// The connected engine, built (and cached) on first use. `Err` when no
/// podman/Docker socket is reachable — the honest degrade, no shell fallback.
#[cfg(feature = "backend-oci")]
fn engine(&self) -> Result<Arc<Engine>> {
let mut guard = self.engine.lock().unwrap();
if let Some(e) = guard.as_ref() {
return Ok(Arc::clone(e));
}
let e = Arc::new(Engine::connect()?);
*guard = Some(Arc::clone(&e));
Ok(e)
}
}
impl Boot for ContainerBoot {
fn boot(&self, spec: &BootSpec) -> Result<Machine> {
spec.validate()?;
let image = self.image_ref(spec)?;
#[cfg(feature = "backend-oci")]
{
let name = Self::container_name(spec);
let env: Vec<String> = spec.env.iter().map(|(k, v)| format!("{k}={v}")).collect();
// Thread the spec's network mode + resource caps onto the live create
// body — all default (`None`) leaves the run byte-identical; `NetMode::None`
// → `--network none`; `cpus`/`mem_limit_mb` set the podman `--cpus`/
// `--memory` (unset = all host cores, unconstrained memory: the hot-infra
// default so FalkorDB/Spark are never throttled to one core).
// `user`/`working_dir` ride along the same way: unnamed (the default)
// leaves the create body byte-identical, and named they decide who
// owns the files the container writes to a bind mount and where its
// relative paths resolve from — see `BootSpec::user`.
self.engine()?.create_and_start_process(
image,
&name,
&env,
&spec.cmd,
&spec.ports,
&spec.port_maps,
&spec.binds,
spec.net.oci_value(),
spec.cpus,
spec.mem_limit_mb,
&spec.security_opt,
&ProcessIdentity::of(spec),
)?;
Ok(Machine::started(name, spec))
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = image;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
}
impl Lifecycle for ContainerBoot {
fn power_on(&self, machine: &Machine) -> Result<()> {
#[cfg(feature = "backend-oci")]
{
self.engine()?.start(&machine.id)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature".into(),
))
}
}
fn power_off(&self, machine: &Machine) -> Result<()> {
#[cfg(feature = "backend-oci")]
{
self.engine()?.stop(&machine.id);
Ok(())
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature".into(),
))
}
}
fn status(&self, machine: &Machine) -> Result<PowerState> {
#[cfg(feature = "backend-oci")]
{
Ok(self.engine()?.power_state(&machine.id))
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature".into(),
))
}
}
}
// ---------------------------------------------------------------------------
// ContainerControl — the exit-code-aware + log-streaming container seam.
// ---------------------------------------------------------------------------
/// The lifecycle state of a container as read from the engine — the richer
/// projection a **job runner** ([`jera`](https://git.gunnar.rs/nordisk/edda))
/// needs, one level below the generic power [`Lifecycle`] (which collapses every
/// non-running state to [`PowerState::Off`] and so cannot tell a clean exit from a
/// crash). Mirrors jera's own `EngineState` so a container boot's status/log model
/// is preserved verbatim when it delegates here.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ContainerState {
/// The container is created and/or running (up).
Running,
/// The container exited on its own with this code (0 = clean).
Exited(i64),
/// No such container — removed, or never created.
Gone,
}
/// **The daemon's own word for a container's state, mapped onto
/// [`ContainerState`] — and the whole reason draupnir reads the container list as
/// JSON rather than through bollard's generated model.**
///
/// MEASURED on oden 2026-09-10, from korp-server's journal:
///
/// ```text
/// draupnir could not list containers named `draupnir-korp-spark-iceberg`:
/// backend error: listing containers: Failed to deserialize JSON: unknown variant
/// `stopped`, expected one of ``, `created`, `running`, `paused`, `restarting`,
/// `exited`, `removing`, `dead` at line 1 column 3186
/// ```
///
/// `bollard::models::ContainerSummaryStateEnum` is generated from **Docker's**
/// OpenAPI document and is a CLOSED set of eight words with no `#[serde(other)]`
/// arm. podman's own state vocabulary (`libpod/define/containerstate.go`) is a
/// different set — it adds `configured`, `initialized`, `stopping` and `stopped`,
/// and `stopped` is what a container stopped through the API reports. serde fails
/// the whole `Vec<ContainerSummary>`, so ONE unrelated stopped container on the box
/// took out the entire listing: korp could not verify that
/// `draupnir-korp-spark-iceberg` was its own and refused to adopt it
/// (`korp/infra: backend spark41 NOT started`).
///
/// A closed enum owned by a third party is the wrong shape for a daemon that is
/// free to invent a word tomorrow, so the state arrives here as a `&str` and an
/// unknown word is an honest [`ContainerState::Gone`] rather than a parse failure
/// that loses the other twenty containers.
///
/// The mapping, and why each arm:
///
/// | daemon word | here | why |
/// |---|---|---|
/// | `running` | [`Running`](ContainerState::Running) | up |
/// | `restarting`, `stopping`, `paused` | [`Running`](ContainerState::Running) | still alive; the outcome is not final, and a poller must keep polling rather than read "vanished" |
/// | `created`, `configured`, `initialized` | [`Running`](ContainerState::Running) | the documented contract of [`ContainerState::Running`] is "created **and/or** running" |
/// | `stopped`, `exited`, `dead` | [`Exited`](ContainerState::Exited) | it ran and is finished; the code is the daemon's |
/// | `removing`, `unknown`, `` (empty) | [`Gone`](ContainerState::Gone) | on its way out, or the daemon will not say |
#[must_use]
pub fn container_state_from_daemon(status: &str, exit_code: i64) -> ContainerState {
match status.trim().to_ascii_lowercase().as_str() {
"running" | "restarting" | "stopping" | "paused" | "created" | "configured"
| "initialized" => ContainerState::Running,
"stopped" | "exited" | "dead" => ContainerState::Exited(exit_code),
// "removing", "unknown", "" and anything the daemon invents next.
_ => ContainerState::Gone,
}
}
/// Every container in a daemon `GET /containers/json?all=true` body whose name
/// begins with `prefix`, as `(name, state)` — the PURE half of
/// [`ContainerControl::list_containers`], so the podman-vocabulary regression above
/// is provable with no daemon.
///
/// Reads only the three fields draupnir needs (`Names`, `State`, `ExitCode`) out of
/// a permissive [`serde_json::Value`], so a daemon that grows a field, or answers
/// with a word bollard's generated model has never heard of, changes nothing here.
/// Names come back from the daemon with a leading `/`, which is stripped. The prefix
/// match is done HERE and not by the daemon's `name=` filter — see
/// [`ContainerControl::list_containers`] for why that filter has the wrong
/// semantics. Output is sorted by name for a janitor's stable log.
#[cfg(any(feature = "backend-oci", test))]
pub fn parse_container_list(body: &str, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
let v: serde_json::Value = serde_json::from_str(body)
.map_err(|e| Error::Backend(format!("listing containers: malformed daemon JSON: {e}")))?;
let arr = v.as_array().ok_or_else(|| {
Error::Backend("listing containers: daemon did not answer with a JSON array".to_string())
})?;
let mut out = Vec::new();
for c in arr {
let state = c.get("State").and_then(|s| s.as_str()).unwrap_or("");
// Docker's summary carries no exit code; podman's does. Absent ⇒ 0, which
// is the answer this engine has always given for a finished container.
let code = c
.get("ExitCode")
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
let names = c.get("Names").and_then(|n| n.as_array());
for n in names.into_iter().flatten() {
let Some(name) = n.as_str() else { continue };
let name = name.strip_prefix('/').unwrap_or(name);
if !name.starts_with(prefix) {
continue;
}
out.push((name.to_owned(), container_state_from_daemon(state, code)));
}
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
/// One container's state out of a daemon `GET /containers/{name}/json` body — the
/// PURE half of [`ContainerControl::container_state`], read the same permissive way
/// as [`parse_container_list`] and for the same reason: `State.Status` is podman's
/// vocabulary, not Docker's, and a generated closed enum turns an unfamiliar word
/// into a total inspect failure that this seam then reports as
/// [`ContainerState::Gone`] — a live container read as "no such container".
///
/// `State.Running` is authoritative when the daemon sets it; the word is what
/// separates a clean exit from a crash, and `State.ExitCode` carries the code.
#[cfg(any(feature = "backend-oci", test))]
pub fn parse_container_state(body: &str) -> Result<ContainerState> {
let v: serde_json::Value = serde_json::from_str(body)
.map_err(|e| Error::Backend(format!("inspecting container: malformed daemon JSON: {e}")))?;
let state = v.get("State");
let code = state
.and_then(|s| s.get("ExitCode"))
.and_then(serde_json::Value::as_i64)
.unwrap_or(0);
if state
.and_then(|s| s.get("Running"))
.and_then(serde_json::Value::as_bool)
== Some(true)
{
return Ok(ContainerState::Running);
}
match state.and_then(|s| s.get("Status")).and_then(|s| s.as_str()) {
Some(word) => Ok(container_state_from_daemon(word, code)),
// No word at all: it is not running, so it finished. Same answer this seam
// has always given.
None => Ok(ContainerState::Exited(code)),
}
}
// ---------------------------------------------------------------------------
// The container NAME is a handle the whole estate shares — so a name held by a
// LEFTOVER has to resolve itself, on the NEXT run.
// ---------------------------------------------------------------------------
/// **Who holds a container name right now**, read out of the daemon's own
/// `GET /containers/{name}/json` body.
///
/// The name draupnir gives a container is deterministic (`draupnir-<spec.name>`,
/// see [`ContainerBoot::container_name`]) and that is on purpose: it is the handle
/// a janitor lists by prefix ([`ContainerControl::list_containers`]), the handle
/// korp adopts its hot infrastructure by, and the handle a human types into
/// `podman logs`. A deterministic name has exactly one cost — two runs of the same
/// job can want it — and this type is what makes that cost payable instead of
/// fatal.
///
/// MEASURED on the gunnar appliance bring-up, 2026-09-20: a ladder run was killed
/// mid-build, its `draupnir-cargo-build-pid1-korp-start` container survived, and
/// from that moment EVERY ISO compose on that box died 381 seconds in with the
/// daemon's own 500 — `the container name "…" is already in use by 01f3042cab74…`
/// — wrapped four layers deep as `vm 'gunnar-installer-metered' FAILED and
/// produced nothing`. One interrupted run wedged the box permanently, and only a
/// human with a shell could unwedge it. That is the defect this seam closes: the
/// collision is not the bug, the box never recovering by itself is.
///
/// `running` is the one fact that decides what may be done about it, and it is
/// read from `State.Running` (the daemon's boolean) rather than inferred from
/// [`ContainerState`] — which deliberately collapses `created`/`configured`
/// /`initialized` onto [`ContainerState::Running`] for a POLLER that must keep
/// polling. A container the daemon created and never started is NOT alive, nobody
/// is building in it, and reading it as "alive" would refuse the very leftover
/// this seam exists to clear (`draupnir-korp-falkordb`, `Created`, was sitting on
/// oden in exactly that state while this was written).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NameHolder {
/// The holder's full container id, as the daemon spells it — what a human
/// needs to tell one leftover from another.
pub id: String,
/// `State.Running`: it is ALIVE, i.e. somebody's job may be running in it.
pub running: bool,
/// The daemon's own state word (`running`, `exited`, `created`, `stopped`,
/// podman's or Docker's), carried verbatim — never re-spelled.
pub status: String,
}
impl NameHolder {
/// The short id a human reads in `podman ps` (the first 12 characters).
#[must_use]
pub fn short_id(&self) -> &str {
&self.id[..self.id.len().min(12)]
}
}
/// The holder of a name out of a daemon `GET /containers/{name}/json` body — the
/// PURE half of [`Engine::holder_of`], read the same permissive way (and for the
/// same reason) as [`parse_container_state`]: `State.Status` is podman's
/// vocabulary, not Docker's, and a generated closed enum would turn an unfamiliar
/// word into "there is nothing there" — the one answer that must never be guessed
/// when the question is "may I destroy this?".
///
/// [`None`] when the body is not an inspect body at all (a 404 page, an error
/// document, garbage): an unreadable answer is NOT "free", and callers treat it as
/// such — see [`ensure_name_free_over`].
#[cfg(any(feature = "backend-oci", test))]
#[must_use]
pub fn parse_name_holder(body: &str) -> Option<NameHolder> {
let v: serde_json::Value = serde_json::from_str(body).ok()?;
let id = v.get("Id").and_then(|i| i.as_str()).unwrap_or("").to_string();
let state = v.get("State")?;
let running = state
.get("Running")
.and_then(serde_json::Value::as_bool)
.unwrap_or(false);
let status = state
.get("Status")
.and_then(|s| s.as_str())
.unwrap_or("")
.to_string();
Some(NameHolder {
id,
running,
status,
})
}
/// **What may be done about the name** — the whole decision, as a pure function of
/// who holds it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NameVerdict {
/// Nobody holds it; create straight away.
Free,
/// A leftover that is NOT running holds it. Remove it, say so out loud, and
/// carry on — this is the crashed-run case, and it is the whole point.
Reap(NameHolder),
/// A LIVE container holds it. Another job is building in it right now; it is
/// refused by name, and nothing is killed.
Refuse(NameHolder),
}
/// [`NameVerdict`] for a name whose holder is `holder` ([`None`] = nobody).
#[must_use]
pub fn name_verdict(holder: Option<NameHolder>) -> NameVerdict {
match holder {
None => NameVerdict::Free,
Some(h) if h.running => NameVerdict::Refuse(h),
Some(h) => NameVerdict::Reap(h),
}
}
/// **A reap that HAPPENED** — draupnir destroyed somebody's leftover, and that is
/// never allowed to be silent.
///
/// draupnir prints nothing, ever (it is a library; the consumer owns the
/// terminal), so the announcement is a value: it is recorded on the engine as the
/// reap is made and drained by the consumer through
/// [`ContainerBoot::drain_reaps`]. A build runner that says nothing when this list
/// is non-empty is the one that will one day silently destroy the container a
/// human was mid-way through inspecting.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Reap {
/// The container name that was held.
pub name: String,
/// The removed leftover's id.
pub id: String,
/// The daemon's state word for it at the moment it was removed.
pub status: String,
}
impl std::fmt::Display for Reap {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"draupnir REAPED the leftover container `{}` ({}, {}) that held the name before this run",
self.name,
&self.id[..self.id.len().min(12)],
if self.status.is_empty() {
"no state word"
} else {
&self.status
}
)
}
}
/// The named refusal for a name a LIVE container holds — the text a human is
/// actually owed instead of a daemon 500 re-wrapped as "produced nothing".
///
/// It says the name, the id, that it is running, WHOSE it is as far as anything
/// here can tell (the name is the job), and the one command that settles it.
#[must_use]
pub fn name_in_use_refusal(name: &str, holder: &NameHolder) -> String {
format!(
"REFUSED [container-name-in-use] `{name}` is held by a RUNNING container ({}, state `{}`). \
That is another job building under this name, not a leftover, and draupnir does not kill \
a live build. Wait for it, or — if you know it is abandoned — `podman rm -f {name}`",
holder.short_id(),
holder.status,
)
}
/// The named refusal for a name that is STILL taken after its leftover was reaped
/// — the podman c/storage case, where `podman rm` returns success but the storage
/// layer keeps the name (`creating container storage: the container name "…" is
/// already in use by …`). Nothing draupnir can reach over the REST API clears
/// that, so it stops here and hands over the one command that does.
#[must_use]
pub fn storage_name_in_use_refusal(name: &str, id: Option<&str>, detail: &str) -> String {
let who = match id {
Some(id) => format!(" by {}", &id[..id.len().min(12)]),
None => String::new(),
};
format!(
"REFUSED [container-name-in-use] `{name}` is STILL held{who} after its leftover was removed \
— the name is stuck in the engine's storage layer, which no REST call clears. \
Run `podman rm --storage {name}` (the daemon said: {detail})"
)
}
/// Whether a daemon error message is the name collision — matched on the daemon's
/// own words, which both podman and Docker end with `is already in use`.
#[must_use]
pub fn daemon_says_name_in_use(message: &str) -> bool {
let m = message.to_ascii_lowercase();
m.contains("already in use")
}
/// The container id a `the container name "x" is already in use by <id>` message
/// blames, when it names one.
#[must_use]
pub fn id_in_name_in_use(message: &str) -> Option<String> {
let tail = message.split("already in use by").nth(1)?;
let id: String = tail
.trim_start()
.chars()
.take_while(char::is_ascii_hexdigit)
.collect();
(id.len() >= 12).then_some(id)
}
/// The two daemon operations the name preflight needs, as a seam — so the whole
/// reap-or-refuse decision is provable with NO daemon (the same treatment
/// [`Engine::create_body`] gave the create path).
#[cfg(any(feature = "backend-oci", test))]
pub trait NameRegistry {
/// Who holds `name` right now, or [`None`] if nobody does.
fn holder_of(&self, name: &str) -> Option<NameHolder>;
/// Remove the container called `name`.
fn remove_name(&self, name: &str) -> Result<()>;
}
/// **Make the name usable, or refuse it by name** — the crash-safety preflight,
/// run at the START of every create and not at the end of any run.
///
/// A `Drop` cannot do this job: the run that leaves the leftover is the run that
/// was `pkill`ed, and a killed process runs no destructor. Recovery therefore has
/// to belong to the NEXT start, which is what this is.
///
/// - nobody holds the name ⇒ [`Ok(None)`], nothing happened;
/// - a non-running leftover holds it ⇒ it is removed and returned as a [`Reap`]
/// for the caller to announce;
/// - a RUNNING container holds it ⇒ [`Error::Backend`] carrying
/// [`name_in_use_refusal`], and it is NOT touched;
/// - the leftover cannot be removed ⇒ the daemon's own words, with the name.
///
/// # Errors
/// The name is held by a live container, or its leftover could not be removed.
#[cfg(any(feature = "backend-oci", test))]
pub fn ensure_name_free_over<R>(registry: &R, name: &str) -> Result<Option<Reap>>
where
R: NameRegistry + ?Sized,
{
match name_verdict(registry.holder_of(name)) {
NameVerdict::Free => Ok(None),
NameVerdict::Refuse(h) => Err(Error::Backend(name_in_use_refusal(name, &h))),
NameVerdict::Reap(h) => {
registry.remove_name(name).map_err(|e| {
Error::Backend(format!(
"REFUSED [container-name-in-use] `{name}` is held by the leftover {} (state `{}`) \
and draupnir could not remove it: {e}. Run `podman rm -f {name}`",
h.short_id(),
h.status,
))
})?;
Ok(Some(Reap {
name: name.to_string(),
id: h.id,
status: h.status,
}))
}
}
}
/// The terminal result of an [`ContainerControl::exec`] into a **running**
/// container: the exec'd process's exit code plus its captured stdout / stderr.
///
/// Shaped like [`RunOutcome`] (a boot-to-completion) so the two read the same, but
/// distinct: a `RunOutcome` is the *container's* lifetime, an `ExecOutcome` is one
/// command run *inside* an already-live container (a `podman exec`). `exit_code` is
/// [`Some`] with the exec'd process status (0 = clean) and [`None`] only if the
/// engine could not report one. A non-zero exit is **not** an error of the call —
/// the command ran, it just failed — so it comes back here, never as an `Err`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ExecOutcome {
/// The exec'd process's exit code (`Some(0)` = clean), or `None` if the engine
/// reported no code.
pub exit_code: Option<i64>,
/// Captured stdout lines, in order.
pub stdout: Vec<String>,
/// Captured stderr lines, in order.
pub stderr: Vec<String>,
}
/// **One process to start inside an already-running container** — the
/// [`ContainerControl::exec_streaming`] request: the argv, the environment set
/// FOR it (on top of the container's own), who it runs as and where it starts.
///
/// The piece a **long-lived toolchain container** needs: one container booted
/// once, every tool run inside it as its own exec, instead of one container
/// booted per tool (nornir's Mímir road, 2026-10-17: 66 container starts to plan
/// a 21-repo workspace). Each call carries what a fresh container's
/// [`BootSpec`] used to: env, user, working dir.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ExecSpec {
/// The command to run (non-empty).
pub argv: Vec<String>,
/// `KEY=VALUE` pairs set for this process only.
pub env: Vec<(String, String)>,
/// `uid`, `uid:gid` or a name; `None` = the container's user.
pub user: Option<String>,
/// Where it starts; `None` = the container's working dir.
pub working_dir: Option<String>,
}
impl ExecSpec {
/// Run `argv` with nothing else set.
pub fn new(argv: Vec<String>) -> Self {
Self { argv, ..Self::default() }
}
/// Set `key=value` for this process.
pub fn with_env(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
self.env.push((key.into(), value.into()));
self
}
/// Run as `user` (`uid`, `uid:gid`, a name).
pub fn with_user(mut self, user: impl Into<String>) -> Self {
self.user = Some(user.into());
self
}
/// Start in `dir`.
pub fn with_working_dir(mut self, dir: impl Into<String>) -> Self {
self.working_dir = Some(dir.into());
self
}
}
/// **What one [`ContainerControl::exec_streaming`] answered, byte for byte** —
/// the exec'd process's exit code and its two streams exactly as written (no
/// line splitting, no trimming, no empty lines dropped): the attach stream is
/// the process's own pipes, not the daemon's log record, so a line longer than
/// a log frame comes back whole.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ExecRun {
/// The process's exit code (`Some(0)` = clean; a signal reads 128 + n), or
/// `None` when the engine reported none.
pub exit_code: Option<i64>,
/// Everything it wrote to stdout.
pub stdout: Vec<u8>,
/// Everything it wrote to stderr.
pub stderr: Vec<u8>,
}
/// **Pure** builder of the `podman exec <id> <argv…>` command vector for a running
/// container — factored out so the exec wiring is testable with **no daemon** (the
/// same treatment as [`Engine::create_body`] for the create path). The returned
/// vector is the canonical command form (`["exec", id, argv0, argv1, …]`) the live
/// [`Engine::exec`] drives over bollard's `/exec` REST op (it takes `id` explicitly
/// and `argv = command[2..]`), and it is exactly what a `podman exec` subprocess
/// would run — so a consumer/test can assert the constructed command without a
/// container. `argv` is the command to run inside the container; the caller
/// ([`ContainerControl::exec`]) rejects an empty one.
pub fn exec_argv(id: &str, argv: &[&str]) -> Vec<String> {
let mut command = Vec::with_capacity(argv.len() + 2);
command.push("exec".to_string());
command.push(id.to_string());
command.extend(argv.iter().map(|a| a.to_string()));
command
}
/// **Container-specific control** beyond the generic power [`Lifecycle`]: an
/// exit-code-aware [`ContainerState`] and a streamed-log drain, an [`exec`] into a
/// running container, plus a stop that removes the container. This is the seam a job
/// handler (jera) maps onto its own boot status + log model, so the ONE bollard
/// engine lives here in draupnir and jera keeps only job policy — no second engine.
///
/// [`exec`]: ContainerControl::exec
///
/// It is a trait (not inherent methods) so a consumer can inject a mock and prove
/// its delegation wiring with no daemon; [`ContainerBoot`] is the production impl.
pub trait ContainerControl {
/// The container's exit-code-aware lifecycle state (running / exited-with-code
/// / gone). A transient inspect hiccup reads [`ContainerState::Gone`].
fn container_state(&self, machine: &Machine) -> ContainerState;
/// New streamed log lines since the last drain (the follow-task buffer), as one
/// **combined** ordered stream. Empty when the backend is not compiled in.
fn drain_logs(&self, machine: &Machine) -> Vec<String>;
/// New streamed log lines since the last drain, **split** into `(stdout,
/// stderr)`. This is the shape a run-to-completion job ([`run_to_completion`])
/// records so stdout and stderr stay apart (jera's `ContainerOutcome` keeps
/// them separate). The **default** routes every combined line to `stdout` (a
/// backend that does not distinguish the streams loses nothing observable);
/// [`ContainerBoot`] overrides it to preserve the real stdout/stderr tag. It
/// drains the same buffer as [`drain_logs`](Self::drain_logs) — call one or the
/// other per tick, not both.
fn drain_logs_split(&self, machine: &Machine) -> (Vec<String>, Vec<String>) {
(self.drain_logs(machine), Vec::new())
}
/// **The container's COMPLETE log, read in one shot**, split into `(stdout,
/// stderr)` — the daemon's own record of everything it printed, not the
/// running total a follow task has managed to deliver so far.
///
/// This exists because the streamed copy is not trustworthy at the one
/// moment a run-to-completion caller needs it: the instant the container
/// exits. The follow task is an independent tokio task parked on a socket,
/// and MEASURED on t14s 2026-09-12 over the live podman engine, 20 runs of
/// a container that printed 901 lines and exited immediately came back with
/// `exit_code = Some(0)` and **0, 0, 0 and 749 lines in four of them**. The
/// same container given a 5 s head start lost nothing in 10 runs, which
/// names the race: created, started, polled, found `Exited` and drained
/// before the follow task had finished — or in three of those four, begun —
/// reading `/logs`.
///
/// Waiting for the follow stream to close is NOT the fix, and that was
/// measured too: podman's `follow=true` response does not end when the
/// container exits. A run that waited for it sat 30 s on a container that
/// had been dead the whole time and still held only 748 of 901 lines. So
/// the answer is to stop asking the stream and ask the daemon, once, with
/// `follow=false` — a request that returns the whole log and ends.
///
/// `None` means *this backend cannot answer*, which is the honest reply from
/// a mock and from a transport error; it is NOT "the container was quiet".
/// A caller falls back to the streamed copy, which is the best it has.
///
/// Must be called while the container still EXISTS — [`stop`](Self::stop)
/// removes it, and with it the log.
fn read_all_logs(&self, machine: &Machine) -> Option<(Vec<String>, Vec<String>)> {
let _ = machine;
None
}
/// Stop + remove the container (idempotent, best-effort).
fn stop(&self, machine: &Machine);
/// **Every container whose name begins with `prefix`, running or not**, as
/// `(name, state)`.
///
/// The call a *janitor* needs, and the one whose absence made consumers
/// reach past this seam. gunnar's bench suite carried
///
/// ```text
/// // draupnir has no list-containers call, so this is `podman` directly
/// Command::new("podman").args(["ps", "-a", "--filter", "name=…", "--format", "{{.Names}}"])
/// ```
///
/// to sweep containers a dead run had left behind. That comment was correct
/// and is the reason this method exists: a boot path with no shell fallback
/// is worth little if the sweep beside it shells out anyway, because the
/// shell is where the `podman`-not-on-PATH and the wrong-engine failures come
/// back in — silently, since a failed `podman ps` and a genuinely empty box
/// produce the same empty stdout.
///
/// Exited and created containers are included: a janitor that saw only
/// running ones would leave exactly the corpses it exists to remove.
///
/// The **default** answers `Unsupported`, so a mock backend says so rather
/// than reporting an empty box — "nothing to sweep" and "I cannot look" must
/// not be the same answer to a caller about to decide nothing needs removing.
fn list_containers(&self, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
let _ = prefix;
Err(Error::Unsupported(
"this backend cannot list containers (needs the OCI engine)".into(),
))
}
/// **The environment the container was created with**, as `KEY=VALUE` lines,
/// straight out of the daemon's own record of it.
///
/// Not "what we passed" — what the daemon holds. That difference is the
/// whole value: a variable dropped between a [`BootSpec`] and the running
/// process is invisible to any check that re-reads the spec, and settings
/// like `GOMAXPROCS` or `TOKIO_WORKER_THREADS` have no observable effect a
/// harness can query from outside, so this is the only read-back they have.
///
/// The second call gunnar's bench shelled out for
/// (`podman inspect --format '{{range .Config.Env}}…'`).
///
/// [`BootSpec`]: crate::BootSpec
fn container_env(&self, machine: &Machine) -> Result<Vec<String>> {
let _ = machine;
Err(Error::Unsupported(
"this backend cannot inspect a container's environment (needs the OCI engine)".into(),
))
}
/// **Exec `argv` inside the already-running container `machine`** — the
/// `podman exec <id> <argv…>` analogue over the ONE OCI engine, returning the
/// exec'd process's [`ExecOutcome`] (exit code + captured stdout/stderr). This
/// is the control primitive a **live start/stop/exec lifecycle** needs: after a
/// detached [`Boot::boot`] hands back a long-lived [`Machine`], a surviving
/// handle can run commands *into* it (health probe, live reconfigure, drain)
/// without tearing it down — the piece [`run_to_completion`] (which owns the
/// whole container lifetime) cannot express.
///
/// **Container-only** (guarded parity with the container-only `net`/`cmd`/`ports`
/// `validate()` checks): a non-container [`Machine`] (a KVM guest / Redfish node)
/// has nothing to exec into and is rejected with an [`Error::Spec`] *before* the
/// engine is touched, and an **empty `argv`** (nothing to run) is likewise
/// rejected. This provided method does the guards + assembles the command via
/// [`exec_argv`] and then delegates the live run to
/// [`exec_command`](Self::exec_command); a backend without an OCI engine (a mock)
/// keeps the default `exec_command` and so this whole path is exercised with no
/// daemon. A non-zero exit of the exec'd process is **not** an `Err` — it is the
/// [`ExecOutcome::exit_code`]; only a guard failure or an engine/connect error
/// returns `Err`.
fn exec(&self, machine: &Machine, argv: &[&str]) -> Result<ExecOutcome> {
if machine.kind != crate::Kind::Container {
return Err(Error::Spec(format!(
"exec is container-only; a {:?} machine has no container to exec into",
machine.kind
)));
}
if argv.is_empty() {
return Err(Error::Spec(
"a container exec needs a non-empty argv (nothing to run)".into(),
));
}
let command = exec_argv(&machine.id, argv);
self.exec_command(&command)
}
/// The backend hook [`exec`](Self::exec) delegates the *live run* to, once the
/// container-only + non-empty-argv guards have passed and the canonical command
/// (`["exec", id, argv…]`, from [`exec_argv`]) is assembled. **Default**:
/// [`Error::Unsupported`] — an engine-less backend (a mock in a unit test)
/// cannot exec, but the guard + command-assembly path in [`exec`](Self::exec)
/// is still fully exercised against it. [`ContainerBoot`] overrides it to drive
/// bollard's `/exec` REST op (`id = command[1]`, `argv = command[2..]`). A
/// consumer normally calls [`exec`](Self::exec), not this.
fn exec_command(&self, command: &[String]) -> Result<ExecOutcome> {
let _ = command;
Err(Error::Unsupported(
"this backend cannot exec into a container (needs the OCI engine)".into(),
))
}
/// **Run one process inside the running container `machine`, streaming its
/// output as it comes** — [`exec`](Self::exec) with an environment, a user
/// and a working dir ([`ExecSpec`]), and with the two streams handed to
/// `sink` as raw byte chunks while it runs and returned whole
/// ([`ExecRun`]). Several may run at once in one container, from several
/// threads: each is its own exec.
///
/// The same guards as [`exec`](Self::exec) (container-only, non-empty argv)
/// run before the engine is touched; the live run is
/// [`exec_streaming_on`](Self::exec_streaming_on). A non-zero exit is the
/// [`ExecRun::exit_code`], never an `Err`; only a guard, create, start,
/// transport or inspect failure is.
fn exec_streaming(
&self,
machine: &Machine,
spec: &ExecSpec,
sink: &mut dyn FnMut(LogStream, &[u8]),
) -> Result<ExecRun> {
if machine.kind != crate::Kind::Container {
return Err(Error::Spec(format!(
"exec is container-only; a {:?} machine has no container to exec into",
machine.kind
)));
}
if spec.argv.is_empty() {
return Err(Error::Spec(
"a container exec needs a non-empty argv (nothing to run)".into(),
));
}
self.exec_streaming_on(&machine.id, spec, sink)
}
/// The backend hook behind [`exec_streaming`](Self::exec_streaming), given
/// the daemon-side container name. **Default**: [`Error::Unsupported`].
fn exec_streaming_on(
&self,
id: &str,
spec: &ExecSpec,
sink: &mut dyn FnMut(LogStream, &[u8]),
) -> Result<ExecRun> {
let _ = (id, spec, sink);
Err(Error::Unsupported(
"this backend cannot exec into a container (needs the OCI engine)".into(),
))
}
}
impl ContainerControl for ContainerBoot {
fn container_state(&self, machine: &Machine) -> ContainerState {
#[cfg(feature = "backend-oci")]
{
match self.engine() {
Ok(e) => e.container_state(&machine.id),
Err(_) => ContainerState::Gone,
}
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
ContainerState::Gone
}
}
fn drain_logs(&self, machine: &Machine) -> Vec<String> {
#[cfg(feature = "backend-oci")]
{
match self.engine() {
Ok(e) => e.drain_logs(&machine.id),
Err(_) => Vec::new(),
}
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
Vec::new()
}
}
fn drain_logs_split(&self, machine: &Machine) -> (Vec<String>, Vec<String>) {
#[cfg(feature = "backend-oci")]
{
match self.engine() {
Ok(e) => e.drain_logs_split(&machine.id),
Err(_) => (Vec::new(), Vec::new()),
}
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
(Vec::new(), Vec::new())
}
}
fn read_all_logs(&self, machine: &Machine) -> Option<(Vec<String>, Vec<String>)> {
#[cfg(feature = "backend-oci")]
{
self.engine().ok().and_then(|e| e.read_all_logs(&machine.id))
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
None
}
}
fn stop(&self, machine: &Machine) {
#[cfg(feature = "backend-oci")]
{
if let Ok(e) = self.engine() {
e.stop(&machine.id);
}
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
}
}
fn list_containers(&self, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
#[cfg(feature = "backend-oci")]
{
self.engine()?.list_containers(prefix)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = prefix;
Err(Error::Unsupported(
"this build has no OCI engine, so it cannot list containers".into(),
))
}
}
fn container_env(&self, machine: &Machine) -> Result<Vec<String>> {
#[cfg(feature = "backend-oci")]
{
self.engine()?.container_env(&machine.id)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = machine;
Err(Error::Unsupported(
"this build has no OCI engine, so it cannot inspect a container".into(),
))
}
}
fn exec_command(&self, command: &[String]) -> Result<ExecOutcome> {
// The guards + assembly ran in the provided `exec`; `command` is the
// canonical `["exec", <id>, <argv…>]` (non-empty argv => len >= 3). Drive
// bollard's `/exec` REST op against `id` with `argv = command[2..]`.
#[cfg(feature = "backend-oci")]
{
let id = &command[1];
let argv: Vec<String> = command[2..].to_vec();
self.engine()?.exec(id, &argv)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = command;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
fn exec_streaming_on(
&self,
id: &str,
spec: &ExecSpec,
sink: &mut dyn FnMut(LogStream, &[u8]),
) -> Result<ExecRun> {
#[cfg(feature = "backend-oci")]
{
self.engine()?.exec_streaming(id, spec, sink)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (id, spec, sink);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
}
// ---------------------------------------------------------------------------
// run_to_completion — the single-call run-to-completion container seam.
// ---------------------------------------------------------------------------
/// The terminal result of a [`run_to_completion`] job: the container's exit code
/// plus its captured logs, split into stdout / stderr.
///
/// `exit_code` is [`Some`] with the process exit status (0 = clean) when the
/// container exited on its own, and [`None`] when it vanished before a code could be
/// read (removed out from under us, killed by signal with no reported status). This
/// mirrors jera's `ContainerOutcome` field-for-field, so `nornir::jobs::run_container`
/// can repoint onto this seam and delete its duplicate `BollardEngine`.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RunOutcome {
/// The container's exit code (`Some(0)` = clean), or `None` if it went away
/// before a code was observed.
pub exit_code: Option<i64>,
/// Captured stdout lines, in order.
pub stdout: Vec<String>,
/// Captured stderr lines, in order.
pub stderr: Vec<String>,
}
/// Knobs for [`run_to_completion`]: how long to wait for the container to exit and
/// how often to poll its state. [`Default`] waits **indefinitely** (parity with
/// jera's blocking `wait_container`) and polls every 200 ms.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RunOptions {
/// Overall budget before giving up. `None` = wait forever for the container to
/// exit (jera parity). `Some(d)` stops + removes the container and returns an
/// [`Error::Backend`] if it has not exited within `d`.
pub timeout: Option<Duration>,
/// How often the container state is polled (and logs drained) while it runs.
pub poll_interval: Duration,
}
impl Default for RunOptions {
fn default() -> Self {
Self {
timeout: None,
poll_interval: Duration::from_millis(200),
}
}
}
impl RunOptions {
/// Wait indefinitely (jera parity), polling on `poll_interval`.
pub fn poll_every(poll_interval: Duration) -> Self {
Self {
timeout: None,
poll_interval,
}
}
/// Cap the run at `timeout`, polling on `poll_interval`.
pub fn bounded(timeout: Duration, poll_interval: Duration) -> Self {
Self {
timeout: Some(timeout),
poll_interval,
}
}
}
/// **Run a container to completion in one call** — start it, wait for it to exit,
/// collect its logs, and return an exit-code-aware [`RunOutcome`]. This is the
/// missing seam that lets jera's run-to-completion `run_container` (and, above it,
/// `nornir::jobs::run_container`) route through draupnir's **one** OCI engine
/// instead of jera's duplicate `BollardEngine` — the run-to-completion analogue of
/// how a VM/container *boot* already delegates through [`Boot`].
///
/// It is written against the always-compiled [`Boot`] + [`ContainerControl`] seam,
/// generic over the backend, so it drives a live [`ContainerBoot`] in production
/// **and** a mock in a unit test with no daemon. The flow is exactly jera's
/// run_container: [`boot`](Boot::boot) (create + start) → poll
/// [`container_state`](ContainerControl::container_state) draining
/// [`drain_logs_split`](ContainerControl::drain_logs_split) each tick until the
/// container reports [`ContainerState::Exited`] (or [`ContainerState::Gone`]) → a
/// final drain → [`stop`](ContainerControl::stop) (remove). A non-zero exit is NOT
/// an `Err` — it comes back in [`RunOutcome::exit_code`] (the *container* failed,
/// the *call* succeeded); only a boot/connect failure or a `timeout` returns `Err`.
///
/// Note: on a live engine the log follow-task flushes asynchronously, so the final
/// drain after exit is a running total and NOT the whole output. Once the container
/// has exited the daemon is asked for its complete log in one shot
/// ([`ContainerControl::read_all_logs`]) and that is what comes back; the streamed
/// copy is the fallback for a backend that cannot answer. It used to be the answer,
/// and 4 of 20 measured runs of a chatty container returned exit-0 with a truncated
/// or empty stdout because of it.
pub fn run_to_completion<B>(backend: &B, spec: &BootSpec, opts: &RunOptions) -> Result<RunOutcome>
where
B: Boot + ContainerControl,
{
let machine = backend.boot(spec)?;
drive_to_completion(backend, &machine, opts)
}
/// [`run_to_completion`] with a **live line sink** — boot `spec`, then drive it
/// through [`drive_to_completion_streaming`] so every log line reaches `sink`
/// while the container is still running. See that function for the ordering and
/// exactly-once guarantees the sink gets.
pub fn run_to_completion_streaming<B>(
backend: &B,
spec: &BootSpec,
opts: &RunOptions,
sink: &mut dyn FnMut(LogStream, &str),
) -> Result<RunOutcome>
where
B: Boot + ContainerControl,
{
let machine = backend.boot(spec)?;
drive_to_completion_streaming(backend, &machine, opts, sink)
}
/// The post-boot run-to-completion loop — drain logs → poll state → final drain →
/// stop — factored out of [`run_to_completion`] so it AND the bind-mount variant
/// ([`ContainerBoot::run_to_completion_with_binds`]) share one implementation
/// (single-source: the drive loop lives once). Takes an already-booted `machine`.
pub fn drive_to_completion<B>(
backend: &B,
machine: &Machine,
opts: &RunOptions,
) -> Result<RunOutcome>
where
B: ContainerControl,
{
// One implementation, with a sink that discards: the non-streaming call is
// the streaming one with nobody listening, so the two can never drift.
drive_to_completion_streaming(backend, machine, opts, &mut |_, _| {})
}
/// **[`run_to_completion`] with a LIVE line sink** — `sink` is called with every
/// log line *as it is drained*, while the container is still running, instead of
/// only at the end.
///
/// # Why this exists
///
/// [`RunOutcome`] is a verdict, and a verdict arrives once. That is right for a
/// container that runs for a second and wrong for one that runs for eight
/// minutes: a consumer that can show nothing until the exit code lands shows a
/// blank screen for the whole build, and a build with no visible progress is a
/// build an operator kills. Njord's appliance build is exactly that shape — three
/// container steps, a console polling a log file — and before this seam existed
/// the only way to keep its live tail was to keep its hand-rolled
/// `Command::new("podman")` runner, i.e. a second container runner in the
/// constellation, which is the thing this crate exists to prevent.
///
/// # What the sink is guaranteed
///
/// Every line reaches the sink **exactly once, in order, per stream**, and the
/// sink's lines add up to [`RunOutcome::stdout`]/[`stderr`](RunOutcome::stderr).
/// That takes one extra step, because the two sources disagree by design: the
/// streamed drain is a running total delivered by an asynchronous follow task,
/// and the authoritative answer is the daemon's complete record read *after* the
/// exit ([`ContainerControl::read_all_logs`] — see [`drive_to_completion`] for
/// the measurement that put it there). The streamed copy is a **prefix** of that
/// record, so once the record arrives only its *tail* — the lines beyond what was
/// already streamed — is handed to the sink. A backend that cannot answer
/// `read_all_logs` (a mock, a transport error) streams and returns the same
/// lines, exactly as before.
///
/// A short authoritative record (fewer lines than were streamed) adds nothing:
/// the sink is never asked to un-say a line it already emitted.
///
/// [`stderr`]: RunOutcome::stderr
pub fn drive_to_completion_streaming<B>(
backend: &B,
machine: &Machine,
opts: &RunOptions,
sink: &mut dyn FnMut(LogStream, &str),
) -> Result<RunOutcome>
where
B: ContainerControl,
{
let mut stdout: Vec<String> = Vec::new();
let mut stderr: Vec<String> = Vec::new();
let deadline = opts.timeout.map(|t| Instant::now() + t);
// Accumulate AND announce in one place, so a line can never be recorded
// without reaching the sink (or reach it twice) — the invariant the streaming
// contract above rests on.
fn take(
dst: &mut Vec<String>,
src: Vec<String>,
stream: LogStream,
sink: &mut dyn FnMut(LogStream, &str),
) {
for line in src {
sink(stream, &line);
dst.push(line);
}
}
let exit_code = loop {
// Drain incrementally so a long-running, chatty container doesn't buffer
// unboundedly before we ever read it — and, with a sink attached, this is
// also the tick that makes an eight-minute build visible while it runs.
let (out, err) = backend.drain_logs_split(machine);
take(&mut stdout, out, LogStream::Stdout, sink);
take(&mut stderr, err, LogStream::Stderr, sink);
match backend.container_state(machine) {
ContainerState::Exited(code) => break Some(code),
ContainerState::Gone => break None,
ContainerState::Running => {}
}
if let Some(dl) = deadline {
if Instant::now() >= dl {
backend.stop(machine);
return Err(Error::Backend(format!(
"container `{}` did not run to completion within {:?}",
machine.id,
opts.timeout.unwrap()
)));
}
}
std::thread::sleep(opts.poll_interval);
};
// Final drain: catch the lines emitted between the last poll and exit.
let (out, err) = backend.drain_logs_split(machine);
take(&mut stdout, out, LogStream::Stdout, sink);
take(&mut stderr, err, LogStream::Stderr, sink);
// ── and then ASK, rather than hope ──────────────────────────────────────
//
// The drains above are a running total delivered by an asynchronous follow
// task, and at this instant — the container has just exited — that total is
// routinely short. MEASURED on t14s 2026-09-12: 4 of 20 runs of a container
// that printed 901 lines and exited came back with exit 0 and 0, 0, 0 and
// 749 lines. The consumer is nornir-build-thing's cargo-in-a-container road,
// which reads cargo's `--message-format=json` off this stream; the one line
// naming the produced executable is the second-to-last of ~900, so a short
// stream is a build that reports success and no binary. That is what took
// down the gunnar appliance compose that day.
//
// So take the daemon's own complete record while the container still exists
// ([`ContainerControl::read_all_logs`], `follow=false`, one request that
// ends). `None` is "this backend cannot answer" — a mock, or a transport
// error — and then the streamed copy is the best there is, which is what
// every caller had before.
if let Some((all_out, all_err)) = backend.read_all_logs(machine) {
// The streamed copy is a PREFIX of this record, so the sink is owed only
// what it has not seen. `skip` past the streamed count is that tail; a
// record SHORTER than the stream skips everything and says nothing, which
// is the only honest move — a line already handed over cannot be recalled.
for line in all_out.iter().skip(stdout.len()) {
sink(LogStream::Stdout, line);
}
for line in all_err.iter().skip(stderr.len()) {
sink(LogStream::Stderr, line);
}
stdout = all_out;
stderr = all_err;
}
// Remove the container (idempotent) now that we have its code + logs.
backend.stop(machine);
Ok(RunOutcome {
exit_code,
stdout,
stderr,
})
}
impl ContainerBoot {
/// Run `spec` to completion on the live OCI engine — the ergonomic production
/// entry point that forwards to the generic [`run_to_completion`] with `self`.
/// Requires the `backend-oci` feature (else an honest [`Error::Unsupported`]).
pub fn run_to_completion(&self, spec: &BootSpec, opts: &RunOptions) -> Result<RunOutcome> {
#[cfg(feature = "backend-oci")]
{
run_to_completion(self, spec, opts)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (spec, opts);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// [`run_to_completion`](Self::run_to_completion) with a **live line sink** —
/// the production entry point for a long job whose output has to be visible
/// *while it runs* (a build log tailed by a console), not only once it ends.
/// Forwards to the generic [`run_to_completion_streaming`] with `self`;
/// requires `backend-oci` (else an honest [`Error::Unsupported`]).
pub fn run_to_completion_streaming(
&self,
spec: &BootSpec,
opts: &RunOptions,
sink: &mut dyn FnMut(LogStream, &str),
) -> Result<RunOutcome> {
#[cfg(feature = "backend-oci")]
{
run_to_completion_streaming(self, spec, opts, sink)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (spec, opts, sink);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// **The daemon-side name a spec called `spec_name` boots as** — the `id` of
/// the [`Machine`] it returns, and the prefix
/// [`ContainerControl::list_containers`] matches on.
///
/// Public because a *janitor* needs it before there is a `Machine` to ask:
/// a caller sweeping up after a run that panicked, or that died between two
/// steps, has only the names it meant to boot. Hard-coding `"draupnir-"` at
/// the call site would be a second copy of this crate's naming rule, and the
/// one that goes stale.
pub fn container_name_for(spec_name: &str) -> String {
format!("draupnir-{spec_name}")
}
/// Boot `spec` with additional host **bind mounts** (`host:container[:opts]`,
/// `-v` semantics) on top of the spec's own [`BootSpec::binds`]. `binds` empty ⇒
/// identical to [`Boot::boot`].
///
/// This is no longer a second boot body: it appends to the spec's binds and boots
/// through the ONE path, so a mount can never again reach the engine on this leg
/// and be dropped on the other (the korp FalkorDB volume, oden 2026-09-10).
pub fn boot_with_binds(&self, spec: &BootSpec, binds: &[String]) -> Result<Machine> {
if binds.is_empty() {
return self.boot(spec);
}
let merged = with_extra_binds(spec, binds);
self.boot(&merged)
}
/// [`run_to_completion`](Self::run_to_completion) with host **bind mounts** —
/// lets a build-in-a-container job (WiX/MSI under Wine, `pack`) mount its input/
/// output dirs and run through this ONE OCI engine, replacing a
/// `Command::new("podman") -v …` shell twin. `binds` empty ⇒ identical to
/// `run_to_completion`. Requires `backend-oci`.
pub fn run_to_completion_with_binds(
&self,
spec: &BootSpec,
opts: &RunOptions,
binds: &[String],
) -> Result<RunOutcome> {
#[cfg(feature = "backend-oci")]
{
let machine = self.boot_with_binds(spec, binds)?;
drive_to_completion(self, &machine, opts)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (spec, opts, binds);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// **Build an OCI image** from a build context directory + a Containerfile
/// (`podman build` over the one engine). `context_dir` is tarred in-memory and
/// sent to the daemon's `/build`; `containerfile` is its path relative to the
/// context; `tag` names the result. Returns `tag` on success. Kills the
/// `Command::new("podman") build …` shell twin (Skidbladnir `pack.rs`). Requires
/// `backend-oci`; honest [`Error::Backend`] when the socket is unreachable, never
/// a fake image.
pub fn build_image(
&self,
context_dir: &Path,
containerfile: &str,
tag: &str,
) -> Result<String> {
self.build_image_at(context_dir, containerfile, tag, None)
}
/// **Build an OCI image REPRODUCIBLY** — [`build_image`](Self::build_image)
/// with podman's `timestamp` build parameter set to `source_date_epoch`, so
/// the image config's `created` field and every file timestamp in the
/// committed layers are that instant rather than "now".
///
/// # Why this is a separate verb
///
/// An image id is the digest of the image CONFIG, and the config carries a
/// `created` timestamp plus one per history entry. No Dockerfile
/// instruction can reach any of them — they are written by the builder at
/// commit time. So a recipe can be pinned down to the byte and its image id
/// will still move every time it is built, which is exactly what was
/// measured on oden 2026-08-30: six mints of one Containerfile into six
/// fresh empty stores, six different ids. This parameter is the missing
/// half, and with it four mints of the same recipe over fifteen minutes
/// produced one id.
///
/// It also pins the manifest **format** to OCI, and that is not a detail:
/// with the timestamp alone, this door and the `podman build` CLI door
/// produced identical layer diff ids and two DIFFERENT image ids
/// (`3f632569f198…` vs `e255092edde0…`, measured 2026-08-30). An image id is
/// the digest of the image config, the config's media type is part of it,
/// `podman build` defaults to OCI and this endpoint defaults to Docker
/// schema2 — so "reproducible" had to include the format or the two ways
/// this constellation mints an image would never agree.
///
/// ⚠ **podman only, and loudly so.** Both parameters are podman extensions,
/// and `outputformat` is honoured only on podman's *libpod* route — the
/// Docker-compat route accepts it and ignores it (measured) — so a
/// reproducible build is re-pointed at `/vX/libpod/build`. Against a real
/// Docker daemon that path is a 404 and the build fails BY NAME, which is
/// the intended shape: a silently-ignored reproducibility flag is exactly
/// the failure this exists to refuse. The caller should still close the
/// loop through a different door — read the built image back and assert its
/// `created` is the instant that was asked for. `nornir-build-thing`'s
/// `tests/toolchain_images_live.rs` does that, and compares the two doors'
/// ids.
pub fn build_image_reproducible(
&self,
context_dir: &Path,
containerfile: &str,
tag: &str,
source_date_epoch: i64,
) -> Result<String> {
self.build_image_at(context_dir, containerfile, tag, Some(source_date_epoch))
}
/// **The daemon's ID for `image`**, or `None` when it cannot say.
///
/// The missing half of [`build_image_reproducible`]: that returns the TAG,
/// and a tag is a name somebody can move. A caller that has just minted an
/// image and wants to PIN what it minted needs the id, and until now the
/// only ways to get one were to export the image to a tar and scan the
/// archive, or to shell out to `podman inspect` — around the single engine
/// this crate exists to be.
///
/// Measured, 2026-09-05, which is why this exists: proving that a
/// Containerfile with a fixed `SOURCE_DATE_EPOCH` mints reproducibly meant
/// minting twice and comparing ids, and there was no door here that could
/// answer. The comparison was done with the CLI instead, through a
/// DIFFERENT build path, and the two ids differed for that reason rather
/// than for the reason under test — a whole wrong conclusion published and
/// retracted off one missing accessor.
///
/// `Ok(None)` is an honest answer, not a failure: a mock engine has no
/// daemon to ask. Reaching the daemon at all still needs `backend-oci`.
pub fn image_id(&self, image: &str) -> Result<Option<String>> {
#[cfg(feature = "backend-oci")]
{
Ok(self.engine()?.image_id(image))
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = image;
Err(Error::Unsupported(
"asking the daemon for an image id needs the `backend-oci` feature".into(),
))
}
}
/// Build reproducibly and hand back **both** the tag and the id the daemon
/// gave it — the shape a caller pinning its own mint wants, so it does not
/// have to remember to ask afterwards.
///
/// The id is `None` when the engine cannot say; the build still succeeded,
/// and a caller that REQUIRES a pin should treat `None` as a refusal rather
/// than pin the tag and call it evidence.
pub fn build_image_reproducible_pinned(
&self,
context_dir: &Path,
containerfile: &str,
tag: &str,
source_date_epoch: i64,
) -> Result<(String, Option<String>)> {
let t = self.build_image_reproducible(context_dir, containerfile, tag, source_date_epoch)?;
let id = self.image_id(&t)?;
Ok((t, id))
}
fn build_image_at(
&self,
context_dir: &Path,
containerfile: &str,
tag: &str,
source_date_epoch: Option<i64>,
) -> Result<String> {
#[cfg(feature = "backend-oci")]
{
self.engine()?
.build_image(context_dir, containerfile, tag, source_date_epoch)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (context_dir, containerfile, tag, source_date_epoch);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// **Is an OCI image already present locally?** — the `podman image exists <ref>`
/// / `docker image inspect <ref>` shell twin over the one engine. `Ok(false)` is
/// reserved for the daemon answering *"no such image"*; any other failure
/// (socket unreachable, permission denied) is an honest [`Error::Backend`], so a
/// caller can never mistake a broken engine for a missing image and rebuild the
/// world. The 404→`false` vs error→`Err` decision lives in [`image_present_on`]
/// (daemon-free, mock-tested). Requires `backend-oci`.
pub fn image_present(&self, image: &str) -> Result<bool> {
#[cfg(feature = "backend-oci")]
{
image_present_on(self.engine()?.as_ref(), image)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = image;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// **Extract a path from an image** to the host — the `podman create` + `podman
/// cp <container>:<path> <host>` twin over the one engine. Creates a throwaway
/// (un-started) container from `image`, downloads a tar of `container_path` via
/// the daemon, and unpacks it into `host_dest`, then removes the container. The
/// unpacked tree is rooted at the basename of `container_path` (same layout
/// `podman cp` yields). Requires `backend-oci`; honest [`Error::Backend`] when the
/// socket is unreachable, never a partial fake.
pub fn extract_path(&self, image: &str, container_path: &str, host_dest: &Path) -> Result<()> {
#[cfg(feature = "backend-oci")]
{
self.engine()?
.extract_path(image, container_path, host_dest)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (image, container_path, host_dest);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// **Export a local image to a tar archive on disk** — the `podman save -o` twin
/// over the one engine, and the missing half of the airgap story: without it a
/// bundle has no image ARTIFACT to carry, so an offline target is left running
/// [`build_image`](Self::build_image) at first use, i.e. reaching for a network
/// on a box that has none.
///
/// **Streams.** Bytes go daemon → `BufWriter` → file a chunk at a time; the
/// gigabyte is never resident. (`nordisk-spark-iceberg:4.1.2` measures 1.39 GB
/// on oden, which is also why these images ride the bundle and not a crate.)
///
/// **Three failures stay three failures**, which is the whole point of routing
/// this through [`ImagePresence`] rather than a shell exit code:
/// - the image is not on this host → *"no such image locally"*, and **no file is
/// created** (a zero-byte tar that later "verifies" as present is the exact
/// silent-corruption path this avoids);
/// - the socket is unreachable/denied → *"container engine unreachable"*;
/// - the stream died mid-archive → [`ArchiveFault::Truncated`], caught by
/// re-reading the written file's tar headers before returning success.
///
/// On success the returned [`ImageArchive`] carries the tags and config digest
/// **read back out of the archive**, not the ones the caller asked for — so a
/// manifest records what the tar actually holds. Requires `backend-oci`.
pub fn export_image(&self, image: &str, dest: &Path) -> Result<ImageArchive> {
#[cfg(feature = "backend-oci")]
{
export_image_on(self.engine()?.as_ref(), image, dest)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = (image, dest);
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
/// **Import an image archive back into the local store** — the `podman load -i`
/// twin over the one engine, and what an airgapped target runs on unfold so the
/// service that needs the image finds it already there.
///
/// The archive is **verified before a single byte is streamed at the daemon**
/// ([`inspect_image_archive`]): a truncated or non-image tar fails here, cheaply
/// and by name, instead of half-loading and leaving the store in a state nobody
/// can describe. Only then is the file streamed to `/images/load` in chunks —
/// again never resident in memory.
///
/// Returns the [`ImageArchive`] describing what was loaded, so a caller can log
/// the tags and digest it actually installed rather than the ones it hoped for.
/// Requires `backend-oci`.
pub fn import_image(&self, src: &Path) -> Result<ImageArchive> {
#[cfg(feature = "backend-oci")]
{
import_image_on(self.engine()?.as_ref(), src)
}
#[cfg(not(feature = "backend-oci"))]
{
let _ = src;
Err(Error::Unsupported(
"container backend needs the `backend-oci` feature (drives the podman/Docker REST API via bollard)"
.into(),
))
}
}
}
/// **Pure** in-memory tar of an OCI build-context directory — factored out of
/// [`Engine::build_image`] so the context packing is testable with no daemon
/// (build a temp dir → tar → read back the entries). Every file under `context_dir`
/// is added at its relative path (archive root `.`), the shape the daemon's `/build`
/// endpoint expects. Feature-gated because it rides the `tar` crate that comes with
/// `backend-oci`.
#[cfg(feature = "backend-oci")]
fn context_tar(context_dir: &Path) -> Result<Vec<u8>> {
let mut buf: Vec<u8> = Vec::new();
{
let mut builder = tar::Builder::new(&mut buf);
builder.append_dir_all(".", context_dir).map_err(|e| {
Error::Backend(format!("tar build context {}: {e}", context_dir.display()))
})?;
builder
.finish()
.map_err(|e| Error::Backend(format!("finish build-context tar: {e}")))?;
}
Ok(buf)
}
/// The **daemon-touching primitives** of the image *build* and *extract* paths,
/// pulled behind a seam so the surrounding logic — a `BuildInfo.error_detail`
/// failing the build, the returned tag on a clean stream, and the basename-rooted
/// tar unpack — is drivable by a mock with **no socket** (the same treatment
/// [`create_body`](Engine::create_body) / [`context_tar`] already got). The live
/// [`Engine`] implements it over bollard; a unit-test mock scripts the daemon's
/// replies to prove [`build_image_on`] / [`extract_path_on`] with no daemon.
#[cfg(feature = "backend-oci")]
trait ImageDaemon {
/// Stream a build of the tarred `context` (dockerfile `containerfile`, tagged
/// `tag`), drained to **one item per `BuildInfo`**: `Ok(None)` = a progress
/// line, `Ok(Some(msg))` = a daemon-reported build error (an `error_detail`,
/// its `message` — empty string if the detail carried none), and `Err(..)` = a
/// transport failure. Draining to a `Vec` keeps the async stream inside the
/// engine while the pass/fail decision ([`build_image_on`]) stays daemon-free.
///
/// `source_date_epoch` is podman's `timestamp` build parameter: when `Some`,
/// the image config's `created` field AND every file timestamp in the
/// committed layers are set to that instant instead of "now". It is the half
/// of a reproducible image build that no Containerfile can reach — see
/// [`ContainerBoot::build_image_reproducible`].
fn build_stream(
&self,
context: Vec<u8>,
containerfile: &str,
tag: &str,
source_date_epoch: Option<i64>,
) -> Vec<Result<Option<String>>>;
/// Create a throwaway (un-started) container from `image`, download a tar of
/// `container_path` from the daemon, and remove the container (best-effort, even
/// on a download error). Returns the raw tar bytes — rooted at the basename of
/// `container_path`, the `podman cp` layout [`extract_path_on`] then unpacks.
fn download_path_tar(&self, image: &str, container_path: &str) -> Result<Vec<u8>>;
/// Ask the daemon whether `image` resolves locally, flattened to the three
/// answers [`image_present_on`] decides on. The engine maps a bollard
/// `inspect_image` reply here; a mock scripts it with no socket.
fn inspect_image(&self, image: &str) -> ImagePresence;
/// **The daemon's own ID for `image`**, or `None` when this engine cannot
/// say (a mock, or a daemon that answered without one).
///
/// [`inspect_image`](Self::inspect_image) throws this away — it flattens the
/// whole reply to `Present`/`Absent`, and the reply is where the id lives.
/// That cost real work on 2026-09-05: a caller that had just minted an image
/// could not name what it minted without exporting it to a tar and scanning
/// the archive, so `build_image_reproducible` handed back the TAG and the
/// only way to learn the id was to shell out to `podman inspect` — around
/// the one engine this crate exists to be.
///
/// Defaulted rather than required so every existing implementor (the mocks)
/// keeps compiling and answers the honest `None`.
fn image_id(&self, image: &str) -> Option<String> {
let _ = image;
None
}
/// Stream `image`'s archive from the daemon into `sink`, returning the byte
/// count written. Takes a `&mut dyn Write` rather than returning a `Vec<u8>`
/// **specifically** so the gigabyte never becomes a value: the caller owns a
/// `BufWriter<File>` and each chunk goes straight through it. The surrounding
/// decisions (presence probe, dir creation, read-back verification) live in
/// [`export_image_on`], daemon-free.
fn export_to(&self, image: &str, sink: &mut dyn std::io::Write) -> Result<u64>;
/// Stream the archive at `src` into the daemon's image store. Takes a path, not
/// bytes, for the same reason: the engine reads it in chunks. Verification of
/// what is at that path already happened in [`import_image_on`] before this is
/// called, so an implementation may assume a well-formed archive and report only
/// what the daemon says about loading it.
fn load_archive(&self, src: &Path) -> Result<()>;
}
/// The daemon's answer to "do you have this image?", split so the **missing** case
/// is distinguishable from a **broken** one. `podman image exists` collapses both
/// into a non-zero exit — that conflation is exactly what this seam exists to kill:
/// a caller that reads an unreachable socket as "absent" self-heals by rebuilding an
/// image it already has, forever.
#[cfg(feature = "backend-oci")]
#[derive(Debug, Clone, PartialEq, Eq)]
enum ImagePresence {
/// The daemon resolved the reference.
Present,
/// The daemon answered, and has no such image (a 404).
Absent,
/// The daemon could not be asked (socket down, denied, transport) — carries the
/// message that becomes the [`Error::Backend`].
Failed(String),
}
/// **Decide image presence over the [`ImageDaemon`] seam.** `Present` → `Ok(true)`,
/// `Absent` → `Ok(false)`, `Failed` → [`Error::Backend`]. Pure but for the injected
/// daemon, so a mock proves all three arms — in particular that a transport failure
/// never reads as "absent".
#[cfg(feature = "backend-oci")]
fn image_present_on<D: ImageDaemon>(daemon: &D, image: &str) -> Result<bool> {
match daemon.inspect_image(image) {
ImagePresence::Present => Ok(true),
ImagePresence::Absent => Ok(false),
ImagePresence::Failed(msg) => Err(Error::Backend(format!("inspect image {image}: {msg}"))),
}
}
/// **Drive an image build over the [`ImageDaemon`] seam.** Tars `context_dir`, streams
/// the build, and fails on the first `error_detail` the daemon reports (an
/// [`Error::Backend`] naming the tag + message) — else returns `tag`. Everything but
/// the injected daemon is pure, so a mock proves the error-detail→`Err` and
/// returned-tag paths with no socket.
#[cfg(feature = "backend-oci")]
fn build_image_on<D: ImageDaemon>(
daemon: &D,
context_dir: &Path,
containerfile: &str,
tag: &str,
source_date_epoch: Option<i64>,
) -> Result<String> {
let tar = context_tar(context_dir)?;
for step in daemon.build_stream(tar, containerfile, tag, source_date_epoch) {
if let Some(msg) = step? {
return Err(Error::Backend(format!("build image {tag}: {msg}")));
}
}
Ok(tag.to_string())
}
/// **Drive a path extract over the [`ImageDaemon`] seam.** Downloads the path's tar
/// from a throwaway container, then unpacks it into `host_dest` (created if absent).
/// The tar the daemon hands back is rooted at the basename of `container_path`
/// (`podman cp` layout), so the unpacked tree lands at `host_dest/<basename>/…`. A
/// download error surfaces as `Err` **before** any unpack; only the injected daemon
/// touches a socket, so a mock proves the unpack + basename layout with no daemon.
#[cfg(feature = "backend-oci")]
fn extract_path_on<D: ImageDaemon>(
daemon: &D,
image: &str,
container_path: &str,
host_dest: &Path,
) -> Result<()> {
let tar_bytes = daemon.download_path_tar(image, container_path)?;
std::fs::create_dir_all(host_dest)
.map_err(|e| Error::Backend(format!("create extract dest {}: {e}", host_dest.display())))?;
tar::Archive::new(std::io::Cursor::new(tar_bytes))
.unpack(host_dest)
.map_err(|e| {
Error::Backend(format!(
"unpack extracted tar into {}: {e}",
host_dest.display()
))
})?;
Ok(())
}
// ---------------------------------------------------------------------------
// Image ARCHIVES — the export/import half of the one engine (the airgap path).
// ---------------------------------------------------------------------------
/// **What an image archive actually contains**, read out of the archive itself.
///
/// Deliberately NOT "what the caller named the file". An airgap bundle's whole
/// promise is that the target can trust what it received without a registry to ask,
/// so the identity of a packed image has to come from its bytes: the `RepoTags` and
/// the config digest inside the tar's own `manifest.json`.
///
/// The [`config_digest`](Self::config_digest) is the image's content identity (what
/// `podman images` shows as IMAGE ID). It is the ONLY thing tying a shipped image
/// back to the recipe that produced it — an image built on the build host is not
/// byte-reproducible from its `Containerfile` (timestamps, layer ordering, and any
/// network-fetched jar all differ), so "the recipe and the image agree" is a claim
/// no bundle can verify. Recording the digest at pack time and re-reading it on
/// unfold is what CAN be verified, and this type is that record.
///
/// Pure data — defined with no feature so a manifest elsewhere in the constellation
/// can name it without pulling `bollard`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ImageArchive {
/// Where the archive sits on disk.
pub path: std::path::PathBuf,
/// Archive size in bytes — the exported tar, uncompressed.
pub bytes: u64,
/// Every `RepoTags` entry across the archive's manifest, in archive order. May be
/// empty for an OCI-layout archive that names its content only by digest.
pub tags: Vec<String>,
/// The image config digest as bare lowercase hex, no `sha256:` prefix (the IMAGE
/// ID). Empty only if the archive named no config, which
/// [`inspect_image_archive`] treats as malformed rather than passing through.
pub config_digest: String,
}
/// **Why an image archive could not be believed.** Split into named cases because a
/// bundle that cannot tell "the file is not there" from "the file is half a file"
/// cannot report a useful failure to an operator standing at an offline box — and
/// because the second case is the one that silently produces a target running an
/// image that is missing layers.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ArchiveFault {
/// The file could not be opened or read at all (absent, permissions, IO).
Unreadable(String),
/// A tar entry's declared extent runs past the end of the file: the archive was
/// cut short (a dead socket mid-export, a partial copy, a full disk). `declared`
/// is where the last entry claims to end; `actual` is the file's real length.
Truncated {
/// Byte offset the last entry's header says its data ends at.
declared: u64,
/// The file's actual length.
actual: u64,
},
/// A well-formed tar that is not an image archive — no `manifest.json` at the
/// root. (An empty file lands here too, since an empty tar has no entries.)
NotAnImageArchive,
/// `manifest.json` was found but did not parse as the docker-archive shape.
Malformed(String),
}
impl std::fmt::Display for ArchiveFault {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
ArchiveFault::Unreadable(m) => write!(f, "unreadable: {m}"),
ArchiveFault::Truncated { declared, actual } => write!(
f,
"truncated: a tar entry declares bytes out to {declared} but the file is only {actual} bytes"
),
ArchiveFault::NotAnImageArchive => {
write!(f, "not an image archive: no manifest.json at the tar root")
}
ArchiveFault::Malformed(m) => write!(f, "malformed manifest.json: {m}"),
}
}
}
/// Chunk size for streaming an archive INTO the daemon (`/images/load`). 4 MiB is
/// large enough that a 1.4 GB image is ~350 round trips rather than ~350 000, and
/// small enough that peak memory is a rounding error. Peak resident is one chunk.
#[cfg(feature = "backend-oci")]
const IMPORT_CHUNK: usize = 4 * 1024 * 1024;
/// Sanity bound on `manifest.json`. A real one is a few KB even for an archive of
/// many images; 8 MiB is far past any plausible value. It exists because the size
/// is read out of a tar header in a file whose trustworthiness is exactly what is
/// being decided, so it must never be used as an unchecked allocation hint.
#[cfg(feature = "backend-oci")]
const MAX_MANIFEST_BYTES: u64 = 8 * 1024 * 1024;
/// **Read an image archive's identity — with NO daemon and NO socket.**
///
/// Opens `path`, walks the tar's *headers* (seek-skipping every entry body, so a
/// 1.4 GB archive costs a handful of seeks rather than a 1.4 GB read), checks the
/// last entry's declared extent against the real file length, and parses the root
/// `manifest.json`.
///
/// This is the function a bundle's verify-on-receipt calls. It is deliberately
/// engine-free: an airgapped target must be able to reject a bad archive **before**
/// it starts a container engine, and a build host must be able to check what it
/// packed without one either. Requires `backend-oci` for the `tar` + `serde_json`
/// it rides.
#[cfg(feature = "backend-oci")]
pub fn inspect_image_archive(path: &Path) -> Result<ImageArchive> {
let file = std::fs::File::open(path)
.map_err(|e| archive_error(path, &ArchiveFault::Unreadable(e.to_string())))?;
let len = file
.metadata()
.map_err(|e| archive_error(path, &ArchiveFault::Unreadable(e.to_string())))?
.len();
let (tags, config_digest) =
scan_image_archive(file, len).map_err(|fault| archive_error(path, &fault))?;
Ok(ImageArchive {
path: path.to_path_buf(),
bytes: len,
tags,
config_digest,
})
}
/// Lean build: no `tar`/`serde_json`, so there is nothing to read an archive with.
#[cfg(not(feature = "backend-oci"))]
pub fn inspect_image_archive(path: &Path) -> Result<ImageArchive> {
let _ = path;
Err(Error::Unsupported(
"reading an image archive needs the `backend-oci` feature (tar + serde_json)".into(),
))
}
/// One place that turns an [`ArchiveFault`] into the crate's [`Error`], so every
/// caller names the file the same way and the fault's own wording survives.
#[cfg(feature = "backend-oci")]
fn archive_error(path: &Path, fault: &ArchiveFault) -> Error {
Error::Backend(format!("image archive {}: {fault}", path.display()))
}
/// **The pure scanner** behind [`inspect_image_archive`] — generic over `Read + Seek`
/// so the whole truncation / no-manifest / malformed-JSON decision tree is drivable
/// from an in-memory `Cursor` fixture with no filesystem and no daemon. Returns the
/// archive's `(tags, config_digest)`.
///
/// `len` is passed in rather than measured because a `Cursor` and a `File` learn
/// their length differently, and the truncation check needs the authoritative one.
#[cfg(feature = "backend-oci")]
fn scan_image_archive<R: std::io::Read + std::io::Seek>(
reader: R,
len: u64,
) -> std::result::Result<(Vec<String>, String), ArchiveFault> {
let mut archive = tar::Archive::new(reader);
let entries = archive
.entries_with_seek()
.map_err(|e| ArchiveFault::Unreadable(e.to_string()))?;
let mut manifest: Option<Vec<u8>> = None;
// The furthest byte any entry header claims the archive extends to. Compared
// against the real length once the walk is done — a stream cut mid-layer leaves
// a perfectly parseable header whose data simply is not all there, and header
// parsing alone will never notice.
let mut declared_end: u64 = 0;
for entry in entries {
// A header that will not parse is a cut in the HEADER itself; report it as
// read failure with the real cause rather than inventing byte offsets for a
// `Truncated` we cannot actually measure.
let mut entry = entry.map_err(|e| {
ArchiveFault::Unreadable(format!(
"tar entry could not be read (archive is {len} bytes): {e}"
))
})?;
let size = entry.size();
let data_at = entry.raw_file_position();
// tar pads every entry body out to a 512-byte block.
let padded = size.div_ceil(512).saturating_mul(512);
declared_end = declared_end.max(data_at.saturating_add(padded));
let is_manifest = entry
.path()
.map(|p| p.as_ref() == Path::new("manifest.json"))
.unwrap_or(false);
if is_manifest {
// Bounded: a real manifest.json is a few KB. The size comes from a tar
// header inside a file we are still deciding whether to trust, so it is
// never used as an allocation hint unchecked — a corrupt header claiming
// 4 EiB must not become a 4 EiB reserve.
if size > MAX_MANIFEST_BYTES {
return Err(ArchiveFault::Malformed(format!(
"manifest.json declares {size} bytes, over the {MAX_MANIFEST_BYTES}-byte \
sanity bound — this is not a real image manifest"
)));
}
let mut buf = Vec::with_capacity(size as usize);
std::io::Read::read_to_end(&mut entry, &mut buf)
.map_err(|e| ArchiveFault::Unreadable(format!("read manifest.json: {e}")))?;
manifest = Some(buf);
}
}
if declared_end > len {
return Err(ArchiveFault::Truncated {
declared: declared_end,
actual: len,
});
}
let manifest = manifest.ok_or(ArchiveFault::NotAnImageArchive)?;
parse_archive_manifest(&manifest)
}
/// **Parse a docker-archive `manifest.json`** into `(tags, config_digest)`. Pure —
/// bytes in, answer out.
///
/// The shape is a JSON array of entries, each with `Config` (a path to the config
/// blob) and `RepoTags`. `Config` appears as either `<hex>.json` (docker) or
/// `blobs/sha256/<hex>` (podman's newer writer), so the digest is taken as the last
/// path segment with any `.json` suffix stripped — both spellings land on the same
/// hex.
#[cfg(feature = "backend-oci")]
fn parse_archive_manifest(
bytes: &[u8],
) -> std::result::Result<(Vec<String>, String), ArchiveFault> {
let value: serde_json::Value =
serde_json::from_slice(bytes).map_err(|e| ArchiveFault::Malformed(e.to_string()))?;
let entries = value
.as_array()
.ok_or_else(|| ArchiveFault::Malformed("top level is not a JSON array".into()))?;
if entries.is_empty() {
return Err(ArchiveFault::Malformed(
"manifest.json is an empty array — the archive names no image".into(),
));
}
let mut tags: Vec<String> = Vec::new();
for entry in entries {
if let Some(list) = entry.get("RepoTags").and_then(|t| t.as_array()) {
tags.extend(list.iter().filter_map(|t| t.as_str()).map(str::to_string));
}
}
let config = entries[0]
.get("Config")
.and_then(|c| c.as_str())
.ok_or_else(|| ArchiveFault::Malformed("first entry has no Config".into()))?;
let digest = config
.rsplit('/')
.next()
.unwrap_or(config)
.trim_end_matches(".json")
.to_string();
if digest.is_empty() {
return Err(ArchiveFault::Malformed(format!(
"Config {config:?} yields no digest"
)));
}
Ok((tags, digest))
}
/// **Drive an image export over the [`ImageDaemon`] seam.** Presence is probed
/// FIRST, so "you never built this image" and "your container engine is down" stay
/// two different sentences and neither one leaves a stub file behind. Then the
/// archive is streamed to `dest` and immediately read back
/// ([`inspect_image_archive`]) — a stream that died mid-layer becomes an
/// [`ArchiveFault::Truncated`] here, at pack time, rather than a mystery on the
/// airgapped box a week later.
///
/// The archive's own tags are checked against `image`: an archive that does not
/// contain the tag it was asked for is a hard error, because the tag is where an
/// image's *pairing* lives (korp's `nordisk-falkordb-valkey:9-<date>` encodes both
/// the Valkey major and the FalkorDB module version, and a bundle silently carrying
/// the wrong one is worse than a bundle carrying nothing). An archive with NO tags
/// at all is accepted — an OCI-layout export identifies by digest alone, and the
/// digest is recorded.
///
/// Everything but the injected daemon is pure, so a mock proves all of it with no
/// socket.
#[cfg(feature = "backend-oci")]
fn export_image_on<D: ImageDaemon>(daemon: &D, image: &str, dest: &Path) -> Result<ImageArchive> {
match daemon.inspect_image(image) {
ImagePresence::Present => {}
ImagePresence::Absent => {
return Err(Error::Backend(format!(
"export image {image}: no such image locally — nothing was written to {}",
dest.display()
)));
}
ImagePresence::Failed(msg) => {
return Err(Error::Backend(format!(
"export image {image}: container engine unreachable: {msg}"
)));
}
}
if let Some(parent) = dest.parent().filter(|p| !p.as_os_str().is_empty()) {
std::fs::create_dir_all(parent).map_err(|e| {
Error::Backend(format!("create export dir {}: {e}", parent.display()))
})?;
}
{
let file = std::fs::File::create(dest).map_err(|e| {
Error::Backend(format!("create export file {}: {e}", dest.display()))
})?;
let mut sink = std::io::BufWriter::new(file);
daemon.export_to(image, &mut sink)?;
// Flush explicitly: a BufWriter dropped with a failing flush swallows the
// error, which would hand back a short archive that looks written.
std::io::Write::flush(&mut sink).map_err(|e| {
Error::Backend(format!("flush export file {}: {e}", dest.display()))
})?;
}
let archive = inspect_image_archive(dest)?;
if !archive.tags.is_empty() && !archive.tags.iter().any(|t| t == image) {
return Err(Error::Backend(format!(
"export image {image}: the archive at {} carries {:?}, not {image} — refusing to \
record an image under a tag it does not have",
dest.display(),
archive.tags
)));
}
Ok(archive)
}
/// **Drive an image import over the [`ImageDaemon`] seam.** Verifies the archive
/// BEFORE handing it to the daemon, so a truncated or non-image file costs a few
/// seeks instead of a gigabyte pushed at a socket followed by a store in a state
/// nobody can describe. Returns what was loaded, read from the archive.
#[cfg(feature = "backend-oci")]
fn import_image_on<D: ImageDaemon>(daemon: &D, src: &Path) -> Result<ImageArchive> {
let archive = inspect_image_archive(src)?;
daemon.load_archive(src)?;
Ok(archive)
}
// ---------------------------------------------------------------------------
// Socket reachability — WHY the API socket cannot be used, said honestly.
// ---------------------------------------------------------------------------
//
// This whole section exists because of one line that used to read:
//
// if !std::path::Path::new(path).exists() { "…socket not found at {path}…" }
//
// `Path::exists()` is `fs::metadata(path).is_ok()`, so it collapses EVERY stat
// failure into "does not exist". **EACCES and ENOENT are not the same fact and
// they do not have the same fix**, and reporting the first as the second sends
// the reader to enable a unit that is already running.
//
// MEASURED on oden 2026-08-11 — the misreport that cost the debugging time:
//
// draupnir: backend error: podman/Docker API socket not found at
// /run/podman/podman.sock (enable with `systemctl --user enable --now
// podman.socket`, …)
//
// while at the same moment `systemctl is-active podman.socket` said **active**,
// `sudo podman ps` listed a running container, and the socket was right there:
// `/run/podman/podman.sock` `root root 0660` inside `/run/podman` `root root
// 0700`. korp-server runs `User=korp` (uid 111), so the stat was refused at the
// DIRECTORY, and `exists()` turned "you may not look" into "it is not there".
//
// Two further defects in that one message, both fixed here:
//
// * **Scope.** It always said `systemctl --user`, even when the path came from
// a `DOCKER_HOST` naming a SYSTEM socket (`/run/podman/podman.sock`) — i.e.
// wrong advice for precisely the configuration that produced it. A path under
// `/run/user/<uid>/` is the rootless USER socket; anything else is the system
// one, and the two have different unit scopes and different fixes.
//
// * **Silence about ownership.** The one thing the reader needs for an EACCES
// — who owns it, what group, what mode — was never printed, though the
// process could stat the blocking component perfectly well.
//
// Everything below is pure std and **not** behind `backend-oci`, deliberately:
// the diagnosis is the part worth testing, and gating it would mean the lean
// default build never exercises it.
/// Owner, group and permission bits of one path component, with uid/gid resolved
/// to NAMES where `/etc/passwd` and `/etc/group` can resolve them.
///
/// Names are read out of those files with plain `std::fs` — no `getpwuid` FFI and
/// no `id`/`stat` subprocess (LAW: pure Rust, zero shell). An unresolvable id is
/// rendered as its decimal number, which is honest rather than blank.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Ownership {
/// Owning user — name if resolvable, else the decimal uid.
pub user: String,
/// Owning group — name if resolvable, else the decimal gid.
pub group: String,
/// Permission bits only (`mode & 0o7777`).
pub mode: u32,
}
impl std::fmt::Display for Ownership {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}:{} {:04o}", self.user, self.group, self.mode)
}
}
/// Resolve a numeric id to a name out of a colon-separated `/etc/passwd`-style
/// database (`name:x:id:…`), falling back to the decimal id.
///
/// One parser for both files because `/etc/passwd` and `/etc/group` agree on the
/// first three fields. Reading the file is the pure-Rust route; the FFI
/// (`getpwuid_r`) is the thing the LAW keeps for last resort, and it is not
/// needed for a diagnostic string.
fn name_for_id(db: &str, id: u32) -> Option<String> {
let text = std::fs::read_to_string(db).ok()?;
for line in text.lines() {
let mut f = line.split(':');
let name = f.next()?;
let _passwd = f.next();
if f.next().and_then(|n| n.trim().parse::<u32>().ok()) == Some(id) {
return Some(name.to_string());
}
}
None
}
/// Ownership of an already-stat'd path, names resolved where possible.
fn ownership_of(md: &std::fs::Metadata) -> Ownership {
use std::os::unix::fs::MetadataExt;
let (uid, gid) = (md.uid(), md.gid());
Ownership {
user: name_for_id("/etc/passwd", uid).unwrap_or_else(|| uid.to_string()),
group: name_for_id("/etc/group", gid).unwrap_or_else(|| gid.to_string()),
mode: md.mode() & 0o7777,
}
}
/// **What a stat of the container API socket path actually found** — the fact the
/// old `Path::exists()` threw away.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SocketProbe {
/// The path stats. The socket is there and this process can at least reach it.
Reachable,
/// **ENOENT** — genuinely absent. Nothing is listening and nothing created it.
Absent,
/// **EACCES/EPERM** — something on the path is there and this process may not
/// get past it. `blocked_at` is the deepest component that DID stat (i.e. the
/// one whose permissions are doing the blocking, or the socket itself), with
/// its ownership when it could be read.
Denied {
/// The path component the probe was stopped at.
blocked_at: String,
/// Its owner/group/mode, when the component could be stat'd.
owner: Option<Ownership>,
},
/// Any other stat failure (ELOOP, ENOTDIR, EIO …) — carried verbatim rather
/// than folded into one of the above, because guessing is what this whole
/// section is a fix for.
Other(String),
}
/// Stat `path` and classify the outcome — the honest replacement for
/// `Path::exists()`.
///
/// On `PermissionDenied` it walks UP the ancestors to find the deepest component
/// it *can* stat, because that component is the one whose mode is refusing the
/// traversal. On oden that turns a useless "not found at
/// /run/podman/podman.sock" into "refused at /run/podman (root:root 0700)",
/// which names the actual obstacle.
pub fn probe_socket_path(path: &Path) -> SocketProbe {
match std::fs::metadata(path) {
Ok(_) => SocketProbe::Reachable,
Err(e) => match e.kind() {
std::io::ErrorKind::NotFound => SocketProbe::Absent,
std::io::ErrorKind::PermissionDenied => {
// The first ancestor that stats is the gatekeeper: everything
// below it was invisible precisely because of its mode.
for anc in path.ancestors().skip(1) {
if let Ok(md) = std::fs::metadata(anc) {
return SocketProbe::Denied {
blocked_at: anc.display().to_string(),
owner: Some(ownership_of(&md)),
};
}
}
SocketProbe::Denied {
blocked_at: path.display().to_string(),
owner: None,
}
}
_ => SocketProbe::Other(e.to_string()),
},
}
}
/// **Is this the rootless USER socket or the SYSTEM one?** — the distinction the
/// old hint ignored, and got backwards for the exact path that produced it.
///
/// A rootless podman socket lives under `/run/user/<uid>/`, is owned by that user
/// and is managed by that user's `systemd --user`. Anything else (`/run/podman/…`,
/// `/var/run/docker.sock`) belongs to the SYSTEM manager, where `systemctl --user`
/// is not merely unhelpful but addresses a different init instance entirely.
pub fn is_user_scope_socket(path: &Path) -> bool {
path.starts_with("/run/user/") || path.starts_with("/var/run/user/")
}
/// The `systemctl` invocation that would actually apply to THIS socket's scope.
fn enable_hint(path: &Path) -> &'static str {
if is_user_scope_socket(path) {
"systemctl --user enable --now podman.socket"
} else {
"sudo systemctl enable --now podman.socket"
}
}
/// **Turn a probe into the message a reader can act on** — `None` when the socket
/// is reachable, `Some(reason)` when it is not.
///
/// Split out from the stat so the wording is testable with no filesystem at all:
/// every branch here is a pure function of the probe.
pub fn socket_unreachable_reason(path: &Path, probe: &SocketProbe) -> Option<String> {
let p = path.display();
match probe {
SocketProbe::Reachable => None,
SocketProbe::Absent => Some(format!(
"podman/Docker API socket not found at {p} — nothing exists at that path \
(enable with `{}`, or point DOCKER_HOST at a running socket)",
enable_hint(path)
)),
SocketProbe::Denied { blocked_at, owner } => {
let mut m = format!(
"podman/Docker API socket at {p} EXISTS but this process may not reach it \
(permission denied) — do NOT enable the unit, it is already there. \
Refused at {blocked_at}"
);
match owner {
Some(o) => {
m.push_str(&format!(" ({o})."));
// The honest split: a group other than root is something a
// service declaration can actually join. `root` is not — saying
// "add yourself to root" would be advice nobody should take.
if o.group == "root" || o.group == "0" {
m.push_str(
" Its group is root, so joining a group CANNOT fix this: \
either the host gives the socket a non-root group \
(a `SocketGroup=` drop-in on podman.socket, a HOST decision) \
or the service runs as root. Adding a service account to the \
root group is not a fix.",
);
} else {
m.push_str(&format!(
" Grant access by making the service account a member of `{}` — \
for a systemd unit that is `SupplementaryGroups={}`, \
which skidbladnir renders from the service's own declaration.",
o.group, o.group
));
}
}
None => m.push('.'),
}
Some(m)
}
SocketProbe::Other(e) => Some(format!(
"podman/Docker API socket at {p} could not be examined: {e}"
)),
}
}
/// The one call the engine makes: stat, classify, and word it. `None` ⇒ go ahead.
pub fn socket_unreachable(path: &Path) -> Option<String> {
socket_unreachable_reason(path, &probe_socket_path(path))
}
// ---------------------------------------------------------------------------
// ROOTLESS ENGINE READINESS — what the engine REQUIRES of the host, and which
// half of it draupnir owns.
// ---------------------------------------------------------------------------
//
// The 2026-08-11 oden setup took five hand-typed commands to get korp-server
// (`User=korp`, uid 111, `nologin`) a podman socket. They split cleanly in two,
// and the split is the design:
//
// HOST / ACCOUNT — `loginctl enable-linger korp`, `/etc/subuid`, `/etc/subgid`.
// Nothing to do with containers. They are logind and shadow-utils facts about
// an ACCOUNT, they need root, and skidbladnir already owns account and service
// lifecycle and already drives systemd over D-Bus. It performs them
// (`skidbladnir::rootless`); draupnir does not, and must not grow a second
// `/etc/subuid` writer.
//
// ENGINE — where the socket is, whether it answers, whether `podman system
// migrate` is owed. These are facts about podman, which is this module's
// subject. draupnir STATES them and VERIFIES them; it mutates nothing.
//
// So: skidbladnir sets the host up, draupnir says what "set up" has to mean and
// checks afterwards whether it took. Neither repeats the other, and the two facts
// they share (the uid, and the socket path derived from it) are derived by the
// same arithmetic on both sides rather than passed between them.
/// **Where a rootless podman puts its API socket for `uid`.**
///
/// One place, so that a service unit's `DOCKER_HOST`, this engine's connect path
/// and skidbladnir's post-setup verification cannot disagree by a directory. On
/// oden this resolved to `/run/user/111/podman/podman.sock` — VERIFIED as
/// `srw-rw---- korp korp` once lingering was on.
pub fn rootless_socket_path(uid: u32) -> std::path::PathBuf {
std::path::PathBuf::from(format!("/run/user/{uid}/podman/podman.sock"))
}
/// **The uid this process is actually running as**, read from `/proc/self/status`.
///
/// Zero shell (no `id -u`) and no FFI (no `getuid()`), per the LAW — the kernel
/// publishes it as text and `std::fs` can read text. The `Uid:` line is
/// `real effective saved filesystem`; the REAL uid is taken, because that is the
/// uid logind created `/run/user/<uid>` for.
///
/// `None` off Linux or where `/proc` is not mounted — in which case a caller must
/// be told the uid rather than guess one, which is precisely the defect this
/// replaced: [`engine_socket_url`] fell back to a hardcoded `/run/user/1000`, so
/// a service running as uid 111 with no `DOCKER_HOST` looked for its socket in
/// *another account's* runtime directory and got EACCES or ENOENT depending on
/// that account's luck.
pub fn current_uid() -> Option<u32> {
let status = std::fs::read_to_string("/proc/self/status").ok()?;
parse_uid_line(&status)
}
/// **The API socket URL this engine connects to** — resolved the ONE way, and public
/// so a caller can ask *before* it asks the engine to do anything.
///
/// Honour `DOCKER_HOST`, else the rootless user socket under `XDG_RUNTIME_DIR`
/// (parity with jera), else the runtime directory of the uid this process is
/// ACTUALLY running as. The ladder is pure in [`engine_socket_url_from`]; this reads
/// the environment and the kernel and applies it.
///
/// That last fallback used to be the literal string `/run/user/1000`, and it is
/// wrong for precisely the deployment this engine exists in. A systemd system unit
/// with `User=korp` runs as uid 111 and is given NO `XDG_RUNTIME_DIR` (only a
/// session or a `--user` manager sets one), so with no `DOCKER_HOST` the engine went
/// looking in **uid 1000's** runtime directory — another account's `/run/user/1000`,
/// mode 0700. On a box where that account is logged in the answer is EACCES, on a
/// box where it is not the answer is ENOENT, and neither mentions the actual
/// mistake. `1000` is a desktop's first user, not a fact about the process.
pub fn engine_socket_url() -> String {
engine_socket_url_from(
std::env::var("DOCKER_HOST").ok().as_deref(),
std::env::var("XDG_RUNTIME_DIR").ok().as_deref(),
current_uid(),
)
}
/// The three-rung ladder behind [`engine_socket_url`], with every input passed in
/// so it can be tested without touching the process environment.
pub fn engine_socket_url_from(
docker_host: Option<&str>,
xdg_runtime_dir: Option<&str>,
uid: Option<u32>,
) -> String {
if let Some(h) = docker_host {
return h.to_string();
}
if let Some(xdg) = xdg_runtime_dir {
return format!("unix://{xdg}/podman/podman.sock");
}
match uid {
Some(uid) => format!("unix://{}", rootless_socket_path(uid).display()),
// No /proc: say which socket is meant rather than guess a uid. The system
// socket is the only one whose path does not depend on an identity we could
// not read.
None => "unix:///run/podman/podman.sock".to_string(),
}
}
/// **Can the engine's socket be reached right now — WITHOUT connecting?** — the URL
/// the engine would use, and what a stat of it found.
///
/// This exists for the caller that has to decide whether to WAIT: a rootless
/// `podman.socket` is created by the account's `systemd --user`, and a system
/// service for the same account can be started in the same second as that manager
/// (MEASURED on oden 2026-09-10: korp-server, `user@111.service` and
/// `podman.socket` all at 16:57:05, korp-server first). An absent socket at that
/// instant is a race, not a fact; a denied one is a fact. The engine cannot tell
/// the two apart from inside `connect`, where every failure has to be terminal, so
/// the probe is offered separately, ahead of it.
///
/// `None` for a non-`unix://` `DOCKER_HOST` (`tcp://…`): there is nothing to stat,
/// and only the connect can say.
pub fn engine_socket_probe() -> (String, Option<SocketProbe>) {
let url = engine_socket_url();
let probe = url
.strip_prefix("unix://")
.map(|path| probe_socket_path(Path::new(path)));
(url, probe)
}
/// Pull the real uid out of `/proc/<pid>/status` text (pure → testable).
fn parse_uid_line(status: &str) -> Option<u32> {
for line in status.lines() {
if let Some(rest) = line.strip_prefix("Uid:") {
return rest.split_whitespace().next()?.parse().ok();
}
}
None
}
/// **Is `podman system migrate` owed for this account?**
///
/// It is the one command from the oden session draupnir will NOT run, and saying
/// so plainly is better than a wrapper that pretends. Two independent reasons:
///
/// 1. **There is no API for it.** The libpod REST surface exposes
/// `/libpod/system/{info,df,prune,events,version}` and no `migrate`. It exists
/// only in the CLI, so "running" it means `Command::new("podman")` — a
/// shell-out under the LAW, and this crate deleted its last two of those (the
/// `podman build`/`podman run -v` subprocesses) to get here.
/// 2. **It is usually not owed at all.** What it does is re-create the user
/// namespace mapping recorded for EXISTING containers and pods. An account that
/// has never run podman has no stored mappings, so a fresh account picks the new
/// `/etc/subuid` range up on its first run and needs nothing. It is owed only
/// when podman ran BEFORE the subordinate-id grant — which is exactly what
/// happened on oden, and is exactly why it was needed there.
///
/// The detection is therefore "has podman ever run as this account", i.e. does its
/// libpod storage exist. That is a fact on disk, not a guess.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MigrateVerdict {
/// podman has no stored state for this account: the next run adopts the new
/// range by itself.
NotOwed(String),
/// podman has run here before. If that was before the subordinate-id grant, the
/// stored mappings are stale and only `podman system migrate` refreshes them.
Owed(String),
}
impl MigrateVerdict {
pub fn detail(&self) -> &str {
match self {
MigrateVerdict::NotOwed(s) | MigrateVerdict::Owed(s) => s,
}
}
}
/// Decide from the ONE fact that decides it: does this account already have libpod
/// storage. Pure — `home` and the existence flag are passed in — so both arms are
/// testable without a podman install.
pub fn migrate_verdict(account: &str, storage_exists: bool) -> MigrateVerdict {
if !storage_exists {
MigrateVerdict::NotOwed(format!(
"podman has never run as `{account}` (no libpod storage), so there are no \
stored user-namespace mappings to refresh — the first run adopts the \
/etc/subuid grant by itself. `podman system migrate` is not needed."
))
} else {
MigrateVerdict::Owed(format!(
"podman has already run as `{account}` and holds stored user-namespace \
mappings. If the /etc/subuid grant was made AFTER that, they are stale and \
every container keeps the old single-id mapping. Only `podman system \
migrate`, run AS `{account}`, refreshes them — and draupnir will not run \
it: there is no libpod REST endpoint for it, so running it means a podman \
subprocess, which this crate does not do. Run it once, as the service \
account, or delete the account's containers and let them be re-created."
))
}
}
/// The libpod storage directory for an account — the path [`migrate_verdict`] asks
/// about. `XDG_DATA_HOME` is not consulted: a `nologin` service account has no
/// session to set it, which is the whole shape of this problem.
pub fn libpod_storage_dir(home: &str) -> std::path::PathBuf {
std::path::Path::new(home).join(".local/share/containers/storage/libpod")
}
/// **Everything the engine needs the host to have already done**, checked.
///
/// This is the "did it take?" side of the seam: skidbladnir applies the host setup,
/// then this says whether a rootless engine can now actually be reached for that
/// uid, and what the remaining obstacle is if not.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RootlessReadiness {
pub uid: u32,
pub socket: std::path::PathBuf,
pub probe: SocketProbe,
/// `None` when the engine is reachable; otherwise the obstacle, worded for a
/// reader — including the ENOENT/EACCES distinction the rest of this section
/// exists to preserve.
pub obstacle: Option<String>,
}
impl RootlessReadiness {
pub fn is_ready(&self) -> bool {
self.obstacle.is_none()
}
}
/// Probe the rootless socket for `uid` and word the result.
///
/// Note what this does NOT say. An `Absent` rootless socket has a *different* first
/// question from an absent system socket: not "is the unit enabled" but "does
/// `/run/user/<uid>` exist at all", because without lingering it does not, and
/// `systemctl --user` inside it has nothing to talk to. So the wording branches on
/// the runtime directory before it branches on the unit.
pub fn rootless_readiness(uid: u32) -> RootlessReadiness {
let socket = rootless_socket_path(uid);
let probe = probe_socket_path(&socket);
let runtime_dir = std::path::PathBuf::from(format!("/run/user/{uid}"));
let obstacle = match &probe {
SocketProbe::Reachable => None,
SocketProbe::Absent if !runtime_dir.exists() => Some(format!(
"no rootless engine for uid {uid}: {} does not exist, so there is no user \
manager to enable `podman.socket` in and nowhere for the socket to live. \
The account must LINGER first — `skidbladnir::rootless` performs that over \
org.freedesktop.login1 — and the runtime directory then appears by itself.",
runtime_dir.display()
)),
_ => socket_unreachable_reason(&socket, &probe),
};
RootlessReadiness {
uid,
socket,
probe,
obstacle,
}
}
// ---------------------------------------------------------------------------
// Engine — the real podman/Docker REST engine (feature `backend-oci`).
// ---------------------------------------------------------------------------
/// A shared, append-only log buffer the follow tasks push into and the caller
/// drains. Each entry is `(is_stderr, line)` so a drain can either flatten to the
/// combined ordered stream ([`Engine::drain_logs`]) or split it back into
/// stdout/stderr ([`Engine::drain_logs_split`]) — jera's `run_container` keeps the
/// two apart, so preserving the tag makes that repoint lossless.
#[cfg(feature = "backend-oci")]
type LogBuf = Arc<Mutex<Vec<(bool, String)>>>;
/// **Turn engine log FRAMES into lines** and append them, tagged by stream.
///
/// One joiner for both readers — the follow task and the one-shot
/// [`Engine::read_all_logs`] — because a frame is a chunk of the daemon's
/// multiplexed stream and not, by contract, a single line. MEASURED 2026-10-21 on
/// podman: a line longer than 8 KiB (a bench's one-line JSON envelope) comes as
/// several frames that end mid-line, and each one handed on as a "line" cut the
/// envelope apart, so its reader found none. A frame's text after its last
/// newline is held per stream and joined with that stream's next frame;
/// [`LineJoin::flush`] hands on what is left when the stream ends.
#[cfg(any(feature = "backend-oci", test))]
#[derive(Default)]
struct LineJoin {
/// The unterminated tail of stdout (`[0]`) and stderr (`[1]`).
held: [String; 2],
}
#[cfg(any(feature = "backend-oci", test))]
impl LineJoin {
fn push(&mut self, into: &mut Vec<(bool, String)>, is_err: bool, text: &str) {
let held = &mut self.held[is_err as usize];
held.push_str(text);
let Some(end) = held.rfind('\n') else { return };
let rest = held.split_off(end + 1);
for line in held.split('\n') {
let line = line.trim_end_matches('\r');
if !line.is_empty() {
into.push((is_err, line.to_string()));
}
}
*held = rest;
}
fn flush(&mut self, into: &mut Vec<(bool, String)>) {
for (i, held) in self.held.iter_mut().enumerate() {
let line = std::mem::take(held);
let line = line.trim_end_matches('\r');
if !line.is_empty() {
into.push((i == 1, line.to_string()));
}
}
}
}
#[cfg(feature = "backend-oci")]
fn push_log_frame(join: &mut LineJoin, into: &mut Vec<(bool, String)>, out: &bollard::container::LogOutput) {
let is_err = matches!(out, bollard::container::LogOutput::StdErr { .. });
join.push(into, is_err, &bollard::container::LogOutput::to_string(out));
}
/// The live bollard engine: a `Docker` handle + a dedicated tokio runtime that
/// drives its async API from draupnir's synchronous [`Boot`]/[`Lifecycle`] seam.
#[cfg(feature = "backend-oci")]
struct Engine {
docker: bollard::Docker,
rt: EngineRuntime,
/// The URL [`engine_socket_url`] resolved, kept so the two state reads
/// ([`Engine::list_containers`], [`Engine::container_state`]) can talk to the
/// SAME daemon over a raw `GET` — see [`daemon_get`] for why they must.
socket: String,
/// Per-container streamed-log buffers, filled by the follow tasks, drained by
/// [`Engine::drain_logs`] (the API-streamed equivalent of reader threads).
logs: Mutex<std::collections::HashMap<String, LogBuf>>,
/// Leftover containers this engine REAPED to free a deterministic name,
/// awaiting a drain by the consumer. draupnir never prints, but it never
/// destroys anything silently either.
reaps: Mutex<Vec<Reap>>,
}
/// The engine's own tokio runtime, shut down **in the background** when the
/// engine drops. A plain `Runtime` dropped inside an async context (a consumer
/// that drops its `ContainerBoot` in an `async fn`, e.g. nornir's startup probe)
/// panics: "Cannot drop a runtime in a context where blocking is not allowed".
/// `shutdown_background` never blocks, so the engine is safe to drop anywhere.
#[cfg(feature = "backend-oci")]
struct EngineRuntime(Option<tokio::runtime::Runtime>);
#[cfg(feature = "backend-oci")]
impl From<tokio::runtime::Runtime> for EngineRuntime {
fn from(rt: tokio::runtime::Runtime) -> Self {
Self(Some(rt))
}
}
#[cfg(feature = "backend-oci")]
impl std::ops::Deref for EngineRuntime {
type Target = tokio::runtime::Runtime;
fn deref(&self) -> &tokio::runtime::Runtime {
self.0.as_ref().expect("the engine runtime lives until the engine drops")
}
}
#[cfg(feature = "backend-oci")]
impl Drop for EngineRuntime {
fn drop(&mut self) {
if let Some(rt) = self.0.take() {
rt.shutdown_background();
}
}
}
/// **A raw `GET` against the podman/Docker REST API, body returned as text** — the
/// transport under the two state reads, and the ONLY place draupnir bypasses
/// bollard's typed API.
///
/// It exists because bollard's models are generated from **Docker's** OpenAPI
/// document and podman answers with **podman's** vocabulary: a single container in
/// state `stopped` fails the deserialisation of the whole `/containers/json`
/// response (measured on oden, see [`container_state_from_daemon`]). bollard has no
/// escape hatch for the raw body, so the two calls whose answer is a state word
/// read the JSON themselves and parse it permissively. Everything else — create,
/// start, stop, exec, logs, images, archives — stays on bollard: this is a narrow
/// hole for one wrong type, not a second engine.
///
/// No new crate enters the graph: `hyper` / `hyper-util` / `http-body-util` are
/// already there as bollard's own transport, now named directly.
///
/// `unix://…` connects the socket; anything else is taken as `host:port` (with a
/// `tcp://` / `http://` scheme stripped), which is what `DOCKER_HOST` carries for a
/// remote daemon. A non-2xx status is an [`Error::Backend`] carrying the daemon's
/// own body, so a 404 stays distinguishable from a transport failure.
#[cfg(feature = "backend-oci")]
async fn daemon_get(socket_url: &str, path: &str) -> Result<String> {
use http_body_util::{BodyExt, Empty};
use hyper::body::Bytes;
use hyper_util::rt::TokioIo;
async fn send<I>(io: I, path: &str) -> Result<String>
where
I: hyper::rt::Read + hyper::rt::Write + Unpin + Send + 'static,
{
let (mut sender, conn) = hyper::client::conn::http1::handshake(io)
.await
.map_err(|e| Error::Backend(format!("daemon handshake: {e}")))?;
// The connection task drives the socket; it ends when the response is done.
tokio::spawn(async move {
let _ = conn.await;
});
let req = hyper::Request::builder()
.method("GET")
.uri(path)
// A unix socket has no authority, but HTTP/1.1 demands a Host header.
.header(hyper::header::HOST, "docker")
.body(Empty::<Bytes>::new())
.map_err(|e| Error::Backend(format!("daemon request {path}: {e}")))?;
let res = sender
.send_request(req)
.await
.map_err(|e| Error::Backend(format!("daemon GET {path}: {e}")))?;
let status = res.status();
let body = res
.into_body()
.collect()
.await
.map_err(|e| Error::Backend(format!("daemon GET {path}: reading body: {e}")))?
.to_bytes();
let text = String::from_utf8_lossy(&body).into_owned();
if !status.is_success() {
return Err(Error::Backend(format!(
"daemon GET {path}: HTTP {}: {}",
status.as_u16(),
text.trim()
)));
}
Ok(text)
}
if let Some(p) = socket_url.strip_prefix("unix://") {
let stream = tokio::net::UnixStream::connect(p)
.await
.map_err(|e| Error::Backend(format!("connect container socket {p}: {e}")))?;
send(TokioIo::new(stream), path).await
} else {
let addr = socket_url
.strip_prefix("tcp://")
.or_else(|| socket_url.strip_prefix("http://"))
.unwrap_or(socket_url)
.trim_end_matches('/');
let stream = tokio::net::TcpStream::connect(addr)
.await
.map_err(|e| Error::Backend(format!("connect container daemon {addr}: {e}")))?;
send(TokioIo::new(stream), path).await
}
}
#[cfg(feature = "backend-oci")]
impl Engine {
/// Connect over the API socket. Does NOT try to start a daemon (zero-shell) —
/// returns a clear [`Error::Backend`] if the socket is absent/unreachable.
///
/// The URL is [`engine_socket_url`] and the reachability check is the same stat
/// [`engine_socket_probe`] offers to callers ahead of time — one resolution, one
/// classification, whether the question is asked before the connect or by it.
fn connect() -> Result<Self> {
let (url, probe) = engine_socket_probe();
// ENOENT and EACCES are different facts with different fixes — see the
// `SocketProbe` section above for the oden measurement that forced this
// apart. `Path::exists()` reported the second as the first.
if let Some(probe) = probe {
let path = url.strip_prefix("unix://").unwrap_or(&url);
if let Some(reason) = socket_unreachable_reason(Path::new(path), &probe) {
return Err(Error::Backend(reason));
}
}
let docker = bollard::Docker::connect_with_unix(&url, 120, bollard::API_DEFAULT_VERSION)
.map_err(|e| Error::Backend(format!("connect container socket {url}: {e}")))?;
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.map_err(|e| Error::Backend(format!("build tokio runtime for bollard: {e}")))?;
Ok(Engine {
docker,
rt: rt.into(),
socket: url,
logs: Mutex::new(std::collections::HashMap::new()),
reaps: Mutex::new(Vec::new()),
})
}
/// **Pure** builder of the `ContainerCreateBody` for `image` — factored out so
/// the env/cmd/exposed-port wiring is testable with no daemon. A non-empty `cmd`
/// overrides the image entrypoint; `env` is carried as `KEY=VALUE`; each port is
/// both exposed AND published to the same host port via a `HostConfig` binding.
/// Byte-for-byte parity with jera's former `BollardEngine::create_body`, so the
/// one engine here produces the same container jera did.
// Kept as the documented jera-parity entry (bind-free); the engine now routes
// through `create_body_with_binds`, so this is a thin delegate + public API.
#[allow(dead_code)]
pub fn create_body(
image: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
) -> bollard::models::ContainerCreateBody {
Self::create_body_with_binds(image, env, cmd, ports, &[])
}
/// Like [`create_body`](Self::create_body) but also attaches host **bind
/// mounts** — `host:container[:opts]` strings (`-v` semantics) — to the
/// `HostConfig.binds`. With `binds` empty this is **byte-for-byte identical**
/// to [`create_body`](Self::create_body) (the mount field stays `None`), so
/// every existing caller is unchanged; a non-empty `binds` lets a build-in-a-
/// container job (WiX/MSI, pack) route through this one OCI engine instead of a
/// `Command::new("podman") -v …` shell twin.
pub fn create_body_with_binds(
image: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
binds: &[String],
) -> bollard::models::ContainerCreateBody {
Self::create_body_with_binds_and_net(image, env, cmd, ports, binds, None)
}
/// Like [`create_body_with_binds`](Self::create_body_with_binds) but also sets
/// the container **network mode** on `HostConfig.network_mode` — `Some("none")`
/// is the airgap `--network none` (loopback-only, no egress). With `network_mode`
/// `None` this is **byte-for-byte identical** to `create_body_with_binds` (the
/// field stays unset, and no `HostConfig` is minted when ports+binds are empty
/// too), so every existing caller is unchanged. This is the **load-bearing wire**
/// for Skidbladnir's airgap container route: jera renders `--network none` and
/// threads its `oci_value()` (`"none"`) here so it reaches the live podman run.
pub fn create_body_with_binds_and_net(
image: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
binds: &[String],
network_mode: Option<&str>,
) -> bollard::models::ContainerCreateBody {
// The `ports`-only (host==container) form: no distinct maps, no res caps.
// Byte-identical to before — it now flows through the one pair-aware core.
Self::create_body_full(
image,
env,
cmd,
ports,
&[],
binds,
network_mode,
None,
None,
&[],
)
}
/// The **one** create-body builder every other `create_body*` entry routes
/// through: it publishes both the `host == container` single-port form
/// ([`ports`]) **and** the distinct [`PortMap`] `host:container` form
/// ([`port_maps`]), attaches host `binds`, sets the `network_mode`, and applies
/// the `cpus`/`mem_mb` resource caps — every knob optional and each defaulting
/// to the byte-identical no-op (empty ports/maps/binds + `None` net/caps ⇒ no
/// `HostConfig` is minted, exactly as a bare image spec always rendered).
///
/// A `PortMap { host, container }` renders `exposed_ports[container/tcp]` +
/// `port_bindings[container/tcp] = host` — i.e. podman `-p host:container`, so
/// a service on a fixed in-container port (FalkorDB's `6379`) is reachable on a
/// different host port (a per-zone offset `6380`/`6381`). This is the load-
/// bearing wire for korp's per-zone infra: the demo zone (`6379:6379`) already
/// worked via `ports`; test/prod (`6380:6379`, `6381:6379`) need this map.
///
/// [`ports`]: BootSpec::ports
/// [`port_maps`]: BootSpec::port_maps
#[allow(clippy::too_many_arguments)]
pub fn create_body_full(
image: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
port_maps: &[crate::PortMap],
binds: &[String],
network_mode: Option<&str>,
cpus: Option<f64>,
mem_mb: Option<u32>,
security_opt: &[String],
) -> bollard::models::ContainerCreateBody {
Self::create_body_process(
image,
env,
cmd,
ports,
port_maps,
binds,
network_mode,
cpus,
mem_mb,
security_opt,
&ProcessIdentity::default(),
)
}
/// [`create_body_full`](Self::create_body_full) plus the container process's
/// **identity**: [`ProcessIdentity::user`] → the create body's `User` (podman
/// `--user`) and [`ProcessIdentity::working_dir`] → `WorkingDir`
/// (`--workdir`). A default (both-`None`) `process` renders a body
/// **byte-for-byte identical** to `create_body_full`, which is why that entry
/// point can delegate here and every caller written before this existed is
/// unchanged.
///
/// These two are on the *container config*, not the `HostConfig`, so unlike
/// the caps they never mint anything: an unnamed user or workdir leaves the
/// field `None` and the image's own `USER`/`WORKDIR` in force.
#[allow(clippy::too_many_arguments)]
pub fn create_body_process(
image: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
port_maps: &[crate::PortMap],
binds: &[String],
network_mode: Option<&str>,
cpus: Option<f64>,
mem_mb: Option<u32>,
security_opt: &[String],
process: &ProcessIdentity,
) -> bollard::models::ContainerCreateBody {
use bollard::models::{ContainerCreateBody, HostConfig, PortBinding};
use std::collections::HashMap;
let mut exposed: Vec<String> = Vec::new();
let mut bindings: HashMap<String, Option<Vec<PortBinding>>> = HashMap::new();
// Helper: expose `container/tcp` and bind it to `host` on 0.0.0.0. Kept as
// a closure so the two forms (host==container `ports` and distinct
// `port_maps`) render identically — a `ports` entry `p` is just the
// `host == container == p` case, so `[6379]` and `PortMap::same(6379)`
// produce the same wire.
let mut publish = |host: u16, container: u16| {
let key = format!("{container}/tcp");
if !exposed.contains(&key) {
exposed.push(key.clone());
}
bindings.insert(
key,
Some(vec![PortBinding {
host_ip: Some("0.0.0.0".to_string()),
host_port: Some(host.to_string()),
}]),
);
};
for &p in ports {
publish(p, p);
}
for pm in port_maps {
publish(pm.host, pm.container);
}
// Resource caps: MiB → bytes, cores → nano-cpus. Filter a non-positive
// value so an unconstrained (`None`) or mistakenly-zero cap never mints a
// `HostConfig` and stays byte-identical to the cap-less create body.
let nano_cpus = cpus.filter(|c| *c > 0.0).map(|c| (c * 1e9) as i64);
let memory = mem_mb
.filter(|m| *m > 0)
.map(|m| i64::from(m) * 1024 * 1024);
// Mint a `HostConfig` when there is anything to carry — ports, binds, a
// network mode, OR a resource cap. When all are absent, `host_config`
// stays `None`, exactly as a bare image spec always rendered (byte-parity).
let host_config = if bindings.is_empty()
&& binds.is_empty()
&& network_mode.is_none()
&& nano_cpus.is_none()
&& memory.is_none()
&& security_opt.is_empty()
&& process.cap_add.is_empty()
&& process.devices.is_empty()
&& process.cap_drop.is_empty()
&& !process.read_only_rootfs
&& process.tmpfs.is_empty()
{
None
} else {
Some(HostConfig {
port_bindings: if bindings.is_empty() {
None
} else {
Some(bindings)
},
binds: if binds.is_empty() {
None
} else {
Some(binds.to_vec())
},
network_mode: network_mode.map(str::to_string),
nano_cpus,
memory,
// Verbatim, and `None` when empty: an empty `Vec` here would
// render `"SecurityOpt": []` and stop being byte-identical to a
// create body that never named one.
security_opt: if security_opt.is_empty() {
None
} else {
Some(security_opt.to_vec())
},
// `None` when empty, for the same byte-parity reason.
cap_add: if process.cap_add.is_empty() {
None
} else {
Some(process.cap_add.clone())
},
devices: if process.devices.is_empty() {
None
} else {
Some(
process
.devices
.iter()
.map(|d| bollard::models::DeviceMapping {
path_on_host: Some(d.clone()),
path_in_container: Some(d.clone()),
cgroup_permissions: Some("rwm".to_string()),
})
.collect(),
)
},
// the sealed container (a Runa `Sandbox`): each `None` when unnamed
cap_drop: if process.cap_drop.is_empty() {
None
} else {
Some(process.cap_drop.clone())
},
readonly_rootfs: process.read_only_rootfs.then_some(true),
tmpfs: if process.tmpfs.is_empty() {
None
} else {
Some(process.tmpfs.iter().cloned().collect())
},
..Default::default()
})
};
ContainerCreateBody {
image: Some(image.to_string()),
cmd: if cmd.is_empty() {
None
} else {
Some(cmd.to_vec())
},
env: if env.is_empty() {
None
} else {
Some(env.to_vec())
},
exposed_ports: if exposed.is_empty() {
None
} else {
Some(exposed)
},
// Verbatim, and `None` when unnamed — the image's own USER/WORKDIR
// then stand, and the body is byte-identical to one written before
// these two fields could be set at all.
user: process.user.clone(),
working_dir: process.working_dir.clone(),
host_config,
..Default::default()
}
}
/// Like [`create_body_with_binds_and_net`](Self::create_body_with_binds_and_net)
/// but also sets the container **resource caps** on `HostConfig`: `cpus` →
/// `nano_cpus` (podman `--cpus`, `n * 1e9`) and `mem_mb` → `memory` (podman
/// `--memory`, MiB → bytes). Both **default to `None` = unconstrained**, which is
/// the deliberate hot-infra default — a container with no `--cpus` sees **all**
/// host cores (FalkorDB's OpenMP pool, a Spark executor must never be throttled to
/// one core). With `cpus`+`mem_mb` both `None` this is **byte-for-byte identical**
/// to `create_body_with_binds_and_net` (no `HostConfig` is minted purely for an
/// unset cap), so every existing caller is unchanged. A non-positive cap is dropped
/// here (guarded at [`BootSpec::validate`]) rather than passed to the daemon.
// Now a thin `port_maps`-free delegate to `create_body_full`; kept as a public
// parity entry (and used by the resource-cap unit test), so it is dead in a
// non-test lib build — same treatment as the `create_body` parity delegate.
#[allow(dead_code)]
#[allow(clippy::too_many_arguments)]
pub fn create_body_with_res(
image: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
binds: &[String],
network_mode: Option<&str>,
cpus: Option<f64>,
mem_mb: Option<u32>,
) -> bollard::models::ContainerCreateBody {
// The `ports`-only (host==container) form with resource caps: no distinct
// maps. Delegates to the one pair-aware core with empty `port_maps`, so it
// is byte-identical to the former standalone implementation.
Self::create_body_full(
image,
env,
cmd,
ports,
&[],
binds,
network_mode,
cpus,
mem_mb,
&[],
)
}
/// Pull `image` if not present, then create + start it as a detached container
/// named `name` with `env` (`KEY=VALUE`), an optional `cmd` entrypoint override,
/// and published `ports`. A background follow task streams the container's logs
/// into a shared buffer this container's [`drain_logs`](Self::drain_logs) drains.
// The boot paths now route through the net-aware
// `create_and_start_with_binds_and_net` (so `spec.net` reaches the live run), so
// these two thin delegates are the kept net-less/bind-less parity entries — same
// treatment as the public `create_body`/`create_body_with_binds` delegate pair.
#[allow(dead_code)]
fn create_and_start(
&self,
image: &str,
name: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
) -> Result<()> {
self.create_and_start_with_binds(image, name, env, cmd, ports, &[])
}
/// [`create_and_start`](Self::create_and_start) plus host **bind mounts**
/// (`host:container[:opts]`, `-v` semantics). `binds` empty ⇒ identical to
/// `create_and_start`.
#[allow(dead_code)]
fn create_and_start_with_binds(
&self,
image: &str,
name: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
binds: &[String],
) -> Result<()> {
self.create_and_start_with_binds_and_net(
image,
name,
env,
cmd,
ports,
&[],
binds,
None,
None,
None,
&[],
)
}
/// [`create_and_start_with_binds`](Self::create_and_start_with_binds) plus the
/// container **network mode** (`Some("none")` ⇒ `--network none`, the airgap
/// case). `network_mode` `None` ⇒ byte-identical to `create_and_start_with_binds`
/// (the live create body is unchanged). This is where the net choice actually
/// reaches the live podman `create_container` call.
// The engine's create-body carries this many independent knobs (image/name/env/
// cmd/ports/binds/net); grouping them into a struct would just shadow the
// `ContainerCreateBody` fields, so the flat arg list is the honest shape here.
#[allow(clippy::too_many_arguments)]
fn create_and_start_with_binds_and_net(
&self,
image: &str,
name: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
port_maps: &[crate::PortMap],
binds: &[String],
network_mode: Option<&str>,
cpus: Option<f64>,
mem_mb: Option<u32>,
security_opt: &[String],
) -> Result<()> {
self.create_and_start_process(
image,
name,
env,
cmd,
ports,
port_maps,
binds,
network_mode,
cpus,
mem_mb,
security_opt,
&ProcessIdentity::default(),
)
}
/// [`create_and_start_with_binds_and_net`](Self::create_and_start_with_binds_and_net)
/// plus the container process's **identity** ([`ProcessIdentity`]: podman
/// `--user` / `--workdir`). A default `process` is byte-identical to that
/// call, which is why it delegates here. This is where a spec's
/// [`user`](crate::BootSpec::user) / [`working_dir`](crate::BootSpec::working_dir)
/// actually reach the live create.
#[allow(clippy::too_many_arguments)]
fn create_and_start_process(
&self,
image: &str,
name: &str,
env: &[String],
cmd: &[String],
ports: &[u16],
port_maps: &[crate::PortMap],
binds: &[String],
network_mode: Option<&str>,
cpus: Option<f64>,
mem_mb: Option<u32>,
security_opt: &[String],
process: &ProcessIdentity,
) -> Result<()> {
use bollard::query_parameters::{
CreateContainerOptionsBuilder, CreateImageOptionsBuilder, StartContainerOptions,
};
use futures::StreamExt;
// CRASH SAFETY, BEFORE the runtime is entered (a nested `block_on` would
// panic): the name is deterministic, so the previous run of this same job
// may still be holding it. A leftover that is not running is REAPED and
// recorded for the caller to announce; a LIVE holder is refused by name and
// never killed. This replaces a blind `remove_container(force)` whose result
// was thrown away — which both murdered a concurrent build without a word
// and, when it failed, let the create hit the daemon anyway and come back as
// a bare 500 four layers below the user.
self.ensure_name_free(name)?;
let docker = &self.docker;
self.rt.block_on(async {
// Pull the image if it is not already local — unless the spec says it
// must be local already (`--pull=never`): then a missing image is refused.
if process.never_pull && docker.inspect_image(image).await.is_err() {
return Err(Error::Backend(format!(
"image {image} is not local, and this container is never pulled (--pull=never)"
)));
}
if docker.inspect_image(image).await.is_err() {
let (repo, tag) = image.rsplit_once(':').unwrap_or((image, "latest"));
let opts = CreateImageOptionsBuilder::new()
.from_image(repo)
.tag(tag)
.build();
let mut pull = docker.create_image(Some(opts), None, None);
while let Some(item) = pull.next().await {
item.map_err(|e| Error::Backend(format!("pull image {image}: {e}")))?;
}
}
let body = Self::create_body_process(
image,
env,
cmd,
ports,
port_maps,
binds,
network_mode,
cpus,
mem_mb,
security_opt,
process,
);
docker
.create_container(
Some(CreateContainerOptionsBuilder::new().name(name).build()),
body,
)
.await
.map_err(|e| {
Error::Backend(Self::create_refusal(
name,
format!("create container {name}: {e}"),
&e.to_string(),
))
})?;
docker
.start_container(name, None::<StartContainerOptions>)
.await
.map_err(|e| Error::Backend(format!("start container {name}: {e}")))?;
Ok::<(), Error>(())
})?;
// Wire a background log-follow task into a shared buffer this container's
// `drain_logs` drains — the API-streamed equivalent of reader threads.
//
// ROOT-LAW #0 SANCTIONED EXCEPTION: this is `tokio::spawn`, but the task is
// pure blocking-IO — it parks on `stream.next().await` (a docker `/logs`
// follow socket) and only trims/pushes each line into a `Mutex<Vec<..>>`.
// No CPU fan-out, no compute. ROOT-LAW #0 forbids `tokio::spawn`-for-CPU
// and bare `std::thread::spawn`, NOT an IO task on the backend's own async
// runtime; a `gatling::background::Job` would be wrong here — it wants an
// OS thread blocking on the async stream, re-entering this very runtime.
// So it correctly stays an async task on `self.rt`.
let buf: LogBuf = Arc::new(Mutex::new(Vec::new()));
self.logs
.lock()
.unwrap()
.insert(name.to_string(), Arc::clone(&buf));
let docker = self.docker.clone();
let name_owned = name.to_string();
self.rt.spawn(async move {
use bollard::query_parameters::LogsOptionsBuilder;
let mut stream = docker.logs(
&name_owned,
Some(
LogsOptionsBuilder::new()
.follow(true)
.stdout(true)
.stderr(true)
.build(),
),
);
let mut join = LineJoin::default();
while let Some(item) = stream.next().await {
match item {
Ok(out) => {
let mut b = buf.lock().unwrap();
push_log_frame(&mut join, &mut b, &out);
}
Err(_) => break,
}
}
join.flush(&mut buf.lock().unwrap());
});
Ok(())
}
/// **The container's whole log in one request** (`follow=false`), the
/// authoritative answer behind [`ContainerControl::read_all_logs`].
///
/// Not the follow socket: that one does not end when the container does.
/// MEASURED on t14s 2026-09-12 — a run that waited for podman's
/// `follow=true` response to close sat 30 s on a container that had already
/// exited and still held only 748 of the 901 lines it printed. This request
/// returns what the daemon has and finishes.
///
/// Any transport error answers `None` rather than a short list: a partial
/// read handed back as the whole is the defect this exists to remove.
fn read_all_logs(&self, name: &str) -> Option<(Vec<String>, Vec<String>)> {
use bollard::query_parameters::LogsOptionsBuilder;
use futures::StreamExt;
let docker = self.docker.clone();
let name = name.to_string();
self.rt.block_on(async move {
let mut stream = docker.logs(
&name,
Some(
LogsOptionsBuilder::new()
.follow(false)
.stdout(true)
.stderr(true)
.build(),
),
);
let mut all: Vec<(bool, String)> = Vec::new();
let mut join = LineJoin::default();
while let Some(item) = stream.next().await {
match item {
Ok(out) => push_log_frame(&mut join, &mut all, &out),
Err(_) => return None,
}
}
join.flush(&mut all);
let mut out = Vec::new();
let mut err = Vec::new();
for (is_err, line) in all {
if is_err {
err.push(line);
} else {
out.push(line);
}
}
Some((out, err))
})
}
/// Build an OCI image from `context_dir` + `containerfile`, tagging it `tag`.
/// Tars the context in-memory ([`context_tar`]) and streams it to the daemon's
/// `/build`; a `BuildInfo` carrying an `error` fails the call. Returns `tag`. The
/// error-detail→fail + returned-tag decision lives in [`build_image_on`] (daemon-
/// free, mock-tested); this engine only supplies the live `/build` stream.
fn build_image(
&self,
context_dir: &Path,
containerfile: &str,
tag: &str,
source_date_epoch: Option<i64>,
) -> Result<String> {
build_image_on(self, context_dir, containerfile, tag, source_date_epoch)
}
/// Extract `container_path` from `image` to `host_dest`: create a throwaway
/// (un-started) container, download a tar of the path from the daemon, remove the
/// container, and unpack the tar into `host_dest`. The `podman create`+`cp` twin.
/// The download+create+force-remove is [`download_path_tar`](Self::download_path_tar);
/// the unpack + basename layout is [`extract_path_on`] (daemon-free, mock-tested).
fn extract_path(&self, image: &str, container_path: &str, host_dest: &Path) -> Result<()> {
extract_path_on(self, image, container_path, host_dest)
}
/// The container's exit-code-aware state (running / exited-with-code / gone) —
/// what [`ContainerControl::container_state`] surfaces. A 404 / transient inspect
/// error reads [`ContainerState::Gone`]; a live boot's next poll retries.
///
/// Read over [`daemon_get`] + [`parse_container_state`] rather than bollard's
/// `inspect_container`: podman answers `State.Status` in its OWN vocabulary
/// (`stopped`, `configured`, `stopping`), which bollard's generated enum rejects
/// — and a rejected inspect lands in the `Err` arm here, i.e. a container that
/// is plainly there would read [`ContainerState::Gone`].
fn container_state(&self, name: &str) -> ContainerState {
let socket = self.socket.clone();
let path = format!("/v1.41/containers/{name}/json");
self.rt
.block_on(async move { daemon_get(&socket, &path).await })
.and_then(|body| parse_container_state(&body))
// A 404 / transient inspect error is "no such container", as before.
.unwrap_or(ContainerState::Gone)
}
/// Drain the streamed log lines accumulated for `name` since the last drain,
/// as one **combined** ordered stream (stdout + stderr interleaved as emitted).
fn drain_logs(&self, name: &str) -> Vec<String> {
match self.logs.lock().unwrap().get(name) {
Some(buf) => std::mem::take(&mut *buf.lock().unwrap())
.into_iter()
.map(|(_, line)| line)
.collect(),
None => Vec::new(),
}
}
/// Drain the streamed log lines for `name`, **split** into `(stdout, stderr)` —
/// the shape jera's run-to-completion `ContainerOutcome` keeps apart. Order
/// within each stream is preserved. Empties the same buffer `drain_logs` reads.
fn drain_logs_split(&self, name: &str) -> (Vec<String>, Vec<String>) {
match self.logs.lock().unwrap().get(name) {
Some(buf) => {
let mut out = Vec::new();
let mut err = Vec::new();
for (is_err, line) in std::mem::take(&mut *buf.lock().unwrap()) {
if is_err {
err.push(line);
} else {
out.push(line);
}
}
(out, err)
}
None => (Vec::new(), Vec::new()),
}
}
/// Start a previously-created (stopped) container.
fn start(&self, name: &str) -> Result<()> {
use bollard::query_parameters::StartContainerOptions;
self.rt.block_on(async {
self.docker
.start_container(name, None::<StartContainerOptions>)
.await
.map_err(|e| Error::Backend(format!("start container {name}: {e}")))
})
}
/// Stop + remove the container (idempotent, best-effort).
fn stop(&self, name: &str) {
use bollard::query_parameters::{RemoveContainerOptionsBuilder, StopContainerOptions};
self.rt.block_on(async {
let _ = self
.docker
.stop_container(name, None::<StopContainerOptions>)
.await;
let _ = self
.docker
.remove_container(
name,
Some(RemoveContainerOptionsBuilder::new().force(true).build()),
)
.await;
});
self.logs.lock().unwrap().remove(name);
}
/// Every container whose name begins with `prefix`, over a raw
/// `GET /containers/json?all=true` ([`daemon_get`] + [`parse_container_list`])
/// — the live drive behind [`ContainerControl::list_containers`].
///
/// NOT bollard's `list_containers`: its generated `ContainerSummaryStateEnum` is
/// Docker's closed eight-word set, and one container in podman's `stopped`
/// state failed the deserialisation of the WHOLE list — see
/// [`container_state_from_daemon`] for the oden measurement.
///
/// `all(true)`, so exited and created containers come back too: a sweep that
/// saw only running containers would leave precisely the corpses it exists
/// to remove.
///
/// The name filter is applied **here** rather than handed to the daemon.
/// Docker's and podman's `name=` filter is a *substring* match, not a
/// prefix, so `name=foo-` also returns `bar-foo-1`; a janitor that removed
/// on that basis would delete another owner's containers. The daemon's
/// filter is a bandwidth optimisation with the wrong semantics, so this asks
/// for the list and matches the prefix itself.
///
/// Names come back from the daemon with a leading `/`, which is stripped.
///
/// ⚠ `prefix` matches the **daemon-side** name, and that is not the name in
/// the [`BootSpec`]: [`Machine::container_name`] prepends `draupnir-`, and
/// `machine.id` is the result. `list_containers(&spec.name)` matches nothing.
/// Pass `&machine.id`, or a `draupnir-…` prefix. Caught by this crate's own
/// live test on the first run.
///
/// [`BootSpec`]: crate::BootSpec
fn list_containers(&self, prefix: &str) -> Result<Vec<(String, ContainerState)>> {
let socket = self.socket.clone();
let body = self.rt.block_on(async move {
daemon_get(&socket, "/v1.41/containers/json?all=true").await
})?;
parse_container_list(&body, prefix)
}
/// The container's configured environment, over bollard's
/// `inspect_container` — the live drive behind
/// [`ContainerControl::container_env`].
///
/// The daemon's record of what the process was started with, not the spec it
/// was asked for. A container with no environment at all answers an empty
/// vector; a container that is not there is an [`Error::Backend`], because
/// "no variables" and "no container" are different answers and a read-back
/// that conflated them would pass on a container that had vanished.
fn container_env(&self, name: &str) -> Result<Vec<String>> {
use bollard::query_parameters::InspectContainerOptions;
self.rt.block_on(async {
let info = self
.docker
.inspect_container(name, None::<InspectContainerOptions>)
.await
.map_err(|e| Error::Backend(format!("inspecting container `{name}`: {e}")))?;
Ok(info
.config
.and_then(|c| c.env)
.unwrap_or_default()
.into_iter()
.collect())
})
}
/// **Exec `argv` inside the running container `name`** — the live drive behind
/// [`ContainerControl::exec`], over podman's `/exec` REST op (`create_exec` →
/// `start_exec` → `inspect_exec`), never a subprocess (zero-shell). Attaches
/// stdout+stderr, drains them split (same tagging as the log-follow task), and
/// reads the process's exit code back from `inspect_exec`. A non-zero exit is
/// surfaced in [`ExecOutcome::exit_code`], not as an `Err`; only a create/start/
/// inspect/transport failure returns `Err`.
fn exec(&self, name: &str, argv: &[String]) -> Result<ExecOutcome> {
use bollard::container::LogOutput;
use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults};
use futures::StreamExt;
let docker = &self.docker;
self.rt.block_on(async {
let config = CreateExecOptions::<String> {
cmd: Some(argv.to_vec()),
attach_stdout: Some(true),
attach_stderr: Some(true),
..Default::default()
};
let created = docker
.create_exec(name, config)
.await
.map_err(|e| Error::Backend(format!("create exec in {name}: {e}")))?;
let mut stdout: Vec<String> = Vec::new();
let mut stderr: Vec<String> = Vec::new();
match docker
.start_exec(&created.id, None::<StartExecOptions>)
.await
.map_err(|e| Error::Backend(format!("start exec in {name}: {e}")))?
{
StartExecResults::Attached { mut output, .. } => {
while let Some(item) = output.next().await {
match item {
Ok(out) => {
let is_err = matches!(out, LogOutput::StdErr { .. });
let line = LogOutput::to_string(&out);
let line = line.trim_end_matches(['\n', '\r']).to_string();
if !line.is_empty() {
if is_err {
stderr.push(line);
} else {
stdout.push(line);
}
}
}
Err(e) => {
return Err(Error::Backend(format!(
"read exec output in {name}: {e}"
)))
}
}
}
}
StartExecResults::Detached => {}
}
let inspect = docker
.inspect_exec(&created.id)
.await
.map_err(|e| Error::Backend(format!("inspect exec in {name}: {e}")))?;
Ok(ExecOutcome {
exit_code: inspect.exit_code,
stdout,
stderr,
})
})
}
/// **Exec one [`ExecSpec`] inside the running container `name`, streaming** —
/// the live drive behind [`ContainerControl::exec_streaming`]: `create_exec`
/// with the spec's env / user / working dir, `start_exec` attached, every
/// stdout/stderr chunk handed to `sink` AS BYTES (the multiplexed attach
/// stream carries the process's pipes; joined, the chunks are exactly what
/// it wrote), then `inspect_exec` for the code.
///
/// The attach stream can end a moment before the daemon has recorded the
/// exit (podman reports `Running` with no code for a few ms), so the code is
/// read until the exec is no longer running — 10 s at most, then whatever
/// the daemon has.
fn exec_streaming(
&self,
name: &str,
spec: &ExecSpec,
sink: &mut dyn FnMut(LogStream, &[u8]),
) -> Result<ExecRun> {
use bollard::container::LogOutput;
use bollard::exec::{CreateExecOptions, StartExecOptions, StartExecResults};
use futures::StreamExt;
let docker = &self.docker;
let mut run = ExecRun::default();
let exec_id = self.rt.block_on(async {
let env: Vec<String> = spec.env.iter().map(|(k, v)| format!("{k}={v}")).collect();
let config = CreateExecOptions::<String> {
cmd: Some(spec.argv.clone()),
env: if env.is_empty() { None } else { Some(env) },
user: spec.user.clone(),
working_dir: spec.working_dir.clone(),
attach_stdout: Some(true),
attach_stderr: Some(true),
..Default::default()
};
let created = docker
.create_exec(name, config)
.await
.map_err(|e| Error::Backend(format!("create exec in {name}: {e}")))?;
match docker
.start_exec(&created.id, None::<StartExecOptions>)
.await
.map_err(|e| Error::Backend(format!("start exec in {name}: {e}")))?
{
StartExecResults::Attached { mut output, .. } => {
while let Some(item) = output.next().await {
match item {
Ok(LogOutput::StdErr { message }) => {
sink(LogStream::Stderr, &message);
run.stderr.extend_from_slice(&message);
}
Ok(LogOutput::StdOut { message } | LogOutput::Console { message }) => {
sink(LogStream::Stdout, &message);
run.stdout.extend_from_slice(&message);
}
Ok(LogOutput::StdIn { .. }) => {}
Err(e) => {
return Err(Error::Backend(format!(
"read exec output in {name}: {e}"
)))
}
}
}
}
StartExecResults::Detached => {}
}
Ok::<String, Error>(created.id)
})?;
let deadline = Instant::now() + Duration::from_secs(10);
let mut pause = Duration::from_millis(5);
loop {
let inspect = self
.rt
.block_on(docker.inspect_exec(&exec_id))
.map_err(|e| Error::Backend(format!("inspect exec in {name}: {e}")))?;
let running = inspect.running.unwrap_or(false);
if !running || Instant::now() >= deadline {
run.exit_code = inspect.exit_code;
return Ok(run);
}
std::thread::sleep(pause);
pause = (pause * 2).min(Duration::from_millis(200));
}
}
/// The container's power state: `On` while running, else `Off` (a gone/unknown
/// container reads `Off`).
fn power_state(&self, name: &str) -> PowerState {
use bollard::query_parameters::InspectContainerOptions;
self.rt.block_on(async {
match self
.docker
.inspect_container(name, None::<InspectContainerOptions>)
.await
{
Ok(info) => {
let running = info.state.as_ref().and_then(|s| s.running).unwrap_or(false);
if running {
PowerState::On
} else {
PowerState::Off
}
}
Err(_) => PowerState::Off,
}
})
}
/// Whether `needle` has appeared in the container's stdout/stderr logs so far —
/// the log-match readiness probe. Drains the (non-follow) log stream once and
/// scans it; a container that has produced no output yet simply reads `false`.
fn log_contains(&self, name: &str, needle: &str) -> Result<bool> {
use bollard::query_parameters::LogsOptionsBuilder;
use futures::StreamExt;
The file goes on; the console shows all of it.