diff --git a/hive-priv/src/container.rs b/hive-priv/src/container.rs index 037e3823..f5050efc 100644 --- a/hive-priv/src/container.rs +++ b/hive-priv/src/container.rs @@ -14,6 +14,45 @@ use tokio::process::Command; use crate::shared::{container_system_name, validate_agent_name}; +/// The program `program` names, ready for arguments. The one place the real +/// tools are resolved, so a test can substitute stand-ins. +fn real_command(program: &str) -> Command { + Command::new(program) +} + +/// What the lifecycle ops run and how long they wait for it: the host's +/// tools and timings in [`Host::REAL`], stand-ins in tests. +struct Host<'a> { + spawn: &'a (dyn Fn(&str) -> Command + Sync), + /// 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. + release_timeout: std::time::Duration, + /// Poll cadence while waiting for the registration to go away. + release_poll: std::time::Duration, + /// 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. + probe_timeout: std::time::Duration, +} + +impl Host<'static> { + const REAL: Self = Host { + spawn: &real_command, + release_timeout: std::time::Duration::from_secs(20), + release_poll: std::time::Duration::from_millis(500), + probe_timeout: std::time::Duration::from_secs(5), + }; +} + +impl Host<'_> { + fn command(&self, program: &str) -> Command { + (self.spawn)(program) + } +} + /// 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) { @@ -34,19 +73,24 @@ async fn write_line_event(writer: &mut OwnedWriteHalf, stream: PrivStream, data: /// 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)> { + start_container_with(&Host::REAL, name).await +} + +async fn start_container_with(host: &Host<'_>, 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") + let _ = host + .command("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 + container_run_with(host, &["start", &machine]).await } /// Most bytes of journal text `StartFailureLog` returns. @@ -106,27 +150,41 @@ pub(crate) async fn start_failure_log(name: &str, since_unix: u64) -> Result<(St let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| d.as_secs()); - start_failure_log_with(name, since_unix, now, |argv| async move { - let child = Command::new("journalctl") - .args(&argv) - .kill_on_drop(true) - .output(); - let out = tokio::time::timeout(START_LOG_TIMEOUT, child) - .await - .context("journalctl timed out")? - .context("invoke journalctl")?; - if !out.status.success() { - bail!("journalctl failed ({})", out.status); - } - Ok(String::from_utf8_lossy(&out.stdout).into_owned()) + start_failure_log_with(name, since_unix, now, |argv| { + run_journalctl(&Host::REAL, START_LOG_TIMEOUT, argv) }) .await } +/// `journalctl` with `argv`: its stdout, if it exits 0 within `limit`. +async fn run_journalctl( + host: &Host<'_>, + limit: std::time::Duration, + argv: Vec, +) -> Result { + let child = host + .command("journalctl") + .args(&argv) + .kill_on_drop(true) + .output(); + let out = tokio::time::timeout(limit, child) + .await + .context("journalctl timed out")? + .context("invoke journalctl")?; + if !out.status.success() { + bail!("journalctl failed ({})", out.status); + } + Ok(String::from_utf8_lossy(&out.stdout).into_owned()) +} + /// `StopContainer`. pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> { + stop_container_with(&Host::REAL, name).await +} + +async fn stop_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> { validate_agent_name(name)?; - stop_and_release(&container_system_name(name)).await + stop_and_release(host, &container_system_name(name)).await } /// `KillContainer`: `nixos-container` has no kill verb, so this uses @@ -134,15 +192,23 @@ pub(crate) async fn stop_container(name: &str) -> Result<(String, String)> { /// 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)> { + kill_container_with(&Host::REAL, name).await +} + +async fn kill_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> { validate_agent_name(name)?; let machine = container_system_name(name); - machinectl_run(&["kill", &machine, "--signal=SIGKILL"]).await + machinectl_run(host, &["kill", &machine, "--signal=SIGKILL"]).await } /// `DestroyContainer`. pub(crate) async fn destroy_container(name: &str) -> Result<(String, String)> { + destroy_container_with(&Host::REAL, name).await +} + +async fn destroy_container_with(host: &Host<'_>, name: &str) -> Result<(String, String)> { validate_agent_name(name)?; - container_run(&["destroy", &container_system_name(name)]).await + container_run_with(host, &["destroy", &container_system_name(name)]).await } /// Shared body for `CreateContainer` / `UpdateContainer`: validate the name @@ -168,15 +234,28 @@ pub(crate) async fn container_toplevel_action( ) -> Result<(String, String)> { validate_agent_name(name)?; let toplevel = crate::closure::applicable_for(name, toplevel)?; + apply_toplevel(&Host::REAL, verb, name, toplevel, stream, writer).await +} + +/// [`container_toplevel_action`] past the check that `toplevel` is the +/// closure realised for `name`. +async fn apply_toplevel( + host: &Host<'_>, + verb: &str, + name: &str, + toplevel: &str, + stream: bool, + writer: &mut OwnedWriteHalf, +) -> Result<(String, String)> { let system_name = container_system_name(name); if verb == "update" { - return swap_container_profile(&system_name, toplevel, stream, writer).await; + return swap_container_profile(host, &system_name, toplevel, stream, writer).await; } let args = [verb, &system_name, "--system-path", toplevel]; if stream { - container_run_streaming(&args, writer).await + container_run_streaming(host, &args, writer).await } else { - container_run(&args).await + container_run_with(host, &args).await } } @@ -193,13 +272,15 @@ pub(crate) async fn container_toplevel_action( /// 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( + host: &Host<'_>, 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") + let set_out = host + .command("nix-env") .args(["-p", &profile, "--set", toplevel]) .output() .await @@ -216,14 +297,16 @@ async fn swap_container_profile( } let unit = format!("container@{system_name}"); - let state_out = Command::new("systemctl") + let state_out = host + .command("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") + let reload_out = host + .command("systemctl") .args(["reload", &unit]) .output() .await @@ -362,7 +445,12 @@ pub(crate) async fn run_bounded( /// 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") + container_run_with(&Host::REAL, args).await +} + +async fn container_run_with(host: &Host<'_>, args: &[&str]) -> Result<(String, String)> { + let out = host + .command("nixos-container") .args(args) .output() .await @@ -396,8 +484,9 @@ pub(crate) async fn container_run(args: &[&str]) -> Result<(String, String)> { /// 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") +async fn machinectl_run(host: &Host<'_>, args: &[&str]) -> Result<(String, String)> { + let out = host + .command("machinectl") .args(args) .output() .await @@ -421,21 +510,6 @@ async fn machinectl_run(args: &[&str]) -> Result<(String, String)> { 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 — @@ -451,13 +525,14 @@ const MACHINE_PROBE_TIMEOUT: std::time::Duration = std::time::Duration::from_sec /// 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") +async fn machine_registered(host: &Host<'_>, machine: &str) -> bool { + let probe = host + .command("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 { + match tokio::time::timeout(host.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"); @@ -472,16 +547,16 @@ async fn machine_registered(machine: &str) -> bool { /// 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; +async fn wait_name_released(host: &Host<'_>, machine: &str) -> bool { + let deadline = tokio::time::Instant::now() + host.release_timeout; loop { - if !machine_registered(machine).await { + if !machine_registered(host, machine).await { return true; } if tokio::time::Instant::now() >= deadline { return false; } - tokio::time::sleep(NAME_RELEASE_POLL).await; + tokio::time::sleep(host.release_poll).await; } } @@ -514,9 +589,9 @@ async fn wait_name_released(machine: &str) -> bool { /// 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 { +async fn stop_and_release(host: &Host<'_>, machine: &str) -> Result<(String, String)> { + let stop = container_run_with(host, &["stop", machine]).await; + if wait_name_released(host, 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. @@ -535,7 +610,7 @@ async fn stop_and_release(machine: &str) -> Result<(String, String)> { 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() + host.release_timeout.as_secs() ) } @@ -545,13 +620,14 @@ async fn stop_and_release(machine: &str) -> Result<(String, String)> { /// on success (all output was streamed); the error string includes stderr /// tail on failure. async fn container_run_streaming( + host: &Host<'_>, args: &[&str], writer: &mut OwnedWriteHalf, ) -> Result<(String, String)> { use tokio::io::AsyncBufReadExt as _; - use tokio::process::Command; - let mut child = Command::new("nixos-container") + let mut child = host + .command("nixos-container") .args(args) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) @@ -774,3 +850,544 @@ mod start_log_tests { assert_eq!(small, "a\n"); } } + +#[cfg(test)] +mod host_tests { + use std::path::PathBuf; + use std::time::Duration; + + use tokio::io::AsyncReadExt as _; + use tokio::net::UnixStream; + + use super::*; + use crate::shared::test_support::scratch; + + const TOPLEVEL: &str = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-bob"; + + /// Programs that are `sh -c