diff --git a/hive-c0re/src/priv_client.rs b/hive-c0re/src/priv_client.rs index 791782ef..e55d3208 100644 --- a/hive-c0re/src/priv_client.rs +++ b/hive-c0re/src/priv_client.rs @@ -8,8 +8,9 @@ use anyhow::{Context as _, Result, bail}; use hive_priv_sock::{ - BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery, NetworkIsolation, - PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, + BindMount, CredentialMount, InfraAction, InfraContainer, JournalQuery, + MEASURE_AGENT_DISK_TIMEOUT, NetworkIsolation, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, + PrivStream, }; use std::os::fd::{AsRawFd as _, OwnedFd, RawFd}; @@ -429,25 +430,44 @@ pub async fn read_subvolume_usage(agent_name: &str) -> Result<(u64, u64)> { parse_qgroup_usage(&stdout).with_context(|| format!("parse qgroup usage for {agent_name}")) } +/// How long hive-c0re waits for hive-priv's answer to a disk measurement: +/// hive-priv's own bound plus a margin, so a hung hive-priv can't stall the +/// caller while a slow walk still arrives as hive-priv's timeout error. +const MEASURE_AGENT_DISK_WAIT: std::time::Duration = + MEASURE_AGENT_DISK_TIMEOUT.saturating_add(std::time::Duration::from_mins(1)); + /// Measure an agent's disk usage in bytes (`state/` + `work/` + container /// rootfs, one bounded `du` run by hive-priv as root). /// /// # Errors -/// Returns an error if the hive-priv call fails, the `du` walk times out or -/// exits non-zero, or the returned total isn't a number. +/// Returns an error if the hive-priv call fails or doesn't answer within +/// [`MEASURE_AGENT_DISK_WAIT`], the `du` walk times out or exits non-zero, +/// the agent has none of the paths, or the returned total isn't a number. pub async fn measure_agent_disk(agent_name: &str) -> Result { - let (stdout, _) = check( + let resp = within( + MEASURE_AGENT_DISK_WAIT, call(&PrivRequest::MeasureAgentDisk { agent_name: agent_name.to_owned(), - }) - .await?, - )?; + }), + ) + .await?; + let (stdout, _) = check(resp)?; stdout .trim() .parse() .with_context(|| format!("parse disk usage for {agent_name}: {stdout:?}")) } +/// Await `fut`, failing if it takes longer than `limit`. +async fn within( + limit: std::time::Duration, + fut: impl std::future::Future>, +) -> Result { + tokio::time::timeout(limit, fut) + .await + .map_err(|_| anyhow::anyhow!("hive-priv did not answer within {}s", limit.as_secs()))? +} + /// Set or clear a btrfs qgroup size limit on an agent's state subvolume. /// `limit_bytes = None` clears it. Requires quota enabled. Via hive-priv. /// @@ -577,3 +597,26 @@ fn ok(resp: PrivResponse) -> Result<()> { check(resp)?; Ok(()) } + +#[cfg(test)] +mod tests { + use super::within; + use std::time::Duration; + + #[tokio::test] + async fn a_silent_hive_priv_times_out_as_an_error() { + let err = within( + Duration::from_millis(20), + std::future::pending::>(), + ) + .await + .expect_err("a call that never answers must fail"); + assert!(err.to_string().contains("did not answer"), "{err}"); + } + + #[tokio::test] + async fn an_answer_inside_the_limit_passes_through() { + let got = within(Duration::from_secs(5), async { Ok(7_u64) }).await; + assert_eq!(got.expect("answered in time"), 7); + } +} diff --git a/hive-priv-sock/src/lib.rs b/hive-priv-sock/src/lib.rs index c8e61b8b..ad4b8619 100644 --- a/hive-priv-sock/src/lib.rs +++ b/hive-priv-sock/src/lib.rs @@ -291,6 +291,11 @@ pub struct NetworkIsolation { pub gateway_ip: String, } +/// Wall-clock bound hive-priv puts on one [`PrivRequest::MeasureAgentDisk`] +/// walk. A caller's own wait has to be longer than this, so a slow walk +/// reaches it as hive-priv's timeout error rather than as its own. +pub const MEASURE_AGENT_DISK_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(10); + /// A request to the privileged helper. /// /// Wire format: one JSON object per line over `/run/hive/priv.sock`. @@ -565,7 +570,8 @@ pub enum PrivRequest { /// bytes as a decimal string in `stdout`. /// /// hive-priv derives all three paths from `agent_name`; the request - /// carries no path. A path that doesn't exist is skipped. A timeout, a + /// carries no path. A path that doesn't exist is skipped; an agent with + /// none of the three has no measurable size and gets an error. A timeout, a /// non-zero `du` exit or unparseable output is an error, never a partial /// or zero total. Requires root: `hive-core` cannot read the rootfs or /// the agent-owned files under `state/` and `work/`. diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index 5c8dd607..5962d06b 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -23,9 +23,9 @@ use std::path::{Path, PathBuf}; use anyhow::{Context as _, Result, anyhow, bail}; use hive_priv_sock::{ AGENT_PREFIX, AGENT_RUNTIME_ROOT, AGENT_STATE_ROOT, BindMount, CredentialMount, InfraAction, - InfraContainer, JournalQuery, META_DIR, MIGRATE_STAGING_ROOT, NetworkIsolation, - PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, - PrivStreamLine, SIBLING_CONTAINERS, + InfraContainer, JournalQuery, MEASURE_AGENT_DISK_TIMEOUT, META_DIR, MIGRATE_STAGING_ROOT, + NetworkIsolation, PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, + PrivStream, PrivStreamLine, SIBLING_CONTAINERS, }; use tokio::io::{AsyncWriteExt, BufReader}; use tokio::net::unix::OwnedWriteHalf; @@ -1684,10 +1684,6 @@ async fn read_subvolume_usage(agent_name: &str) -> Result<(String, String)> { )) } -/// Wall-clock bound on one [`measure_agent_disk`] walk. It only stops a walk -/// wedged on I/O from holding the caller's sampler forever. -const DISK_USAGE_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(10); - /// The host paths `MeasureAgentDisk` sums for `agent_name`: its `state/` /// and `work/` dirs under `AGENT_STATE_ROOT` and its container rootfs. /// These are the only paths that op ever walks. @@ -1703,10 +1699,8 @@ fn agent_disk_paths(agent_name: &str) -> [PathBuf; 3] { } /// `MeasureAgentDisk` — one `du` over the [`agent_disk_paths`] that exist, -/// returning the grand total in bytes. `-x` keeps each walk on its -/// operand's filesystem (so the bind-mounted nix store is never counted), -/// `-P` never follows a symlink, `-b` counts apparent size, `-c` adds the -/// `total` line [`parse_du_total`] reads. +/// returning the grand total in bytes. An agent with none of them has no +/// measurable size, which is an error rather than a total of 0. async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> { let mut paths = Vec::new(); for path in agent_disk_paths(agent_name) { @@ -1716,18 +1710,33 @@ async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> { Err(e) => return Err(e).with_context(|| format!("stat {}", path.display())), } } - // `du` with no operand would measure its working directory instead. + let total = du_total(agent_name, du_command(&paths)?, MEASURE_AGENT_DISK_TIMEOUT).await?; + Ok((total.to_string(), String::new())) +} + +/// `du -s -x -b -c -P -- `. `-x` keeps each walk on its operand's +/// filesystem (so the bind-mounted nix store is never counted), `-P` never +/// follows a symlink, `-b` counts apparent size, `-c` adds the `total` line +/// [`parse_du_total`] reads. Errors on no paths: `du` with no operand would +/// measure its working directory instead. +fn du_command(paths: &[PathBuf]) -> Result { if paths.is_empty() { - return Ok(("0".to_owned(), String::new())); + bail!("none of the agent's state, work or rootfs paths exist"); } let mut cmd = Command::new("du"); - cmd.args(["-s", "-x", "-b", "-c", "-P", "--"]) - .args(&paths) - .kill_on_drop(true); - let Ok(out) = tokio::time::timeout(DISK_USAGE_TIMEOUT, cmd.output()).await else { + cmd.args(["-s", "-x", "-b", "-c", "-P", "--"]).args(paths); + Ok(cmd) +} + +/// Run `cmd` (a [`du_command`]) for at most `limit` and parse its total. +/// The child is killed on timeout. A timeout, a spawn failure, a non-zero +/// exit or output without a total is an error, never a partial total. +async fn du_total(agent_name: &str, mut cmd: Command, limit: std::time::Duration) -> Result { + cmd.kill_on_drop(true); + let Ok(out) = tokio::time::timeout(limit, cmd.output()).await else { bail!( "du for {agent_name} timed out after {}s; killed it", - DISK_USAGE_TIMEOUT.as_secs() + limit.as_secs() ); }; let out = out.with_context(|| format!("spawn du for {agent_name}"))?; @@ -1743,9 +1752,8 @@ async fn measure_agent_disk(agent_name: &str) -> Result<(String, String)> { ); } let stdout = String::from_utf8_lossy(&out.stdout); - let total = parse_du_total(&stdout) - .ok_or_else(|| anyhow!("no `total` line in du output for {agent_name}: {stdout:?}"))?; - Ok((total.to_string(), String::new())) + parse_du_total(&stdout) + .ok_or_else(|| anyhow!("no `total` line in du output for {agent_name}: {stdout:?}")) } /// The byte count on `du -c`'s final `\ttotal` line. `None` when the @@ -3178,16 +3186,17 @@ fn ensure_socket_dir_in(root: &Path, name: &str) -> Result<()> { #[cfg(test)] mod tests { use super::{ - BindMount, BoundedRun, OwnedFd, PAUSED_MARKER_FILE, PrivRequest, StagedFile, + BindMount, BoundedRun, Command, OwnedFd, PAUSED_MARKER_FILE, PrivRequest, StagedFile, agent_disk_paths, check_fd_agreement, contains_secret_shaped_run, describe_forge_admin, - ensure_plain_filename, ensure_socket_dir_in, exec_measure_agent_disk, git_overlay_flags, - limits_dropin_body, open_dir, open_export_dest, parse_du_total, partial_name, publish_file, - redact_secret_line, remove_marker_in, run_bounded, single_output_path, toplevel_attr, - validate_credential_name, validate_snapshot_name, write_agent_dir_file, - write_bridge_dns_marker_in, + du_command, du_total, ensure_plain_filename, ensure_socket_dir_in, exec_measure_agent_disk, + git_overlay_flags, limits_dropin_body, open_dir, open_export_dest, parse_du_total, + partial_name, publish_file, redact_secret_line, remove_marker_in, run_bounded, + single_output_path, toplevel_attr, validate_credential_name, validate_snapshot_name, + write_agent_dir_file, write_bridge_dns_marker_in, }; use std::path::PathBuf; use std::sync::atomic::{AtomicU32, Ordering}; + use std::time::Duration; fn bind(container_path: &str) -> BindMount { BindMount { @@ -4240,4 +4249,41 @@ mod tests { assert_eq!(parse_du_total("-1\ttotal\n"), None); assert_eq!(parse_du_total("100 total\n"), None); } + + #[test] + fn no_existing_paths_is_an_error_not_zero() { + let err = du_command(&[]).expect_err("no operands must not run du"); + assert!(err.to_string().contains("none of"), "{err}"); + } + + #[tokio::test] + async fn non_zero_du_exit_is_an_error() { + let err = du_total("iris", Command::new("false"), Duration::from_secs(5)) + .await + .expect_err("a failed du must not yield a total"); + assert!(err.to_string().contains("failed"), "{err}"); + } + + #[tokio::test] + async fn a_walk_past_the_limit_is_killed_and_errors() { + let mut cmd = Command::new("sleep"); + cmd.arg("30"); + let started = std::time::Instant::now(); + let err = du_total("iris", cmd, Duration::from_millis(50)) + .await + .expect_err("a walk past the limit must fail"); + assert!(err.to_string().contains("timed out"), "{err}"); + assert!(started.elapsed() < Duration::from_secs(10)); + } + + #[tokio::test] + async fn du_total_sums_a_real_walk() { + let dir = std::env::temp_dir().join(format!("hive-priv-du-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + std::fs::write(dir.join("f"), b"hello").unwrap(); + let cmd = du_command(std::slice::from_ref(&dir)).unwrap(); + let total = du_total("iris", cmd, Duration::from_secs(30)).await; + std::fs::remove_dir_all(&dir).unwrap(); + assert!(total.unwrap() >= 5); + } }