meta lock waiters keep their queue place; deadlines cover pin_agent git and kill the whole process group

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
+Each git or nix child under `META_LOCK` has a deadline, so a hung forge fetch
-forge fetch fails the operation instead of holding the lock: 10 minutes for
+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
+`nix flake update`/`lock` and forge fetch/push, 2 minutes for local git, including
-names the command and the deadline, and the child dies with it. A caller waiting on
+the `rev-parse` and `show` reads `pin_agents` makes. The error names the command
-`META_LOCK` for over a minute logs a `meta_lock: still waiting` warning every
+and the deadline. Each child leads its own process group and the timeout
-minute, naming its own operation and the holder's.
+`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.
 
 ---
 

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
+/// passes. `what` names the command in the error. The child leads its own
-/// deadline drops its future, so a hung remote cannot leave an orphan behind.
+/// 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<Output> {
-    // `kill_on_drop` is what turns the timeout's drop into a kill.
+    // `kill_on_drop` covers the caller dropping this future; `process_group(0)`
-    cmd.kill_on_drop(true);
+    // makes the child's pid the group id the timeout path kills.
-    tokio::time::timeout(deadline, cmd.output())
+    let mut child = cmd
-        .await
+        .stdin(Stdio::null())
-        .map_err(|_| anyhow::anyhow!("{what} timed out after {}s", deadline.as_secs_f64()))?
+        .stdout(Stdio::piped())
-        .with_context(|| what.to_owned())
+        .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<R: AsyncRead + Unpin>(pipe: Option<R>) -> std::io::Result<Vec<u8>> {
+    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<String> {
-    let out = git_command()
+    let out = run_with_deadline(
-        .current_dir(dir)
+        git_command().current_dir(dir).args(["rev-parse", refname]),
-        .args(["rev-parse", refname])
+        GIT_LOCAL_DEADLINE,
-        .output()
+        &format!("git rev-parse {refname} in {}", dir.display()),
-        .await
+    )
-        .with_context(|| 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<String> {
 /// such file. Reads the object database, not the working tree.
 pub async fn git_show_file(dir: &Path, rev: &str, path: &str) -> Result<Option<String>> {
     let spec = format!("{rev}:{path}");
-    let listed = git_command()
+    let listed = run_with_deadline(
+        git_command()
             .current_dir(dir)
-        .args(["ls-tree", "--name-only", rev, "--", path])
+            .args(["ls-tree", "--name-only", rev, "--", path]),
-        .output()
+        GIT_LOCAL_DEADLINE,
-        .await
+        &format!("git ls-tree {spec} in {}", dir.display()),
-        .with_context(|| 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<Option<S
     if listed.stdout.iter().all(u8::is_ascii_whitespace) {
         return Ok(None);
     }
-    let shown = git_command()
+    let shown = run_with_deadline(
-        .current_dir(dir)
+        git_command().current_dir(dir).args(["show", &spec]),
-        .args(["show", &spec])
+        GIT_LOCAL_DEADLINE,
-        .output()
+        &format!("git show {spec} in {}", dir.display()),
-        .await
+    )
-        .with_context(|| format!("git show {spec} in {}", dir.display()))?;
+    .await?;
     if !shown.status.success() {
         bail!(
             "git show {spec} failed ({}): {}",

hive-c0re/src/lifecycle/tests.rs

@@ -475,14 +475,18 @@ async fn a_systemctl_answer_inside_the_bound_passes_through() {
 }
 
 /// A child that outlives the deadline yields an error naming the command and
-/// the deadline, and is killed rather than orphaned.
+/// the deadline, and the child and its grandchild are both killed rather than
+/// orphaned.
 #[tokio::test]
-async fn run_with_deadline_times_out_and_kills_the_child() {
+async fn run_with_deadline_times_out_and_kills_the_process_group() {
     let dir = tempfile::tempdir().expect("tempdir");
-    let pidfile = dir.path().join("pid");
+    let (shpid, sleeppid) = (dir.path().join("sh.pid"), dir.path().join("sleep.pid"));
     let mut cmd = tokio::process::Command::new("sh");
-    cmd.arg("-c")
+    cmd.arg("-c").arg(format!(
-        .arg(format!("echo $$ > {}; exec sleep 30", pidfile.display()));
+        "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)
+    for file in [&shpid, &sleeppid] {
-        .expect("child wrote its pid before the deadline")
+        // 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() {
-            return;
+                gone = true;
+                break;
             }
             tokio::time::sleep(std::time::Duration::from_millis(50)).await;
         }
-    panic!("child {pid} still running after the deadline dropped its future");
+        assert!(gone, "pid {pid} ({}) still running", file.display());
+    }
 }
 
 /// Presence control for the timeout test: a child that finishes in time

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 {
+        tokio::select! {
-            Ok(guard) => break guard,
+            guard = &mut acquire => break guard,
-            Err(_) => on_stall(
+            () = 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]