From c29d3eec4c30a6159fe250f8f86c2d52d1e5a1f3 Mon Sep 17 00:00:00 2001 From: Calvin Grunewald Date: Sat, 31 Oct 2026 19:51:57 -0700 Subject: [PATCH] factory: Next idle epoch removes family-* dirs whose owner.lock is not held --- src/executor/admission.rs | 337 ++++++++++++++++++++++++++++++++++++ tests/execution_capacity.rs | 251 ++++++++++++++++++++++++++- 2 files changed, 585 insertions(+), 3 deletions(-) diff --git a/src/executor/admission.rs b/src/executor/admission.rs index 3efd811..5caad0d 100644 --- a/src/executor/admission.rs +++ b/src/executor/admission.rs @@ -1,5 +1,8 @@ //! Host execution admission for the CLI. File locks, not PID files or clocks, //! own capacity. An exec'd command supervisor retains them if its scheduler dies. +//! The authority holder removes family directories whose owner and borrow locks +//! are free. A family that has not created those locks yet is kept until the +//! directory is older than ten minutes, so a concurrent acquire is not deleted. use crate::errln; use anyhow::{bail, Context, Result}; @@ -17,6 +20,9 @@ use super::admission_order::Ticket; const CONTEXT_ENV: &str = "ASTER_EXECUTION_CONTEXT"; const SPEC_ENV: &str = "ASTER_EXECUTION_SPEC"; const INTERNAL_COMMAND: &str = "__aster_execute"; +/// Longer than the window in which acquire creates a family directory and its +/// lock files, and short enough that a killed acquire does not leak forever. +const STALE_FAMILY_GRACE: Duration = Duration::from_secs(10 * 60); static EXECUTABLE: OnceLock = OnceLock::new(); /// Enable host admission when embedding the executor in an executable which @@ -60,6 +66,120 @@ fn try_lock(file: &File) -> Result { const POLL: Duration = Duration::from_millis(20); +fn open_existing(path: &Path) -> std::io::Result> { + match OpenOptions::new().read(true).write(true).open(path) { + Ok(file) => Ok(Some(file)), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => Err(error), + } +} + +enum LockPresence { + Missing, + Free, + Held, +} + +/// `None` means skip this family: the lock file could not be inspected. +/// A free lock is pushed onto `held` and stays locked until that vector is dropped. +fn inspect_lock(path: &Path, held: &mut Vec) -> Option { + let Some(file) = open_existing(path).ok()? else { + return Some(LockPresence::Missing); + }; + match try_lock(&file) { + Ok(true) => { + held.push(file); + Some(LockPresence::Free) + } + Ok(false) => Some(LockPresence::Held), + Err(_) => None, + } +} + +fn directory_older_than_grace(path: &Path) -> bool { + let Ok(modified) = fs::metadata(path).and_then(|metadata| metadata.modified()) else { + return false; + }; + modified.elapsed().is_ok_and(|age| age > STALE_FAMILY_GRACE) +} + +fn reclaim_stale_family(path: &Path) -> Result<()> { + // Hold owner across the borrow check and the removal. Dropping it first + // lets acquire lock owner and create an unlocked borrow, which then looks + // dead. Acquire does not hold authority.lock. + let mut held = Vec::new(); + let Some(owner) = inspect_lock(&path.join("owner.lock"), &mut held) else { + return Ok(()); + }; + if matches!(owner, LockPresence::Held) { + return Ok(()); + } + let Some(borrow) = inspect_lock(&path.join("borrow.lock"), &mut held) else { + return Ok(()); + }; + if matches!(borrow, LockPresence::Held) { + return Ok(()); + } + let both_free = matches!((owner, borrow), (LockPresence::Free, LockPresence::Free)); + // A missing lock file is not immediate proof of death: acquire creates the + // directory before either file. Keep it inside the grace period. + if !both_free && !directory_older_than_grace(path) { + return Ok(()); + } + match fs::remove_dir_all(path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error) + .with_context(|| format!("remove stale execution family {}", path.display())) + } + } + drop(held); + Ok(()) +} + +fn sweep_stale_families(directory: &Path) -> Result<()> { + let entries = match fs::read_dir(directory) { + Ok(entries) => entries, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()), + Err(error) => { + return Err(error) + .with_context(|| format!("read execution directory {}", directory.display())) + } + }; + for entry in entries { + let entry = match entry { + Ok(entry) => entry, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => { + return Err(error) + .with_context(|| format!("read execution directory {}", directory.display())) + } + }; + let name = entry.file_name(); + let Some(name) = name.to_str() else { + continue; + }; + if !name.starts_with("family-") { + continue; + } + let path = entry.path(); + let file_type = match entry.file_type() { + Ok(file_type) => file_type, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, + Err(error) => { + return Err(error) + .with_context(|| format!("stat execution family {}", path.display())) + } + }; + if !file_type.is_dir() { + continue; + } + reclaim_stale_family(&path)?; + } + Ok(()) +} + fn wait() -> Result<()> { if super::shutdown_requested() { bail!("Cancelled: shutdown requested while waiting for execution admission"); @@ -123,6 +243,11 @@ impl Host { } let epoch = open_lock(&directory.join("epoch.lock"))?; let idle = try_lock(&epoch)?; + // Lock tests decide liveness, so this runs on a busy epoch too. Hold + // authority.lock across the sweep so a new invocation cannot pass + // configured() until it finishes. An acquire already past that point + // does not hold this lock; missing lock files use the grace period. + sweep_stale_families(&directory)?; let settings = directory.join("capacity"); let previous = match fs::read_to_string(&settings) { Ok(text) => Some( @@ -577,3 +702,215 @@ fn supervise() -> Result { .code() .unwrap_or_else(|| 128 + status.signal().unwrap_or(1))) } + +#[cfg(test)] +mod tests { + use super::*; + use std::io::Read; + use std::process::{Command, Stdio}; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + use std::time::SystemTime; + + fn host() -> tempfile::TempDir { + tempfile::tempdir().unwrap() + } + + fn family(dir: &Path, name: &str) -> PathBuf { + let path = dir.join(name); + fs::create_dir(&path).unwrap(); + path + } + + fn touch_locks(path: &Path) { + open_lock(&path.join("owner.lock")).unwrap(); + open_lock(&path.join("borrow.lock")).unwrap(); + } + + fn set_directory_mtime(path: &Path, time: SystemTime) { + File::open(path) + .unwrap_or_else(|error| panic!("open {}: {error}", path.display())) + .set_modified(time) + .unwrap_or_else(|error| panic!("set mtime {}: {error}", path.display())); + } + + /// Holds an exclusive flock in another process. Same-process flock does not + /// conflict on every Unix, and production holders are other processes. + struct HeldLock(std::process::Child); + + impl HeldLock { + fn exclusive(path: &Path) -> Self { + let mut child = Command::new("python3") + .arg("-u") + .arg("-c") + .arg( + "import fcntl, os, sys, time\n\ +fd = os.open(sys.argv[1], os.O_RDWR)\n\ +fcntl.flock(fd, fcntl.LOCK_EX)\n\ +sys.stdout.buffer.write(b'held\\n')\n\ +sys.stdout.buffer.flush()\n\ +time.sleep(120)\n", + ) + .arg(path) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .spawn() + .expect("python3 is required to hold a lock from another process"); + let mut stdout = child.stdout.take().unwrap(); + let mut buf = [0; 5]; + stdout + .read_exact(&mut buf) + .expect("lock holder did not acquire the lock"); + assert_eq!(&buf, b"held\n"); + Self(child) + } + } + + impl Drop for HeldLock { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } + } + + #[test] + fn sweep_removes_dead_families_and_keeps_held_locks() { + let dir = host(); + let dead = family(dir.path(), "family-dead"); + touch_locks(&dead); + let owned = family(dir.path(), "family-owned"); + touch_locks(&owned); + let borrowed = family(dir.path(), "family-borrowed"); + touch_locks(&borrowed); + let owner_only = family(dir.path(), "family-owner-only"); + open_lock(&owner_only.join("owner.lock")).unwrap(); + set_directory_mtime( + &owner_only, + SystemTime::now() - Duration::from_secs(11 * 60), + ); + let _owner = HeldLock::exclusive(&owned.join("owner.lock")); + let _borrow = HeldLock::exclusive(&borrowed.join("borrow.lock")); + let _owner_only = HeldLock::exclusive(&owner_only.join("owner.lock")); + + sweep_stale_families(dir.path()).unwrap(); + + assert!(!dead.exists(), "dead family was kept"); + assert!(owned.exists(), "live owner.lock was removed"); + assert!(borrowed.exists(), "live borrow.lock was removed"); + assert!( + owner_only.exists(), + "held owner.lock was removed because borrow.lock was missing and old" + ); + } + + #[test] + fn sweep_keeps_a_recent_family_without_lock_files_until_the_grace_period() { + let dir = host(); + let fresh = family(dir.path(), "family-fresh"); + let partial = family(dir.path(), "family-partial"); + open_lock(&partial.join("owner.lock")).unwrap(); + sweep_stale_families(dir.path()).unwrap(); + assert!(fresh.exists(), "family without lock files was removed"); + assert!( + partial.exists(), + "family whose borrow.lock was not created yet was removed" + ); + + let stale_at = SystemTime::now() - Duration::from_secs(11 * 60); + set_directory_mtime(&fresh, stale_at); + set_directory_mtime(&partial, stale_at); + sweep_stale_families(dir.path()).unwrap(); + assert!( + !fresh.exists(), + "family without lock files survived the grace period" + ); + assert!( + !partial.exists(), + "old family with a missing lock file was kept" + ); + } + + #[test] + fn sweep_leaves_admission_lock_files_untouched() { + let dir = host(); + let dead = family(dir.path(), "family-dead"); + touch_locks(&dead); + let kept = [ + "slot-0.lock", + "slot-1.lock", + "epoch.lock", + "authority.lock", + "resource-abc.lock", + "capacity", + "family-not-a-dir", + ]; + for name in kept { + fs::write(dir.path().join(name), name).unwrap(); + } + + sweep_stale_families(dir.path()).unwrap(); + + assert!(!dead.exists()); + for name in kept { + let path = dir.path().join(name); + assert_eq!( + fs::read(&path).unwrap_or_else(|error| panic!("read {name}: {error}")), + name.as_bytes(), + "{name} was removed or rewritten" + ); + } + } + + #[test] + fn two_threads_wait_for_authority_lock_before_sweeping() { + let dir = host(); + let dead = family(dir.path(), "family-dead"); + touch_locks(&dead); + let authority = dir.path().join("authority.lock"); + open_lock(&authority).unwrap(); + let holder = HeldLock::exclusive(&authority); + let entered = Arc::new(AtomicUsize::new(0)); + let blocked = Arc::new(AtomicUsize::new(0)); + let mut handles = Vec::new(); + for _ in 0..2 { + let path = dir.path().to_path_buf(); + let entered = Arc::clone(&entered); + let blocked = Arc::clone(&blocked); + handles.push(std::thread::spawn(move || { + let authority = open_lock(&path.join("authority.lock")).unwrap(); + entered.fetch_add(1, Ordering::SeqCst); + let mut reported = false; + while !try_lock(&authority).unwrap() { + if !reported { + blocked.fetch_add(1, Ordering::SeqCst); + reported = true; + } + std::thread::sleep(Duration::from_millis(20)); + } + sweep_stale_families(&path).unwrap(); + drop(authority); + })); + } + let deadline = std::time::Instant::now() + Duration::from_secs(2); + while blocked.load(Ordering::SeqCst) < 2 { + assert!( + std::time::Instant::now() < deadline, + "sweep threads did not block on authority.lock (started {})", + entered.load(Ordering::SeqCst) + ); + std::thread::sleep(Duration::from_millis(10)); + } + assert!( + dead.exists(), + "sweep removed a family while another process held authority.lock" + ); + drop(holder); + for handle in handles { + handle.join().unwrap(); + } + assert!( + !dead.exists(), + "sweep did not run after authority.lock was released" + ); + } +} diff --git a/tests/execution_capacity.rs b/tests/execution_capacity.rs index c66b321..23277ea 100644 --- a/tests/execution_capacity.rs +++ b/tests/execution_capacity.rs @@ -6,9 +6,10 @@ //! Integration matrix: run/affected/streaming; one/multiple ready targets; one/two //! workspaces and invocations; equal/different host keys; same/different workspace //! keys; cold/warm cache; normal/failure/waiting cancellation/scheduler SIGKILL; -//! recursive nested siblings; workspace/state symlink aliases; inherited/new -//! resources; captured/streaming first/second interruption of stubborn children; -//! conflicting `--jobs` in human/quiet/json, plus a same-jobs control. +//! every admitted group SIGKILLed (stale family cleanup); recursive nested siblings; +//! workspace/state symlink aliases; inherited/new resources; captured/streaming +//! first/second interruption of stubborn children; conflicting `--jobs` in +//! human/quiet/json, plus a same-jobs control. //! Fidelity: production CLI, scheduler, flock, process groups, Git affected //! selection, and socket barriers. Python commands replace expensive compilers, //! not the admission or lifecycle code. The test closes their inherited FDs. @@ -1108,3 +1109,247 @@ fn conflicting_jobs_failure_names_the_reason_in_human_output() { assert!(same.finish(), "same-jobs run failed after a slot freed"); assert!(holder.finish(), "holder failed after releasing its targets"); } + +fn child_pids(parent: u32) -> Vec { + let output = Command::new("ps") + .args(["-axo", "pid=,ppid="]) + .output() + .expect("ps"); + String::from_utf8(output.stdout) + .unwrap() + .lines() + .filter_map(|line| { + let mut parts = line.split_whitespace(); + let pid = parts.next()?.parse::().ok()?; + let ppid = parts.next()?.parse::().ok()?; + (ppid == parent).then_some(pid) + }) + .collect() +} + +fn process_group(pid: u32) -> i32 { + unsafe { libc::getpgid(pid as i32) } +} + +fn sigkill_not_self_group(pid: u32) { + let group = process_group(pid); + let mine = unsafe { libc::getpgrp() }; + if group > 1 && group != mine { + unsafe { + libc::kill(-group, libc::SIGKILL); + } + } + unsafe { + libc::kill(pid as i32, libc::SIGKILL); + } +} + +fn wait_until_dead(pids: &[u32]) { + let deadline = Instant::now() + BUDGET; + for pid in pids { + loop { + let alive = unsafe { libc::kill(*pid as i32, 0) } == 0; + if !alive || process_state(*pid).contains('Z') { + break; + } + assert!(Instant::now() < deadline, "pid {pid} survived SIGKILL"); + std::thread::sleep(Duration::from_millis(10)); + } + } +} + +fn process_state(pid: u32) -> String { + let output = Command::new("ps") + .args(["-o", "stat=", "-p", &pid.to_string()]) + .output(); + output + .ok() + .map(|output| String::from_utf8_lossy(&output.stdout).trim().to_string()) + .unwrap_or_default() +} + +fn lock_is_free(path: &Path) -> bool { + use std::os::unix::io::AsRawFd; + let Ok(file) = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + else { + return false; + }; + unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) == 0 } +} + +fn wait_until_free(path: &Path) { + let deadline = Instant::now() + BUDGET; + loop { + if lock_is_free(path) { + return; + } + assert!( + path.exists(), + "lock disappeared before it was free: {}", + path.display() + ); + assert!( + Instant::now() < deadline, + "lock stayed held: {}", + path.display() + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +fn family_dirs(host: &Path) -> Vec { + let mut dirs = fs::read_dir(host) + .unwrap() + .flatten() + .filter(|entry| { + entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with("family-")) + && entry.file_type().is_ok_and(|kind| kind.is_dir()) + }) + .map(|entry| entry.path()) + .collect::>(); + dirs.sort(); + dirs +} + +/// Canonical proof that a killed admitted run does not leak its family directory. +/// +/// Fidelity: real `aster work`, real flocks, real SIGKILL of the scheduler, +/// supervisor, and target groups. The next invocation on the same +/// ASTER_EXECUTION_DIR removes the stale family and leaves a concurrently live +/// run's family in place. Not proven here: NFS locks. +#[test] +fn killed_admitted_run_leaves_no_family_dir_after_next_invocation() { + let f = Fixture::new(); + let killed_ws = f.workspace("killed", "", "normal"); + let live_ws = f.workspace("live", "", "normal"); + let next_ws = f.workspace("next", "", "normal"); + let host = f.root.path().join("host"); + let work = |root: &Path| { + let mut command = f.command(root, 1); + command.args(["work", ".", "--quiet"]); + Run::spawn(command) + }; + + let mut killed = work(&killed_ws); + let killed_target = f.started(); + assert_eq!(killed_target.label, "killed"); + let stale = { + let families = family_dirs(&host); + assert_eq!( + families.len(), + 1, + "admitted run did not create one family dir" + ); + families.into_iter().next().unwrap() + }; + + // Reach configured() while the first family is still live, then wait. This + // invocation must not be the one that sweeps after the kill. + let mut live = work(&live_ws); + live.waiting(&f.listener); + + let scheduler = killed.0.id(); + let supervisors = child_pids(scheduler); + assert!( + !supervisors.is_empty(), + "execution supervisor was not spawned" + ); + let mut targets = Vec::new(); + for supervisor in &supervisors { + targets.extend(child_pids(*supervisor)); + } + assert!(!targets.is_empty(), "admitted target was not spawned"); + let supervisor = supervisors[0]; + let target = targets[0]; + assert_ne!( + process_group(supervisor), + process_group(target), + "target was not in its own process group" + ); + assert_ne!( + process_group(supervisor), + unsafe { libc::getpgrp() }, + "supervisor stayed in the test process group" + ); + + // Kill the scheduler before its supervisor. A live scheduler would drop the + // family directory itself; a live supervisor would remove it after the target + // died. Neither destructor may run. + assert_eq!(unsafe { libc::kill(scheduler as i32, libc::SIGKILL) }, 0); + for pid in &supervisors { + sigkill_not_self_group(*pid); + } + for pid in &targets { + sigkill_not_self_group(*pid); + } + let _ = killed.0.wait(); + let mut killed_pids = vec![scheduler]; + killed_pids.extend(supervisors); + killed_pids.extend(targets); + wait_until_dead(&killed_pids); + assert!( + stale.is_dir(), + "killed run's family dir was removed before the next invocation" + ); + wait_until_free(&stale.join("owner.lock")); + wait_until_free(&stale.join("borrow.lock")); + assert!( + stale.is_dir(), + "family dir disappeared while its locks were released" + ); + + let live_target = f.started(); + assert_eq!(live_target.label, "live"); + let families = family_dirs(&host); + assert!( + families.contains(&stale), + "the invocation that was already waiting swept the killed family: {families:?}" + ); + let live_family = families + .iter() + .find(|path| *path != &stale) + .cloned() + .expect("live run did not create a family dir"); + + let preserved = ["authority.lock", "epoch.lock", "capacity", "slot-0.lock"]; + let before: Vec<_> = preserved + .iter() + .map(|name| { + let path = host.join(name); + ( + *name, + fs::read(&path).unwrap_or_else(|error| panic!("read {name}: {error}")), + ) + }) + .collect::>(); + + let mut next = work(&next_ws); + next.waiting(&f.listener); + assert!(!stale.exists(), "next invocation left the stale family dir"); + assert!( + live_family.is_dir(), + "next invocation removed a live run's family dir" + ); + for (name, bytes) in before { + let path = host.join(name); + assert!(path.exists(), "{name} was removed by the sweep"); + assert_eq!( + fs::read(&path).unwrap(), + bytes, + "{name} was rewritten by the sweep" + ); + } + + live_target.release(); + assert!(live.finish()); + let next_target = f.started(); + assert_eq!(next_target.label, "next"); + next_target.release(); + assert!(next.finish()); +}