Watch
0
0
Fork
You've already forked hyperhive
0
hyperhive/hive-priv/src/container.rs
atlas 6b800600e2 bind a container apply to the closure realised for the agent, and record the deployed pins on drift
hive-priv applies only the path its last RealiseClosure returned for that agent. On drift the swap and create record the pins of the commit the closure was built from.
2026-10-29 22:36:58 +01:00

637 lines
26 KiB
Rust

//! Container lifecycle ops: `StartContainer`, `StopContainer`,
//! `KillContainer`, `CreateContainer`, `UpdateContainer`, `DestroyContainer`
//! and `ListContainers`. Every op takes a logical agent name and reaches only
//! the `h-`-prefixed container [`container_system_name`] derives from it.
//!
//! `CreateContainer` / `UpdateContainer` apply a closure already in the store
//! and stream progress to the caller as [`PrivEvent::Line`]s when asked to.
use anyhow::{Context as _, Result, bail};
use hive_priv_sock::{PrivEvent, PrivStream, PrivStreamLine};
use tokio::io::{AsyncWriteExt, BufReader};
use tokio::net::unix::OwnedWriteHalf;
use tokio::process::Command;
use crate::shared::{container_system_name, validate_agent_name};
/// Write one `PrivEvent::Line` to the client. Best-effort: a write
/// failure is logged but doesn't abort the running subprocess.
async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: &str) {
let event = PrivEvent::Line(PrivStreamLine {
stream,
data: data.to_owned(),
});
if let Ok(mut json) = serde_json::to_string(&event) {
json.push('\n');
if let Err(e) = writer.write_all(json.as_bytes()).await {
tracing::warn!(error = %e, "write_line_event: write failed");
}
}
}
/// `StartContainer`: clear any start-limit lockout left by earlier failures
/// so a now-correct start isn't blocked (`nixos-container start` does not do
/// this itself). Best-effort — if the unit doesn't exist yet (first-time
/// create), `reset-failed` is a no-op and the start proceeds regardless.
pub(crate) async fn start_container(name: &str) -> Result<(String, String)> {
validate_agent_name(name)?;
let machine = container_system_name(name);
// Silenced: a first-time create means the unit doesn't exist yet, and
// `systemctl`'s own "Unit ... not loaded" complaint for that case would
// otherwise land in the journal at error priority on every such start —
// routine, not a signal, since the result is already discarded below.
let _ = Command::new("systemctl")
.args(["reset-failed", &format!("container@{machine}.service")])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.await;
container_run(&["start", &machine]).await
}
/// `StopContainer`.
pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> {
validate_agent_name(name)?;
stop_and_release(&container_system_name(name)).await
}
/// `KillContainer`: `nixos-container` has no kill verb, so this uses
/// `machinectl` to send `SIGKILL` to every process in the container — the
/// right semantics for a forced shutdown after a graceful stop has already
/// been attempted.
pub(crate) async fn kill_container(name: &str) -> Result<(String, String)> {
validate_agent_name(name)?;
let machine = container_system_name(name);
machinectl_run(&["kill", &machine, "--signal=SIGKILL"]).await
}
/// `DestroyContainer`.
pub(crate) async fn destroy_container(name: &str) -> Result<(String, String)> {
validate_agent_name(name)?;
container_run(&["destroy", &container_system_name(name)]).await
}
/// Shared body for `CreateContainer` / `UpdateContainer`: validate the name
/// and that `toplevel` is the closure realised for the agent, and apply it (streaming line events to `writer` when
/// `stream` is set). Nothing is built here: `toplevel` is a closure already in
/// the store, which `RealiseClosure` put there.
///
/// `create` passes `--system-path` so `nixos-container`'s own `buildFlake()`
/// never runs; it builds to a relative `.tmp` out-link that concurrent calls
/// share, which handed an agent another agent's closure.
///
/// `update` does not call `nixos-container update` at all: that action's
/// version-compat probe runs before `--system-path` is honored and dies on every
/// agent. [`swap_container_profile`] does what the action does past that probe:
/// `nix-env --set` the per-container profile, then `systemctl reload` if the
/// container is running.
pub(crate) async fn container_toplevel_action(
verb: &str,
name: &str,
toplevel: &str,
stream: bool,
writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
validate_agent_name(name)?;
let toplevel = crate::closure::applicable_for(name, toplevel)?;
let system_name = container_system_name(name);
if verb == "update" {
return swap_container_profile(&system_name, toplevel, stream, writer).await;
}
let args = [verb, &system_name, "--system-path", toplevel];
if stream {
container_run_streaming(&args, writer).await
} else {
container_run(&args).await
}
}
/// Apply a prebuilt toplevel to an existing container directly, without
/// going through `nixos-container update` (see `container_toplevel_action`'s
/// doc comment for why). Mirrors `nixos-container.pl`'s own `update`
/// action verbatim, past its version probe: point the per-container `nix-
/// env` profile at the new toplevel, then reload the container unit *if*
/// it's currently running — the container's own next start already reads
/// from the profile, so a stopped container needs nothing further.
///
/// Both steps are near-instant in the happy case, but still stream-
/// forwarded like the rest of this operation: a failure here (e.g. a
/// wedged `systemctl reload`) is exactly when the caller most wants the
/// live line in the build log, not just a summary error afterward.
async fn swap_container_profile(
system_name: &str,
toplevel: &str,
stream: bool,
writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
let profile = format!("/nix/var/nix/profiles/per-container/{system_name}/system");
let set_out = Command::new("nix-env")
.args(["-p", &profile, "--set", toplevel])
.output()
.await
.context("invoke nix-env --set")?;
let mut stdout = String::from_utf8_lossy(&set_out.stdout).into_owned();
let mut stderr = String::from_utf8_lossy(&set_out.stderr).into_owned();
log_and_forward(&stdout, &stderr, stream, writer).await;
if !set_out.status.success() {
bail!(
"nix-env -p {profile} --set failed ({}): {}",
set_out.status,
stderr.trim()
);
}
let unit = format!("container@{system_name}");
let state_out = Command::new("systemctl")
.args(["show", "--property=ActiveState", "--value", &unit])
.output()
.await
.context("query container ActiveState")?;
let active = String::from_utf8_lossy(&state_out.stdout).trim() == "active";
if active {
let reload_out = Command::new("systemctl")
.args(["reload", &unit])
.output()
.await
.context("invoke systemctl reload")?;
let r_stdout = String::from_utf8_lossy(&reload_out.stdout).into_owned();
let r_stderr = String::from_utf8_lossy(&reload_out.stderr).into_owned();
log_and_forward(&r_stdout, &r_stderr, stream, writer).await;
if !reload_out.status.success() {
bail!(
"systemctl reload {unit} failed ({}): {}",
reload_out.status,
r_stderr.trim()
);
}
stdout.push_str(&r_stdout);
stderr.push_str(&r_stderr);
}
Ok((stdout, stderr))
}
/// Log a command's captured stdout/stderr the same way every other
/// `nixos-container`-adjacent shellout in this file does, and — when
/// `stream` is set — forward each line to the client as a live
/// [`PrivEvent::Line`] too, so a caller watching the build log sees these
/// lines exactly like any other step's, not just a summary on failure.
async fn log_and_forward(stdout: &str, stderr: &str, stream: bool, writer: &mut OwnedWriteHalf) {
for line in stdout.lines() {
tracing::info!(target: "nixos-container", "{line}");
if stream {
write_line_event(writer, PrivStream::Stdout, line).await;
}
}
for line in stderr.lines() {
tracing::warn!(target: "nixos-container", "{line}");
if stream {
write_line_event(writer, PrivStream::Stderr, line).await;
}
}
}
/// How a [`run_bounded`] child ended.
pub(crate) enum BoundedRun {
Exited {
status: std::process::ExitStatus,
stderr: String,
},
/// Still running at the limit; its whole process group was killed.
TimedOut,
}
/// Run `cmd` to completion or until `limit`, draining both pipes: stdout is
/// discarded, stderr is kept and also logged as it arrives.
///
/// The child leads its own process group and a timeout `SIGKILL`s the whole
/// group, so a grandchild still holding the pipes (or the daemon connection)
/// dies with it instead of outliving the error.
pub(crate) async fn run_bounded(
mut cmd: Command,
limit: std::time::Duration,
) -> std::io::Result<BoundedRun> {
use tokio::io::AsyncBufReadExt as _;
let mut child = cmd
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.process_group(0)
.spawn()?;
// Group leader, so its pid is the pgid.
let pgid = child.id().and_then(|pid| i32::try_from(pid).ok());
let stdout = child.stdout.take().expect("stdout piped");
let stderr = child.stderr.take().expect("stderr piped");
let run = async {
let mut stdout_lines = BufReader::new(stdout).lines();
let mut stderr_lines = BufReader::new(stderr).lines();
let mut stderr_buf = String::new();
// ⚠️ Both pipes are drained concurrently: reading one alone would let
// the other fill its pipe buffer and deadlock the child.
//
// ⚠️ Each stream's EOF is tracked separately rather than breaking on
// the first `None`: `next_line()` on a closed stream returns
// `Ok(None)` immediately and forever, so a loop that keeps polling a
// finished stream spins hot until the other one ends too.
let mut stdout_done = false;
let mut stderr_done = false;
while !(stdout_done && stderr_done) {
tokio::select! {
line = stdout_lines.next_line(), if !stdout_done => {
match line {
Ok(Some(_)) => {}
Ok(None) => stdout_done = true,
Err(e) => {
tracing::warn!(error = %e, "child stdout read error");
stdout_done = true;
}
}
}
line = stderr_lines.next_line(), if !stderr_done => {
match line {
Ok(Some(l)) => {
tracing::info!(target: "nix-store", "{l}");
if !stderr_buf.is_empty() {
stderr_buf.push('\n');
}
stderr_buf.push_str(&l);
}
Ok(None) => stderr_done = true,
Err(e) => {
tracing::warn!(error = %e, "child stderr read error");
stderr_done = true;
}
}
}
}
}
let status = child.wait().await?;
Ok::<_, std::io::Error>(BoundedRun::Exited {
status,
stderr: stderr_buf,
})
};
if let Ok(exited) = tokio::time::timeout(limit, run).await {
return exited;
}
if let Some(pgid) = pgid {
// SAFETY: plain `kill(2)`; `-pgid` targets the child's own process
// group, which `process_group(0)` made it lead.
unsafe {
libc::kill(-pgid, libc::SIGKILL);
}
}
child.wait().await?;
Ok(BoundedRun::TimedOut)
}
/// Invoke `nixos-container` with the given args, log output to journald.
pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> {
let out = Command::new("nixos-container")
.args(args)
.output()
.await
.context("invoke nixos-container")?;
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
// `list` is a read-only enumeration called on the hot path (dashboard
// rescan, forge + boot sweeps) — its stdout is the return value, not
// progress, so logging every container name on every call floods the
// journal. Log stdout only for the mutating ops, where each line is
// genuine progress. stderr is always logged (errors matter regardless).
if args.first() != Some(&"list") {
for line in stdout.lines() {
tracing::info!(target: "nixos-container", "{line}");
}
}
for line in stderr.lines() {
tracing::warn!(target: "nixos-container", "{line}");
}
if !out.status.success() {
bail!(
"nixos-container {} failed ({}): {}",
args.join(" "),
out.status,
stderr.trim()
);
}
Ok((stdout, stderr))
}
/// Invoke `machinectl` with the given args, log output to journald.
/// Used for operations that nixos-container doesn't expose (e.g. sending
/// signals to running containers).
async fn machinectl_run(args: &[&str]) -> Result<(String, String)> {
let out = Command::new("machinectl")
.args(args)
.output()
.await
.context("invoke machinectl")?;
let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
for line in stdout.lines() {
tracing::info!(target: "machinectl", "{line}");
}
for line in stderr.lines() {
tracing::warn!(target: "machinectl", "{line}");
}
if !out.status.success() {
bail!(
"machinectl {} failed ({}): {}",
args.join(" "),
out.status,
stderr.trim()
);
}
Ok((stdout, stderr))
}
/// How long to wait for machined to drop a machine's registration after a
/// shutdown has been asked for. Generous: a container with slow-stopping
/// units legitimately takes a while, and escalating early would SIGKILL a
/// shutdown that was going to finish on its own.
const NAME_RELEASE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(20);
/// Poll cadence while waiting for the registration to go away.
const NAME_RELEASE_POLL: std::time::Duration = std::time::Duration::from_millis(500);
/// Cap on a single registration probe. The probe is a D-Bus round trip to
/// machined; if machined itself is wedged the call would otherwise sit on
/// the D-Bus method timeout, which is far longer than the whole stop
/// sequence should take.
const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
/// True when machined still holds a registration for `machine`.
///
/// This asks machined the same question the registration itself answers —
/// `machinectl show` resolves the name through `GetMachine`, the very lookup
/// that makes a later boot fail with `Failed to register machine: already
/// exists`. So a positive answer here is exactly the condition that breaks
/// the next start, not a proxy for it.
///
/// Deliberately NOT the container's systemd unit state: the unit can be
/// `inactive` while the registration is still held, and that gap is the
/// whole bug this probe exists to catch.
///
/// Errors and timeouts answer "still registered". Being wrong that way costs
/// a redundant SIGKILL to something already gone; being wrong the other way
/// hands back a stop that silently leaked the name.
async fn machine_registered(machine: &str) -> bool {
let probe = Command::new("machinectl")
.args(["show", machine, "--property=Name"])
// Don't leave a probe behind when the timeout below fires.
.kill_on_drop(true)
.output();
match tokio::time::timeout(MACHINE_PROBE_TIMEOUT, probe).await {
Ok(Ok(out)) => out.status.success(),
Ok(Err(e)) => {
tracing::warn!(%machine, error = %e, "machinectl show failed to run; assuming still registered");
true
}
Err(_) => {
tracing::warn!(%machine, "machinectl show timed out; assuming still registered");
true
}
}
}
/// Poll [`machine_registered`] until the name is free or the timeout expires.
/// Returns true once the registration is gone.
async fn wait_name_released(machine: &str) -> bool {
let deadline = tokio::time::Instant::now() + NAME_RELEASE_TIMEOUT;
loop {
if !machine_registered(machine).await {
return true;
}
if tokio::time::Instant::now() >= deadline {
return false;
}
tokio::time::sleep(NAME_RELEASE_POLL).await;
}
}
/// Stop `machine` and only report success once machined has actually released
/// the name.
///
/// `nixos-container stop` exiting 0 does not mean the machine is gone. A
/// process that sits in the machine's cgroup without being a child of the
/// container's init — a shell exec'd in from outside, say — never receives
/// the shutdown's SIGTERM if it has been stopped with SIGSTOP, so the
/// registration outlives the "successful" stop. Every later start of that
/// container then fails with `Failed to register machine: already exists`,
/// and machined re-persists the stale record across its own restart, so
/// there is no cleaning it up after the fact. The only place to catch it is
/// here, in the stop.
///
/// So: ask for the stop, wait for the name, and fail loudly if it is still
/// held — a caller that is told the stop worked will go on to start the
/// container and hit the confusing registration error instead of this one.
///
/// This used to escalate to `machinectl kill --signal=SIGKILL` here. Dropped
/// per real incident data: the case this guards is a container genuinely
/// wedged (e.g. a root-login process survived the stop), and SIGKILL
/// doesn't recover that in practice — only a host-level reboot has.
/// Pretending a kill attempt handled it hides a condition that needs a human
/// to look at the host, so this now just reports the failure instead of
/// quietly (and ineffectually) trying to force it.
///
/// The verify-then-fail lives in the helper rather than at a call site so
/// that every stop gets it: dashboard, reconcile, destroy, cold-start
/// fallback. The start path already distrusts its own exit code the same way;
/// this is the missing half of that pair.
async fn stop_and_release(machine: &str) -> Result<(String, String)> {
let stop = container_run(&["stop", machine]).await;
if wait_name_released(machine).await {
// Happy path, and also the path where a stop that reported failure
// nonetheless brought the machine down. Either way the caller gets
// the original result untouched.
return stop;
}
tracing::error!(
%machine,
"stop finished but machined still holds the registration — the \
container is likely wedged (a process outside its init tree \
survived the stop); this needs a host-level look, not another \
stop attempt"
);
bail!(
"stop {machine}: machined still holds the machine name after {}s. \
This container is likely wedged and starting it again will fail to \
register — needs host-level intervention (a reboot has been the \
only reliable fix in practice).",
NAME_RELEASE_TIMEOUT.as_secs()
)
}
/// Invoke `nixos-container` with the given args and forward output lines
/// to the caller as `PrivEvent::Line` messages in real time, logging each
/// line to journald as it arrives. Returns `(String::new(), String::new())`
/// on success (all output was streamed); the error string includes stderr
/// tail on failure.
async fn container_run_streaming(
args: &[&str],
writer: &mut OwnedWriteHalf,
) -> Result<(String, String)> {
use tokio::io::AsyncBufReadExt as _;
use tokio::process::Command;
let mut child = Command::new("nixos-container")
.args(args)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.context("invoke nixos-container (streaming)")?;
let stdout = child.stdout.take().expect("stdout piped");
let stderr = child.stderr.take().expect("stderr piped");
let mut stdout_lines = BufReader::new(stdout).lines();
let mut stderr_lines = BufReader::new(stderr).lines();
// Collect stderr for the error message; stream both to the client.
let mut stderr_buf = String::new();
// Drive stdout and stderr concurrently. `tokio::select!` interleaves
// them without bias — both streams drain at roughly the same rate
// as the subprocess produces output.
loop {
tokio::select! {
line = stdout_lines.next_line() => {
match line {
Ok(Some(l)) => {
tracing::info!(target: "nixos-container", "{l}");
write_line_event(writer, PrivStream::Stdout, &l).await;
}
Ok(None) => break,
Err(e) => {
tracing::warn!(error = %e, "nixos-container stdout read error");
break;
}
}
}
line = stderr_lines.next_line() => {
match line {
Ok(Some(l)) => {
tracing::warn!(target: "nixos-container", "{l}");
write_line_event(writer, PrivStream::Stderr, &l).await;
if !stderr_buf.is_empty() {
stderr_buf.push('\n');
}
stderr_buf.push_str(&l);
}
Ok(None) => {}
Err(e) => {
tracing::warn!(error = %e, "nixos-container stderr read error");
}
}
}
}
}
// Drain any remaining stderr after stdout closed.
while let Ok(Some(l)) = stderr_lines.next_line().await {
tracing::warn!(target: "nixos-container", "{l}");
write_line_event(writer, PrivStream::Stderr, &l).await;
if !stderr_buf.is_empty() {
stderr_buf.push('\n');
}
stderr_buf.push_str(&l);
}
let status = child.wait().await.context("wait nixos-container")?;
if !status.success() {
// Only the last stderr line is embedded — the full stderr was
// already forwarded line-by-line as PrivEvent::Line messages and
// is captured in build_logs.sqlite by the caller. Keeping the
// error message short avoids bloating the anyhow chain.
bail!(
"nixos-container {} failed ({}): {}",
args.join(" "),
status,
stderr_buf.lines().last().unwrap_or("").trim()
);
}
Ok((String::new(), String::new()))
}
#[cfg(test)]
mod tests {
use super::{BoundedRun, run_bounded};
use crate::shared::test_support::scratch;
/// Whether `pid` is gone, a zombie (`Z`), or being reaped (`X`): none of
/// these runs again, and a process passes through them in that order,
/// so once this is true it stays true.
fn is_dead(pid: &str) -> bool {
match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
Err(_) => true,
Ok(stat) => stat_is_dead(&stat),
}
}
/// The state letter of a `/proc/<pid>/stat` line, which follows the
/// parenthesised command name, is `Z` or `X`.
fn stat_is_dead(stat: &str) -> bool {
stat.rsplit_once(") ")
.is_some_and(|(_, rest)| rest.starts_with(['Z', 'X']))
}
#[test]
fn zombie_and_reaped_states_count_as_dead() {
assert!(stat_is_dead("42 (sleep) Z 1 42 42"));
assert!(stat_is_dead("42 (sleep) X 1 42 42"));
assert!(!stat_is_dead("42 (sleep) S 1 42 42"));
assert!(!stat_is_dead("42 (sleep) R 1 42 42"));
}
/// A child still running at the limit comes back `TimedOut`, and the kill
/// reaches its whole process group: the backgrounded grandchild dies too,
/// not just the shell that started it.
#[tokio::test]
async fn a_child_past_its_limit_is_killed_with_its_group() {
use std::time::{Duration, Instant};
let dir = scratch();
let pidfile = dir.join("grandchild");
let mut cmd = tokio::process::Command::new("sh");
cmd.arg("-c")
.arg(format!("sleep 600 & echo $! > {}; wait", pidfile.display()));
let started = Instant::now();
let run = run_bounded(cmd, Duration::from_secs(2)).await.unwrap();
assert!(matches!(run, BoundedRun::TimedOut), "must time out");
assert!(
started.elapsed() < Duration::from_mins(1),
"the limit did not bound the run"
);
let pid = std::fs::read_to_string(&pidfile).unwrap();
let pid = pid.trim();
let deadline = Instant::now() + Duration::from_secs(5);
while !is_dead(pid) && Instant::now() < deadline {
tokio::time::sleep(Duration::from_millis(50)).await;
}
assert!(is_dead(pid), "grandchild {pid} outlived the timeout");
std::fs::remove_dir_all(&dir).ok();
}
/// The control: a child that exits inside its limit is `Exited`, with its
/// status and its stderr.
#[tokio::test]
async fn a_child_inside_its_limit_returns_its_status_and_stderr() {
let mut cmd = tokio::process::Command::new("sh");
cmd.arg("-c")
.arg("echo progress >&2; echo /nix/store/abc-toplevel");
match run_bounded(cmd, std::time::Duration::from_mins(1))
.await
.unwrap()
{
BoundedRun::Exited { status, stderr } => {
assert!(status.success());
assert_eq!(stderr, "progress");
}
BoundedRun::TimedOut => panic!("a fast child must not time out"),
}
}
}