diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index 56040527..c3868a6a 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -358,6 +358,13 @@ Every public `meta.rs` operation takes the module's internal `META_LOCK` mutex, so concurrent job-queue nodes never race on the repo's `.git/index.lock`. +Each git or nix child `meta.rs` runs under `META_LOCK` has a deadline, so a hung +forge fetch fails the operation instead of holding the lock: 10 minutes for +`nix flake update`/`lock` and forge fetch/push, 2 minutes for local git. The error +names the command and the deadline, and the child dies with it. A caller waiting on +`META_LOCK` for over a minute logs a `meta_lock: still waiting` warning every +minute, naming its own operation and the holder's. + --- ## Container lifecycle (`lifecycle.rs`) diff --git a/hive-c0re/src/forge/repos.rs b/hive-c0re/src/forge/repos.rs index fecb73c6..a1ce5b4a 100644 --- a/hive-c0re/src/forge/repos.rs +++ b/hive-c0re/src/forge/repos.rs @@ -111,12 +111,14 @@ pub async fn push_meta(dir: &Path) -> Result<()> { // raise a dashboard warning and leave the remote intact rather than // silently erasing its history. let url = forge_git_url("core/meta"); - let out = crate::lifecycle::git_command_authed(&core_auth_header(&token)) - .current_dir(dir) - .args(["push", &url, "HEAD:main"]) - .output() - .await - .context("invoke git push core/meta")?; + let out = crate::lifecycle::run_with_deadline( + crate::lifecycle::git_command_authed(&core_auth_header(&token)) + .current_dir(dir) + .args(["push", &url, "HEAD:main"]), + crate::lifecycle::FETCH_DEADLINE, + "git push core/meta", + ) + .await?; if !out.status.success() { let stderr = String::from_utf8_lossy(&out.stderr); if stderr.contains("non-fast-forward") || stderr.contains("fetch first") { diff --git a/hive-c0re/src/lifecycle/git.rs b/hive-c0re/src/lifecycle/git.rs index 610e3beb..395fb462 100644 --- a/hive-c0re/src/lifecycle/git.rs +++ b/hive-c0re/src/lifecycle/git.rs @@ -3,10 +3,36 @@ //! fetch proposal commits into the applied repo. use std::path::Path; +use std::process::Output; +use std::time::Duration; use anyhow::{Context, Result, bail}; use tokio::process::Command; +/// Deadline for a child that talks to a remote (forge fetch/push, `nix flake +/// update`/`lock`): a healthy one takes seconds, so ten minutes only ever +/// fires on a hang. +pub const FETCH_DEADLINE: Duration = Duration::from_mins(10); + +/// Deadline for a git child that touches only the local repo. +pub const GIT_LOCAL_DEADLINE: Duration = Duration::from_mins(2); + +/// Run `cmd` to completion and capture its output, failing once `deadline` +/// passes. `what` names the command in the error. The child is killed when the +/// deadline drops its future, so a hung remote cannot leave an orphan behind. +pub async fn run_with_deadline( + cmd: &mut Command, + deadline: Duration, + what: &str, +) -> Result { + // `kill_on_drop` is what turns the timeout's drop into a kill. + cmd.kill_on_drop(true); + tokio::time::timeout(deadline, cmd.output()) + .await + .map_err(|_| anyhow::anyhow!("{what} timed out after {}s", deadline.as_secs_f64()))? + .with_context(|| what.to_owned()) +} + const GIT_NAME: &str = "c0re"; const GIT_EMAIL: &str = "c0re@hyperhive.local"; @@ -80,14 +106,11 @@ pub fn git_command_authed(auth_header: &str) -> Command { cmd } -/// Run `cmd` as `git ` in `dir`, erroring on a non-zero exit. -async fn run(mut cmd: Command, dir: &Path, args: &[&str]) -> Result<()> { - let out = cmd - .current_dir(dir) - .args(args) - .output() - .await - .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?; +/// Run `cmd` as `git ` in `dir` within `deadline`, erroring on a +/// non-zero exit. +async fn run(mut cmd: Command, dir: &Path, args: &[&str], deadline: Duration) -> Result<()> { + let what = format!("git {} in {}", args.join(" "), dir.display()); + let out = run_with_deadline(cmd.current_dir(dir).args(args), deadline, &what).await?; if !out.status.success() { bail!( "git {} failed ({}): {}", @@ -100,14 +123,14 @@ async fn run(mut cmd: Command, dir: &Path, args: &[&str]) -> Result<()> { } pub async fn git(dir: &Path, args: &[&str]) -> Result<()> { - run(git_command(), dir, args).await + run(git_command(), dir, args, GIT_LOCAL_DEADLINE).await } /// [`git`] against an authenticated remote — for the paths that talk to the /// forge. The credential rides the environment, so `args` (and therefore the /// error above) stay free of it. pub async fn git_authed(dir: &Path, args: &[&str], auth_header: &str) -> Result<()> { - run(git_command_authed(auth_header), dir, args).await + run(git_command_authed(auth_header), dir, args, FETCH_DEADLINE).await } /// Resolve `refname` (a tag, branch, or sha) in `dir` to its full sha. diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index 8578a02e..8e9ce904 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -8,8 +8,9 @@ mod setup; mod tests; pub use git::{ - git, git_authed, git_command, git_command_authed, git_is_ancestor, git_read_tree_reset, - git_rev_parse, git_show_file, git_tag, git_update_ref, git_update_ref_cas, + FETCH_DEADLINE, GIT_LOCAL_DEADLINE, git, git_authed, git_command, git_command_authed, + git_is_ancestor, git_read_tree_reset, git_rev_parse, git_show_file, git_tag, git_update_ref, + git_update_ref_cas, run_with_deadline, }; pub use host_config::{hand_permissions, write_dropins}; pub use setup::{ diff --git a/hive-c0re/src/lifecycle/tests.rs b/hive-c0re/src/lifecycle/tests.rs index 6ab6639b..653c843f 100644 --- a/hive-c0re/src/lifecycle/tests.rs +++ b/hive-c0re/src/lifecycle/tests.rs @@ -473,3 +473,48 @@ async fn a_systemctl_answer_inside_the_bound_passes_through() { .await; assert_eq!(got, Some(7)); } + +/// A child that outlives the deadline yields an error naming the command and +/// the deadline, and is killed rather than orphaned. +#[tokio::test] +async fn run_with_deadline_times_out_and_kills_the_child() { + let dir = tempfile::tempdir().expect("tempdir"); + let pidfile = dir.path().join("pid"); + let mut cmd = tokio::process::Command::new("sh"); + cmd.arg("-c") + .arg(format!("echo $$ > {}; exec sleep 30", pidfile.display())); + let err = run_with_deadline( + &mut cmd, + std::time::Duration::from_millis(500), + "fake fetch", + ) + .await + .expect_err("a sleeping child must hit the deadline"); + let msg = format!("{err:#}"); + assert!(msg.contains("fake fetch timed out after 0.5s"), "{msg}"); + + let pid = std::fs::read_to_string(&pidfile) + .expect("child wrote its pid before the deadline") + .trim() + .to_owned(); + let proc = std::path::PathBuf::from("/proc").join(&pid); + for _ in 0..100 { + if !proc.exists() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + } + panic!("child {pid} still running after the deadline dropped its future"); +} + +/// Presence control for the timeout test: a child that finishes in time +/// returns its output. +#[tokio::test] +async fn run_with_deadline_returns_output_of_a_fast_child() { + let mut cmd = tokio::process::Command::new("echo"); + cmd.arg("hi"); + let out = run_with_deadline(&mut cmd, std::time::Duration::from_secs(30), "echo") + .await + .expect("echo finishes"); + assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "hi"); +} diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index b4201afb..064f6e90 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -5,10 +5,13 @@ //! `docs/agent-lifecycle/config-changes.md::Meta flake`. use std::path::Path; +use std::sync::{Mutex as StdMutex, PoisonError}; +use std::time::Duration; use anyhow::{Context, Result, bail}; use tokio::process::Command; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, MutexGuard}; +use tokio::time::Instant; use crate::coordinator::HiveEnv; use crate::lifecycle; @@ -27,6 +30,73 @@ const GIT_EMAIL: &str = "c0re@hyperhive.local"; /// take turns instead of colliding. static META_LOCK: Mutex<()> = Mutex::const_new(()); +/// Name of the operation holding [`META_LOCK`], empty when free. Read only to +/// name the holder in a stalled-waiter warning. +static META_HOLDER: StdMutex<&'static str> = StdMutex::new(""); + +/// A wait on [`META_LOCK`] longer than this logs a warning, then again every +/// interval, naming the waiter and the holder. +const LOCK_WAIT_WARN: Duration = Duration::from_mins(1); + +/// [`META_LOCK`] guard that records its holder in [`META_HOLDER`]. +struct MetaGuard<'a> { + _lock: MutexGuard<'a, ()>, + holder: &'a StdMutex<&'static str>, +} + +impl Drop for MetaGuard<'_> { + // Runs before the field drop releases the mutex, so the next holder's + // name is never overwritten by this one's clear. + fn drop(&mut self) { + *self.holder.lock().unwrap_or_else(PoisonError::into_inner) = ""; + } +} + +/// Take [`META_LOCK`] for `op`, warning while the wait drags on. +async fn lock_meta(op: &'static str) -> MetaGuard<'static> { + lock_notifying( + &META_LOCK, + &META_HOLDER, + op, + LOCK_WAIT_WARN, + |holder, waited| { + tracing::warn!( + op, + holder, + waited_secs = waited.as_secs(), + "meta_lock: still waiting for META_LOCK" + ); + }, + ) + .await +} + +/// Take `lock`, calling `on_stall(holder, waited)` each time `interval` passes +/// without getting it. +async fn lock_notifying<'a>( + lock: &'a Mutex<()>, + holder: &'a StdMutex<&'static str>, + op: &'static str, + interval: Duration, + mut on_stall: impl FnMut(&'static str, Duration), +) -> MetaGuard<'a> { + let start = Instant::now(); + let guard = loop { + match tokio::time::timeout(interval, lock.lock()).await { + Ok(guard) => break guard, + Err(_) => on_stall( + *holder.lock().unwrap_or_else(PoisonError::into_inner), + start.elapsed(), + ), + } + }; + *holder.lock().unwrap_or_else(PoisonError::into_inner) = op; + MetaGuard { + _lock: guard, + holder, + } +} + // Exclusivity for meta-repo *windows* that span multiple `META_LOCK` // acquisitions is **not** a mutex in this module. `META_LOCK` above // serializes individual git ops but cannot keep another op out of a window @@ -80,7 +150,7 @@ async fn stage_generated_meta_files(dir: &std::path::Path) -> Result<()> { /// rendered contents differ from disk; an unchanged `flake.nix` is a /// no-op. pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> { - let _guard = META_LOCK.lock().await; + let _guard = lock_meta("sync_agents").await; // Before anything is written: a hive without a forge URL would deploy a // whole fleet of agents that silently never log in, because the agent // option treats "unset" as "no forge configured" rather than erroring. @@ -215,7 +285,7 @@ pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> { /// (no separate finalize) because rebuild has no failure-revert /// semantics. The input follows the url `sync_agents` rendered just before. pub async fn lock_update_for_rebuild(name: &str) -> Result<()> { - let _guard = META_LOCK.lock().await; + let _guard = lock_meta("lock_update_for_rebuild").await; let dir = crate::paths::meta_root(); let input = format!("agent-{name}"); nix(&dir, &rebuild_relock_args(&input)).await?; @@ -249,7 +319,7 @@ fn rebuild_relock_args(input: &str) -> [&str; 3] { /// real input keys (e.g. via `inputs_view()` snapshotted from the lock /// file) when targeting specific inputs. pub async fn lock_update(inputs: &[String]) -> Result<()> { - let _guard = META_LOCK.lock().await; + let _guard = lock_meta("lock_update").await; let dir = crate::paths::meta_root(); let mut args: Vec<&str> = vec!["flake", "update"]; for i in inputs { @@ -274,7 +344,7 @@ pub async fn lock_update(inputs: &[String]) -> Result<()> { /// rev, commit if the lock changed. Cheaper than `sync_agents` /// because the per-agent inputs aren't touched. pub async fn lock_update_hyperhive() -> Result<()> { - let _guard = META_LOCK.lock().await; + let _guard = lock_meta("lock_update_hyperhive").await; let dir = crate::paths::meta_root(); nix(&dir, &["flake", "update", "hyperhive"]).await?; if !paths_dirty(&dir, &["flake.lock"]).await? { @@ -292,7 +362,7 @@ pub async fn lock_update_hyperhive() -> Result<()> { /// /// Removing a file or a git stage/commit step fails. pub async fn remove_permission_files() -> Result<()> { - let _guard = META_LOCK.lock().await; + let _guard = lock_meta("remove_permission_files").await; let dir = crate::paths::meta_root(); let mut removed = Vec::new(); for file in crate::workers::permissions_import::META_FILES { @@ -924,12 +994,11 @@ fn render_flake( async fn git_staged_names(dir: &Path) -> Result> { // `--diff-filter=ACM` skips deleted entries — we only care about // additions and modifications for message-building purposes. - let out = lifecycle::git_command() - .current_dir(dir) - .args(["diff", "--cached", "--name-only", "--diff-filter=ACM"]) - .output() - .await - .with_context(|| format!("git diff --cached in {}", dir.display()))?; + let out = git_output( + dir, + &["diff", "--cached", "--name-only", "--diff-filter=ACM"], + ) + .await?; if out.status.success() { let names = String::from_utf8_lossy(&out.stdout) .lines() @@ -940,12 +1009,7 @@ async fn git_staged_names(dir: &Path) -> Result> { return Ok(names); } // Fallback for initial commit (no HEAD yet): parse `git status --porcelain`. - let st = lifecycle::git_command() - .current_dir(dir) - .args(["status", "--porcelain"]) - .output() - .await - .with_context(|| format!("git status in {}", dir.display()))?; + let st = git_output(dir, &["status", "--porcelain"]).await?; let names = String::from_utf8_lossy(&st.stdout) .lines() .filter(|l| l.starts_with("A ") || l.starts_with("M ")) @@ -955,13 +1019,19 @@ async fn git_staged_names(dir: &Path) -> Result> { Ok(names) } +/// Run `git ` in `dir` within [`lifecycle::GIT_LOCAL_DEADLINE`]; the caller +/// reads the exit status. +async fn git_output(dir: &Path, args: &[&str]) -> Result { + lifecycle::run_with_deadline( + lifecycle::git_command().current_dir(dir).args(args), + lifecycle::GIT_LOCAL_DEADLINE, + &format!("git {} in {}", args.join(" "), dir.display()), + ) + .await +} + async fn git(dir: &Path, args: &[&str]) -> Result<()> { - let out = lifecycle::git_command() - .current_dir(dir) - .args(args) - .output() - .await - .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?; + let out = git_output(dir, args).await?; if !out.status.success() { bail!( "git {} failed ({}): {}", @@ -1021,12 +1091,7 @@ async fn git_commit_paths(dir: &Path, message: &str, paths: &[&str]) -> Result<( async fn paths_dirty(dir: &Path, paths: &[&str]) -> Result { let mut args = vec!["diff", "--quiet", "HEAD", "--"]; args.extend_from_slice(paths); - let out = lifecycle::git_command() - .current_dir(dir) - .args(&args) - .output() - .await - .with_context(|| format!("git diff --quiet in {}", dir.display()))?; + let out = git_output(dir, &args).await?; // Exit 0 = no differences; 1 = differences; anything else (e.g. // no HEAD yet on a fresh repo) → treat as dirty so the commit // path runs and surfaces real errors loudly. @@ -1046,17 +1111,28 @@ fn nix_argv<'a>(args: &[&'a str]) -> Vec<&'a str> { /// Run `nix ` in `dir` (flakes enabled), capturing combined output. /// The core of [`nix`]. +/// +/// Every call site in this module is `flake lock` or `flake update`, which +/// fetch inputs from the forge, so the whole function runs under +/// [`lifecycle::FETCH_DEADLINE`]. A call that builds or evaluates a system +/// needs its own bound, not this one. async fn nix_output(dir: &Path, args: &[&str]) -> Result { - Command::new("nix") - .current_dir(dir) - .args(nix_argv(args)) - // Kill the nix child if the caller's future is dropped (e.g. the - // startup-migration timeout around `sync_agents` fires) rather than - // orphaning it against an unreachable forge. No-op on normal exit. - .kill_on_drop(true) - .output() - .await - .with_context(|| format!("nix {} in {}", args.join(" "), dir.display())) + nix_output_with("nix".as_ref(), dir, args, lifecycle::FETCH_DEADLINE).await +} + +/// [`nix_output`] with the program and deadline injectable for tests. +async fn nix_output_with( + program: &std::ffi::OsStr, + dir: &Path, + args: &[&str], + deadline: Duration, +) -> Result { + lifecycle::run_with_deadline( + Command::new(program).current_dir(dir).args(nix_argv(args)), + deadline, + &format!("nix {} in {}", args.join(" "), dir.display()), + ) + .await } /// Turn a finished nix [`Output`](std::process::Output) into `Result<()>`, @@ -2092,3 +2168,61 @@ mod tests { ); } } + +#[cfg(test)] +mod lock_wait_tests { + use super::*; + + /// A waiter behind a held lock is told, per interval, who holds it; once + /// the holder drops, it gets the lock and becomes the recorded holder. + #[tokio::test(start_paused = true)] + async fn stalled_waiter_reports_holder_then_acquires() { + let lock = Mutex::new(()); + let holder = StdMutex::new(""); + let first = lock_notifying(&lock, &holder, "first", Duration::from_mins(1), |_, _| { + panic!("an uncontended lock must not report a stall") + }) + .await; + assert_eq!(*holder.lock().unwrap(), "first"); + + let stalls = std::cell::RefCell::new(Vec::new()); + let waiter = lock_notifying(&lock, &holder, "second", Duration::from_mins(1), |h, w| { + stalls.borrow_mut().push((h, w.as_secs())); + }); + tokio::pin!(waiter); + // 150s of virtual time: two intervals elapse while `first` is held. + tokio::select! { + _ = &mut waiter => panic!("lock is held"), + () = tokio::time::sleep(Duration::from_secs(150)) => {} + } + drop(first); + assert_eq!(*holder.lock().unwrap(), ""); + let second = waiter.await; + assert_eq!(*holder.lock().unwrap(), "second"); + drop(second); + assert_eq!(*stalls.borrow(), vec![("first", 60), ("first", 120)]); + } + /// A hung nix child comes back as a timeout error naming the command. The + /// deadline and program are injected; the production const is untouched. + #[tokio::test] + async fn nix_output_times_out_naming_the_command() { + let dir = tempfile::tempdir().expect("tempdir"); + let fake = dir.path().join("nix"); + std::fs::write(&fake, "#!/bin/sh\nexec sleep 30\n").expect("write"); + std::fs::set_permissions(&fake, std::os::unix::fs::PermissionsExt::from_mode(0o755)) + .expect("chmod"); + let err = nix_output_with( + fake.as_os_str(), + dir.path(), + &["flake", "update", "hyperhive"], + Duration::from_millis(300), + ) + .await + .expect_err("hung nix must time out"); + let msg = format!("{err:#}"); + assert!( + msg.contains("nix flake update hyperhive in") && msg.contains("timed out after 0.3s"), + "{msg}" + ); + } +}