From 161dd8c44bef054fe5e8101aacaa39fed174be29 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 20 Oct 2026 00:23:39 +0200 Subject: [PATCH] meta lock waiters keep their queue place; deadlines cover pin_agent git and kill the whole process group --- docs/scheduler/coordinator.md | 15 +++--- hive-c0re/src/lifecycle/git.rs | 92 ++++++++++++++++++++++---------- hive-c0re/src/lifecycle/tests.rs | 39 ++++++++------ hive-c0re/src/meta.rs | 43 +++++++++++++-- 4 files changed, 138 insertions(+), 51 deletions(-) diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index c3868a6a..a8dbc8cf 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -358,12 +358,15 @@ 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. +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. --- diff --git a/hive-c0re/src/lifecycle/git.rs b/hive-c0re/src/lifecycle/git.rs index 395fb462..e0819757 100644 --- a/hive-c0re/src/lifecycle/git.rs +++ b/hive-c0re/src/lifecycle/git.rs @@ -3,10 +3,11 @@ //! fetch proposal commits into the applied repo. use std::path::Path; -use std::process::Output; +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 @@ -18,19 +19,54 @@ pub const FETCH_DEADLINE: Duration = Duration::from_mins(10); 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. +/// 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` 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()) + // `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"; @@ -135,12 +171,12 @@ pub async fn git_authed(dir: &Path, args: &[&str], auth_header: &str) -> Result< /// 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 = git_command() - .current_dir(dir) - .args(["rev-parse", refname]) - .output() - .await - .with_context(|| format!("git rev-parse {refname} in {}", dir.display()))?; + 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?; if !out.status.success() { bail!( "git rev-parse {refname} failed ({}): {}", @@ -155,12 +191,14 @@ 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 = git_command() - .current_dir(dir) - .args(["ls-tree", "--name-only", rev, "--", path]) - .output() - .await - .with_context(|| format!("git ls-tree {spec} in {}", dir.display()))?; + 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?; if !listed.status.success() { bail!( "git ls-tree {spec} failed ({}): {}", @@ -171,12 +209,12 @@ pub async fn git_show_file(dir: &Path, rev: &str, path: &str) -> Result {}; exec sleep 30", pidfile.display())); + cmd.arg("-c").arg(format!( + "echo $$ > {}; sleep 300 & echo $! > {}; wait", + shpid.display(), + sleeppid.display() + )); let err = run_with_deadline( &mut cmd, std::time::Duration::from_millis(500), @@ -493,18 +497,23 @@ async fn run_with_deadline_times_out_and_kills_the_child() { 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; + 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; } - tokio::time::sleep(std::time::Duration::from_millis(50)).await; + assert!(gone, "pid {pid} ({}) still running", file.display()); } - panic!("child {pid} still running after the deadline dropped its future"); } /// Presence control for the timeout test: a child that finishes in time diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index 064f6e90..4333a208 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -81,10 +81,14 @@ async fn lock_notifying<'a>( 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 { - match tokio::time::timeout(interval, lock.lock()).await { - Ok(guard) => break guard, - Err(_) => on_stall( + tokio::select! { + guard = &mut acquire => break guard, + () = tokio::time::sleep(interval) => on_stall( *holder.lock().unwrap_or_else(PoisonError::into_inner), start.elapsed(), ), @@ -2202,6 +2206,39 @@ mod lock_wait_tests { 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]