diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index a8dbc8cf..56040527 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -358,16 +358,6 @@ 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 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, including -the `rev-parse` and `show` reads `pin_agents` makes. The error names the command -and the deadline. Each child leads its own process group and the timeout -`SIGKILL`s the whole group, so a `git` or credential helper spawned by a hung -`nix` dies too. 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. Waiters keep their place in the lock's FIFO queue across warnings. - --- ## Container lifecycle (`lifecycle.rs`) diff --git a/hive-c0re/src/forge/repos.rs b/hive-c0re/src/forge/repos.rs index a1ce5b4a..fecb73c6 100644 --- a/hive-c0re/src/forge/repos.rs +++ b/hive-c0re/src/forge/repos.rs @@ -111,14 +111,12 @@ 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::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?; + 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")?; 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 e0819757..610e3beb 100644 --- a/hive-c0re/src/lifecycle/git.rs +++ b/hive-c0re/src/lifecycle/git.rs @@ -3,72 +3,10 @@ //! fetch proposal commits into the applied repo. use std::path::Path; -use std::process::{Output, Stdio}; -use std::time::Duration; use anyhow::{Context, Result, bail}; -use tokio::io::{AsyncRead, AsyncReadExt}; 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 leads its own -/// process group and the whole group is `SIGKILL`ed on timeout, so a -/// grandchild (a `git` or credential helper under `nix`) dies with it instead -/// of outliving the lock. -pub async fn run_with_deadline( - cmd: &mut Command, - deadline: Duration, - what: &str, -) -> Result { - // `kill_on_drop` covers the caller dropping this future; `process_group(0)` - // makes the child's pid the group id the timeout path kills. - let mut child = cmd - .stdin(Stdio::null()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .kill_on_drop(true) - .process_group(0) - .spawn() - .with_context(|| what.to_owned())?; - // Read before waiting: a reaped child has no id. - let pgid = child.id().and_then(|pid| i32::try_from(pid).ok()); - let (stdout, stderr) = (child.stdout.take(), child.stderr.take()); - let io = async { tokio::join!(child.wait(), drain(stdout), drain(stderr)) }; - if let Ok((status, stdout, stderr)) = tokio::time::timeout(deadline, io).await { - return Ok(Output { - status: status.with_context(|| what.to_owned())?, - stdout: stdout.with_context(|| what.to_owned())?, - stderr: stderr.with_context(|| what.to_owned())?, - }); - } - 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.with_context(|| what.to_owned())?; - bail!("{what} timed out after {}s", deadline.as_secs_f64()) -} - -/// Everything `pipe` yields until EOF. -async fn drain(pipe: Option) -> std::io::Result> { - let mut buf = Vec::new(); - if let Some(mut p) = pipe { - p.read_to_end(&mut buf).await?; - } - Ok(buf) -} - const GIT_NAME: &str = "c0re"; const GIT_EMAIL: &str = "c0re@hyperhive.local"; @@ -142,11 +80,14 @@ pub fn git_command_authed(auth_header: &str) -> Command { cmd } -/// 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?; +/// 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()))?; if !out.status.success() { bail!( "git {} failed ({}): {}", @@ -159,24 +100,24 @@ async fn run(mut cmd: Command, dir: &Path, args: &[&str], deadline: Duration) -> } pub async fn git(dir: &Path, args: &[&str]) -> Result<()> { - run(git_command(), dir, args, GIT_LOCAL_DEADLINE).await + run(git_command(), dir, args).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, FETCH_DEADLINE).await + run(git_command_authed(auth_header), dir, args).await } /// Resolve `refname` (a tag, branch, or sha) in `dir` to its full sha. pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result { - let out = run_with_deadline( - git_command().current_dir(dir).args(["rev-parse", refname]), - GIT_LOCAL_DEADLINE, - &format!("git rev-parse {refname} in {}", dir.display()), - ) - .await?; + let out = git_command() + .current_dir(dir) + .args(["rev-parse", refname]) + .output() + .await + .with_context(|| format!("git rev-parse {refname} in {}", dir.display()))?; if !out.status.success() { bail!( "git rev-parse {refname} failed ({}): {}", @@ -191,14 +132,12 @@ pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result { /// such file. Reads the object database, not the working tree. pub async fn git_show_file(dir: &Path, rev: &str, path: &str) -> Result> { let spec = format!("{rev}:{path}"); - let listed = run_with_deadline( - git_command() - .current_dir(dir) - .args(["ls-tree", "--name-only", rev, "--", path]), - GIT_LOCAL_DEADLINE, - &format!("git ls-tree {spec} in {}", dir.display()), - ) - .await?; + let listed = git_command() + .current_dir(dir) + .args(["ls-tree", "--name-only", rev, "--", path]) + .output() + .await + .with_context(|| format!("git ls-tree {spec} in {}", dir.display()))?; if !listed.status.success() { bail!( "git ls-tree {spec} failed ({}): {}", @@ -209,12 +148,12 @@ pub async fn git_show_file(dir: &Path, rev: &str, path: &str) -> Result {}; sleep 300 & echo $! > {}; wait", - shpid.display(), - sleeppid.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}"); - - for file in [&shpid, &sleeppid] { - // The pid file proves the process existed before the kill. - let pid = std::fs::read_to_string(file) - .expect("process wrote its pid before the deadline") - .trim() - .to_owned(); - let proc = std::path::PathBuf::from("/proc").join(&pid); - let mut gone = false; - for _ in 0..100 { - if !proc.exists() { - gone = true; - break; - } - tokio::time::sleep(std::time::Duration::from_millis(50)).await; - } - assert!(gone, "pid {pid} ({}) still running", file.display()); - } -} - -/// 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 4333a208..b4201afb 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -5,13 +5,10 @@ //! `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, MutexGuard}; -use tokio::time::Instant; +use tokio::sync::Mutex; use crate::coordinator::HiveEnv; use crate::lifecycle; @@ -30,77 +27,6 @@ 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(); - // One acquire future for the whole wait: dropping it loses the place in - // the mutex's FIFO queue, so the stall timer runs beside it, not around it. - let acquire = lock.lock(); - tokio::pin!(acquire); - let guard = loop { - tokio::select! { - guard = &mut acquire => break guard, - () = tokio::time::sleep(interval) => 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 @@ -154,7 +80,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 = lock_meta("sync_agents").await; + let _guard = META_LOCK.lock().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. @@ -289,7 +215,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 = lock_meta("lock_update_for_rebuild").await; + let _guard = META_LOCK.lock().await; let dir = crate::paths::meta_root(); let input = format!("agent-{name}"); nix(&dir, &rebuild_relock_args(&input)).await?; @@ -323,7 +249,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 = lock_meta("lock_update").await; + let _guard = META_LOCK.lock().await; let dir = crate::paths::meta_root(); let mut args: Vec<&str> = vec!["flake", "update"]; for i in inputs { @@ -348,7 +274,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 = lock_meta("lock_update_hyperhive").await; + let _guard = META_LOCK.lock().await; let dir = crate::paths::meta_root(); nix(&dir, &["flake", "update", "hyperhive"]).await?; if !paths_dirty(&dir, &["flake.lock"]).await? { @@ -366,7 +292,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 = lock_meta("remove_permission_files").await; + let _guard = META_LOCK.lock().await; let dir = crate::paths::meta_root(); let mut removed = Vec::new(); for file in crate::workers::permissions_import::META_FILES { @@ -998,11 +924,12 @@ 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 = git_output( - dir, - &["diff", "--cached", "--name-only", "--diff-filter=ACM"], - ) - .await?; + 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()))?; if out.status.success() { let names = String::from_utf8_lossy(&out.stdout) .lines() @@ -1013,7 +940,12 @@ async fn git_staged_names(dir: &Path) -> Result> { return Ok(names); } // Fallback for initial commit (no HEAD yet): parse `git status --porcelain`. - let st = git_output(dir, &["status", "--porcelain"]).await?; + let st = lifecycle::git_command() + .current_dir(dir) + .args(["status", "--porcelain"]) + .output() + .await + .with_context(|| format!("git status in {}", dir.display()))?; let names = String::from_utf8_lossy(&st.stdout) .lines() .filter(|l| l.starts_with("A ") || l.starts_with("M ")) @@ -1023,19 +955,13 @@ 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 = git_output(dir, args).await?; + let out = lifecycle::git_command() + .current_dir(dir) + .args(args) + .output() + .await + .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?; if !out.status.success() { bail!( "git {} failed ({}): {}", @@ -1095,7 +1021,12 @@ 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 = git_output(dir, &args).await?; + let out = lifecycle::git_command() + .current_dir(dir) + .args(&args) + .output() + .await + .with_context(|| format!("git diff --quiet in {}", dir.display()))?; // 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. @@ -1115,28 +1046,17 @@ 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 { - 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 + 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())) } /// Turn a finished nix [`Output`](std::process::Output) into `Result<()>`, @@ -2172,94 +2092,3 @@ 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)]); - } - - /// Waiters get the lock in arrival order even when one of them stalls past - /// an interval: A queues at 0s, B at 30s, the holder releases at 70s. A - /// re-armed acquire would put A (stalled at 60s) behind B. - #[tokio::test(start_paused = true)] - async fn waiters_acquire_in_arrival_order_across_a_stall() { - use std::sync::Arc; - static LOCK: Mutex<()> = Mutex::const_new(()); - static HOLDER: StdMutex<&'static str> = StdMutex::new(""); - let order = Arc::new(StdMutex::new(Vec::new())); - let first = lock_notifying(&LOCK, &HOLDER, "first", Duration::from_mins(1), |_, _| { - panic!("an uncontended lock must not report a stall") - }) - .await; - let spawn_waiter = |op: &'static str| { - let order = Arc::clone(&order); - tokio::spawn(async move { - let guard = - lock_notifying(&LOCK, &HOLDER, op, Duration::from_mins(1), |_, _| {}).await; - order.lock().unwrap().push(op); - drop(guard); - }) - }; - let a = spawn_waiter("A"); - tokio::time::sleep(Duration::from_secs(30)).await; - let b = spawn_waiter("B"); - tokio::time::sleep(Duration::from_secs(40)).await; - drop(first); - a.await.expect("A"); - b.await.expect("B"); - assert_eq!(*order.lock().unwrap(), vec!["A", "B"]); - } - - /// 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}" - ); - } -}