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

A hung forge fetch inside `nix flake update hyperhive` (or any other meta
git/nix child) now fails after its deadline with an error naming the command,
the child is killed, and META_LOCK, BuildSlot and MetaWindow are released. The
boot sweep already treats that failure as a warning and fans out the rebuilds.
A wait on META_LOCK past 60s logs a warning naming the waiter and the holder.

docs/scheduler/coordinator.md

@@ -358,6 +358,16 @@ 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`)

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))
+    let out = crate::lifecycle::run_with_deadline(
+        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")?;
+            .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") {

hive-c0re/src/lifecycle/git.rs

@@ -3,10 +3,72 @@
 //! 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<Output> {
+    // `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<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";
 const GIT_EMAIL: &str = "c0re@hyperhive.local";
 
@@ -80,14 +142,11 @@ pub fn git_command_authed(auth_header: &str) -> Command {
     cmd
 }
 
-/// Run `cmd` as `git <args>` 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 <args>` 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,24 +159,24 @@ 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.
 pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result<String> {
-    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 ({}): {}",
@@ -132,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])
-        .output()
-        .await
-        .with_context(|| format!("git ls-tree {spec} in {}", dir.display()))?;
+            .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 ({}): {}",
@@ -148,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()
-        .current_dir(dir)
-        .args(["show", &spec])
-        .output()
-        .await
-        .with_context(|| format!("git show {spec} in {}", dir.display()))?;
+    let shown = run_with_deadline(
+        git_command().current_dir(dir).args(["show", &spec]),
+        GIT_LOCAL_DEADLINE,
+        &format!("git show {spec} in {}", dir.display()),
+    )
+    .await?;
     if !shown.status.success() {
         bail!(
             "git show {spec} failed ({}): {}",

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::{

hive-c0re/src/lifecycle/tests.rs

@@ -473,3 +473,57 @@ 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 the child and its grandchild are both killed rather than
+/// orphaned.
+#[tokio::test]
+async fn run_with_deadline_times_out_and_kills_the_process_group() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let (shpid, sleeppid) = (dir.path().join("sh.pid"), dir.path().join("sleep.pid"));
+    let mut cmd = tokio::process::Command::new("sh");
+    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),
+        "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");
+}

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,77 @@ 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
@@ -80,7 +154,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 +289,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 +323,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 +348,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 +366,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 +998,11 @@ fn render_flake(
 async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
     // `--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 +1013,7 @@ async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
         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 +1023,19 @@ async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
     Ok(names)
 }
 
-async fn git(dir: &Path, args: &[&str]) -> Result<()> {
-    let out = lifecycle::git_command()
-        .current_dir(dir)
-        .args(args)
-        .output()
+/// Run `git <args>` in `dir` within [`lifecycle::GIT_LOCAL_DEADLINE`]; the caller
+/// reads the exit status.
+async fn git_output(dir: &Path, args: &[&str]) -> Result<std::process::Output> {
+    lifecycle::run_with_deadline(
+        lifecycle::git_command().current_dir(dir).args(args),
+        lifecycle::GIT_LOCAL_DEADLINE,
+        &format!("git {} in {}", args.join(" "), dir.display()),
+    )
     .await
-        .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
+}
+
+async fn git(dir: &Path, args: &[&str]) -> Result<()> {
+    let out = git_output(dir, args).await?;
     if !out.status.success() {
         bail!(
             "git {} failed ({}): {}",
@@ -1021,12 +1095,7 @@ async fn git_commit_paths(dir: &Path, message: &str, paths: &[&str]) -> Result<(
 async fn paths_dirty(dir: &Path, paths: &[&str]) -> Result<bool> {
     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 +1115,28 @@ fn nix_argv<'a>(args: &[&'a str]) -> Vec<&'a str> {
 
 /// Run `nix <args>` 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<std::process::Output> {
-    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()
+    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<std::process::Output> {
+    lifecycle::run_with_deadline(
+        Command::new(program).current_dir(dir).args(nix_argv(args)),
+        deadline,
+        &format!("nix {} in {}", args.join(" "), dir.display()),
+    )
     .await
-        .with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))
 }
 
 /// Turn a finished nix [`Output`](std::process::Output) into `Result<()>`,
@@ -2092,3 +2172,94 @@ 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}"
+        );
+    }
+}