Compare commits
6 changed files with 76 additions and 375 deletions
|
|
@ -358,16 +358,6 @@ Every public `meta.rs` operation takes the module's internal `META_LOCK`
|
||||||
mutex, so concurrent job-queue nodes never
|
mutex, so concurrent job-queue nodes never
|
||||||
race on the repo's `.git/index.lock`.
|
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`)
|
## Container lifecycle (`lifecycle.rs`)
|
||||||
|
|
|
||||||
|
|
@ -111,14 +111,12 @@ pub async fn push_meta(dir: &Path) -> Result<()> {
|
||||||
// raise a dashboard warning and leave the remote intact rather than
|
// raise a dashboard warning and leave the remote intact rather than
|
||||||
// silently erasing its history.
|
// silently erasing its history.
|
||||||
let url = forge_git_url("core/meta");
|
let url = forge_git_url("core/meta");
|
||||||
let out = crate::lifecycle::run_with_deadline(
|
let out = crate::lifecycle::git_command_authed(&core_auth_header(&token))
|
||||||
crate::lifecycle::git_command_authed(&core_auth_header(&token))
|
.current_dir(dir)
|
||||||
.current_dir(dir)
|
.args(["push", &url, "HEAD:main"])
|
||||||
.args(["push", &url, "HEAD:main"]),
|
.output()
|
||||||
crate::lifecycle::FETCH_DEADLINE,
|
.await
|
||||||
"git push core/meta",
|
.context("invoke git push core/meta")?;
|
||||||
)
|
|
||||||
.await?;
|
|
||||||
if !out.status.success() {
|
if !out.status.success() {
|
||||||
let stderr = String::from_utf8_lossy(&out.stderr);
|
let stderr = String::from_utf8_lossy(&out.stderr);
|
||||||
if stderr.contains("non-fast-forward") || stderr.contains("fetch first") {
|
if stderr.contains("non-fast-forward") || stderr.contains("fetch first") {
|
||||||
|
|
|
||||||
|
|
@ -3,72 +3,10 @@
|
||||||
//! fetch proposal commits into the applied repo.
|
//! fetch proposal commits into the applied repo.
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::process::{Output, Stdio};
|
|
||||||
use std::time::Duration;
|
|
||||||
|
|
||||||
use anyhow::{Context, Result, bail};
|
use anyhow::{Context, Result, bail};
|
||||||
use tokio::io::{AsyncRead, AsyncReadExt};
|
|
||||||
use tokio::process::Command;
|
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_NAME: &str = "c0re";
|
||||||
const GIT_EMAIL: &str = "c0re@hyperhive.local";
|
const GIT_EMAIL: &str = "c0re@hyperhive.local";
|
||||||
|
|
||||||
|
|
@ -142,11 +80,14 @@ pub fn git_command_authed(auth_header: &str) -> Command {
|
||||||
cmd
|
cmd
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Run `cmd` as `git <args>` in `dir` within `deadline`, erroring on a
|
/// Run `cmd` as `git <args>` in `dir`, erroring on a non-zero exit.
|
||||||
/// non-zero exit.
|
async fn run(mut cmd: Command, dir: &Path, args: &[&str]) -> Result<()> {
|
||||||
async fn run(mut cmd: Command, dir: &Path, args: &[&str], deadline: Duration) -> Result<()> {
|
let out = cmd
|
||||||
let what = format!("git {} in {}", args.join(" "), dir.display());
|
.current_dir(dir)
|
||||||
let out = run_with_deadline(cmd.current_dir(dir).args(args), deadline, &what).await?;
|
.args(args)
|
||||||
|
.output()
|
||||||
|
.await
|
||||||
|
.with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?;
|
||||||
if !out.status.success() {
|
if !out.status.success() {
|
||||||
bail!(
|
bail!(
|
||||||
"git {} failed ({}): {}",
|
"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<()> {
|
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
|
/// [`git`] against an authenticated remote — for the paths that talk to the
|
||||||
/// forge. The credential rides the environment, so `args` (and therefore the
|
/// forge. The credential rides the environment, so `args` (and therefore the
|
||||||
/// error above) stay free of it.
|
/// error above) stay free of it.
|
||||||
pub async fn git_authed(dir: &Path, args: &[&str], auth_header: &str) -> Result<()> {
|
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.
|
/// 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> {
|
pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result<String> {
|
||||||
let out = run_with_deadline(
|
let out = git_command()
|
||||||
git_command().current_dir(dir).args(["rev-parse", refname]),
|
.current_dir(dir)
|
||||||
GIT_LOCAL_DEADLINE,
|
.args(["rev-parse", refname])
|
||||||
&format!("git rev-parse {refname} in {}", dir.display()),
|
.output()
|
||||||
)
|
.await
|
||||||
.await?;
|
.with_context(|| format!("git rev-parse {refname} in {}", dir.display()))?;
|
||||||
if !out.status.success() {
|
if !out.status.success() {
|
||||||
bail!(
|
bail!(
|
||||||
"git rev-parse {refname} failed ({}): {}",
|
"git rev-parse {refname} failed ({}): {}",
|
||||||
|
|
@ -191,14 +132,12 @@ pub async fn git_rev_parse(dir: &Path, refname: &str) -> Result<String> {
|
||||||
/// such file. Reads the object database, not the working tree.
|
/// 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>> {
|
pub async fn git_show_file(dir: &Path, rev: &str, path: &str) -> Result<Option<String>> {
|
||||||
let spec = format!("{rev}:{path}");
|
let spec = format!("{rev}:{path}");
|
||||||
let listed = run_with_deadline(
|
let listed = git_command()
|
||||||
git_command()
|
.current_dir(dir)
|
||||||
.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() {
|
if !listed.status.success() {
|
||||||
bail!(
|
bail!(
|
||||||
"git ls-tree {spec} failed ({}): {}",
|
"git ls-tree {spec} failed ({}): {}",
|
||||||
|
|
@ -209,12 +148,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) {
|
if listed.stdout.iter().all(u8::is_ascii_whitespace) {
|
||||||
return Ok(None);
|
return Ok(None);
|
||||||
}
|
}
|
||||||
let shown = run_with_deadline(
|
let shown = git_command()
|
||||||
git_command().current_dir(dir).args(["show", &spec]),
|
.current_dir(dir)
|
||||||
GIT_LOCAL_DEADLINE,
|
.args(["show", &spec])
|
||||||
&format!("git show {spec} in {}", dir.display()),
|
.output()
|
||||||
)
|
.await
|
||||||
.await?;
|
.with_context(|| format!("git show {spec} in {}", dir.display()))?;
|
||||||
if !shown.status.success() {
|
if !shown.status.success() {
|
||||||
bail!(
|
bail!(
|
||||||
"git show {spec} failed ({}): {}",
|
"git show {spec} failed ({}): {}",
|
||||||
|
|
|
||||||
|
|
@ -8,9 +8,8 @@ mod setup;
|
||||||
mod tests;
|
mod tests;
|
||||||
|
|
||||||
pub use git::{
|
pub use git::{
|
||||||
FETCH_DEADLINE, GIT_LOCAL_DEADLINE, git, git_authed, git_command, git_command_authed,
|
git, git_authed, git_command, git_command_authed, git_is_ancestor, git_read_tree_reset,
|
||||||
git_is_ancestor, git_read_tree_reset, git_rev_parse, git_show_file, git_tag, git_update_ref,
|
git_rev_parse, git_show_file, git_tag, git_update_ref, git_update_ref_cas,
|
||||||
git_update_ref_cas, run_with_deadline,
|
|
||||||
};
|
};
|
||||||
pub use host_config::{hand_permissions, write_dropins};
|
pub use host_config::{hand_permissions, write_dropins};
|
||||||
pub use setup::{
|
pub use setup::{
|
||||||
|
|
|
||||||
|
|
@ -473,57 +473,3 @@ async fn a_systemctl_answer_inside_the_bound_passes_through() {
|
||||||
.await;
|
.await;
|
||||||
assert_eq!(got, Some(7));
|
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");
|
|
||||||
}
|
|
||||||
|
|
|
||||||
|
|
@ -5,13 +5,10 @@
|
||||||
//! `docs/agent-lifecycle/config-changes.md::Meta flake`.
|
//! `docs/agent-lifecycle/config-changes.md::Meta flake`.
|
||||||
|
|
||||||
use std::path::Path;
|
use std::path::Path;
|
||||||
use std::sync::{Mutex as StdMutex, PoisonError};
|
|
||||||
use std::time::Duration;
|
|
||||||
|
|
||||||
use anyhow::{Context, Result, bail};
|
use anyhow::{Context, Result, bail};
|
||||||
use tokio::process::Command;
|
use tokio::process::Command;
|
||||||
use tokio::sync::{Mutex, MutexGuard};
|
use tokio::sync::Mutex;
|
||||||
use tokio::time::Instant;
|
|
||||||
|
|
||||||
use crate::coordinator::HiveEnv;
|
use crate::coordinator::HiveEnv;
|
||||||
use crate::lifecycle;
|
use crate::lifecycle;
|
||||||
|
|
@ -30,77 +27,6 @@ const GIT_EMAIL: &str = "c0re@hyperhive.local";
|
||||||
/// take turns instead of colliding.
|
/// take turns instead of colliding.
|
||||||
static META_LOCK: Mutex<()> = Mutex::const_new(());
|
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`
|
// Exclusivity for meta-repo *windows* that span multiple `META_LOCK`
|
||||||
// acquisitions is **not** a mutex in this module. `META_LOCK` above
|
// acquisitions is **not** a mutex in this module. `META_LOCK` above
|
||||||
// serializes individual git ops but cannot keep another op out of a window
|
// 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
|
/// rendered contents differ from disk; an unchanged `flake.nix` is a
|
||||||
/// no-op.
|
/// no-op.
|
||||||
pub async fn sync_agents(hive: &HiveEnv, agents: &[AgentSpec]) -> Result<()> {
|
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
|
// 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
|
// whole fleet of agents that silently never log in, because the agent
|
||||||
// option treats "unset" as "no forge configured" rather than erroring.
|
// 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
|
/// (no separate finalize) because rebuild has no failure-revert
|
||||||
/// semantics. The input follows the url `sync_agents` rendered just before.
|
/// semantics. The input follows the url `sync_agents` rendered just before.
|
||||||
pub async fn lock_update_for_rebuild(name: &str) -> Result<()> {
|
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 dir = crate::paths::meta_root();
|
||||||
let input = format!("agent-{name}");
|
let input = format!("agent-{name}");
|
||||||
nix(&dir, &rebuild_relock_args(&input)).await?;
|
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
|
/// real input keys (e.g. via `inputs_view()` snapshotted from the lock
|
||||||
/// file) when targeting specific inputs.
|
/// file) when targeting specific inputs.
|
||||||
pub async fn lock_update(inputs: &[String]) -> Result<()> {
|
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 dir = crate::paths::meta_root();
|
||||||
let mut args: Vec<&str> = vec!["flake", "update"];
|
let mut args: Vec<&str> = vec!["flake", "update"];
|
||||||
for i in inputs {
|
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`
|
/// rev, commit if the lock changed. Cheaper than `sync_agents`
|
||||||
/// because the per-agent inputs aren't touched.
|
/// because the per-agent inputs aren't touched.
|
||||||
pub async fn lock_update_hyperhive() -> Result<()> {
|
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();
|
let dir = crate::paths::meta_root();
|
||||||
nix(&dir, &["flake", "update", "hyperhive"]).await?;
|
nix(&dir, &["flake", "update", "hyperhive"]).await?;
|
||||||
if !paths_dirty(&dir, &["flake.lock"]).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.
|
/// Removing a file or a git stage/commit step fails.
|
||||||
pub async fn remove_permission_files() -> Result<()> {
|
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 dir = crate::paths::meta_root();
|
||||||
let mut removed = Vec::new();
|
let mut removed = Vec::new();
|
||||||
for file in crate::workers::permissions_import::META_FILES {
|
for file in crate::workers::permissions_import::META_FILES {
|
||||||
|
|
@ -998,11 +924,12 @@ fn render_flake(
|
||||||
async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
|
async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
|
||||||
// `--diff-filter=ACM` skips deleted entries — we only care about
|
// `--diff-filter=ACM` skips deleted entries — we only care about
|
||||||
// additions and modifications for message-building purposes.
|
// additions and modifications for message-building purposes.
|
||||||
let out = git_output(
|
let out = lifecycle::git_command()
|
||||||
dir,
|
.current_dir(dir)
|
||||||
&["diff", "--cached", "--name-only", "--diff-filter=ACM"],
|
.args(["diff", "--cached", "--name-only", "--diff-filter=ACM"])
|
||||||
)
|
.output()
|
||||||
.await?;
|
.await
|
||||||
|
.with_context(|| format!("git diff --cached in {}", dir.display()))?;
|
||||||
if out.status.success() {
|
if out.status.success() {
|
||||||
let names = String::from_utf8_lossy(&out.stdout)
|
let names = String::from_utf8_lossy(&out.stdout)
|
||||||
.lines()
|
.lines()
|
||||||
|
|
@ -1013,7 +940,12 @@ async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
|
||||||
return Ok(names);
|
return Ok(names);
|
||||||
}
|
}
|
||||||
// Fallback for initial commit (no HEAD yet): parse `git status --porcelain`.
|
// 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)
|
let names = String::from_utf8_lossy(&st.stdout)
|
||||||
.lines()
|
.lines()
|
||||||
.filter(|l| l.starts_with("A ") || l.starts_with("M "))
|
.filter(|l| l.starts_with("A ") || l.starts_with("M "))
|
||||||
|
|
@ -1023,19 +955,13 @@ async fn git_staged_names(dir: &Path) -> Result<Vec<String>> {
|
||||||
Ok(names)
|
Ok(names)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// 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
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn git(dir: &Path, args: &[&str]) -> Result<()> {
|
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() {
|
if !out.status.success() {
|
||||||
bail!(
|
bail!(
|
||||||
"git {} failed ({}): {}",
|
"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<bool> {
|
async fn paths_dirty(dir: &Path, paths: &[&str]) -> Result<bool> {
|
||||||
let mut args = vec!["diff", "--quiet", "HEAD", "--"];
|
let mut args = vec!["diff", "--quiet", "HEAD", "--"];
|
||||||
args.extend_from_slice(paths);
|
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.
|
// Exit 0 = no differences; 1 = differences; anything else (e.g.
|
||||||
// no HEAD yet on a fresh repo) → treat as dirty so the commit
|
// no HEAD yet on a fresh repo) → treat as dirty so the commit
|
||||||
// path runs and surfaces real errors loudly.
|
// path runs and surfaces real errors loudly.
|
||||||
|
|
@ -1115,28 +1046,17 @@ fn nix_argv<'a>(args: &[&'a str]) -> Vec<&'a str> {
|
||||||
|
|
||||||
/// Run `nix <args>` in `dir` (flakes enabled), capturing combined output.
|
/// Run `nix <args>` in `dir` (flakes enabled), capturing combined output.
|
||||||
/// The core of [`nix`].
|
/// 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> {
|
async fn nix_output(dir: &Path, args: &[&str]) -> Result<std::process::Output> {
|
||||||
nix_output_with("nix".as_ref(), dir, args, lifecycle::FETCH_DEADLINE).await
|
Command::new("nix")
|
||||||
}
|
.current_dir(dir)
|
||||||
|
.args(nix_argv(args))
|
||||||
/// [`nix_output`] with the program and deadline injectable for tests.
|
// Kill the nix child if the caller's future is dropped (e.g. the
|
||||||
async fn nix_output_with(
|
// startup-migration timeout around `sync_agents` fires) rather than
|
||||||
program: &std::ffi::OsStr,
|
// orphaning it against an unreachable forge. No-op on normal exit.
|
||||||
dir: &Path,
|
.kill_on_drop(true)
|
||||||
args: &[&str],
|
.output()
|
||||||
deadline: Duration,
|
.await
|
||||||
) -> Result<std::process::Output> {
|
.with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))
|
||||||
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<()>`,
|
/// 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}"
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue