From 9676789261149c7700f4c0d83dae4551b686ee58 Mon Sep 17 00:00:00 2001 From: Max Loffgren Date: Mon, 5 Oct 2026 22:30:10 -0500 Subject: [PATCH] fix(core): release every file lock at once, even while a command is starting Reopening a session right after closing it could be refused as already open in another process, and other openmax processes could wait on the session index, trust.lock, or ledger.lock with no visible holder. It happened whenever a bash call, external tool, or hook was being started at the moment of release. Each of these locks is a flock(2), which belongs to the open file description, and the harness released it only by closing its descriptor. Those children are spawned through fork and exec (each needs a session of its own), and between the two a child holds a copy of every descriptor the harness has open, so the lock stayed held until the child exec'd. Every lock is now held through a guard that unlocks explicitly before the file closes: the session owner, the index lock, trust.lock (a refused change included), and ledger.lock. The lock taken and the order locks are taken in are unchanged, so older binaries still exclude this one. --- crates/core/src/execution.rs | 89 +++++++++++++++++++++++++++++++ crates/core/src/ledger.rs | 10 ++-- crates/core/src/sessions.rs | 100 +++++++++++++++++++++++++++++++---- crates/core/src/trust.rs | 26 ++++++++- 4 files changed, 207 insertions(+), 18 deletions(-) diff --git a/crates/core/src/execution.rs b/crates/core/src/execution.rs index ef179664..46f86712 100644 --- a/crates/core/src/execution.rs +++ b/crates/core/src/execution.rs @@ -714,6 +714,8 @@ pub(crate) async fn run_process( // from `openmax --check --run-examples`, never from a session the TUI // draws. configure_process_group(&mut command, request.sandbox.is_none()); + #[cfg(all(test, unix))] + hold_before_exec(&mut command); let mut child = command.spawn().map_err(ProcessError::Spawn)?; let pid = child.id(); @@ -1041,6 +1043,93 @@ fn configure_process_group(command: &mut Command, new_session: bool) { #[cfg(not(unix))] fn configure_process_group(_: &mut Command, _: bool) {} +#[cfg(all(test, unix))] +thread_local! { + /// Set by [`PausedSpawn`] for the next spawn on its thread: the pipe the + /// child reports its fork on, the pipe it waits on before exec, and that + /// pipe's write end. + static PAUSE_BEFORE_EXEC: std::cell::Cell> = + const { std::cell::Cell::new(None) }; +} + +#[cfg(all(test, unix))] +fn hold_before_exec(command: &mut Command) { + use std::os::unix::process::CommandExt; + let Some([forked, resume, resume_writer]) = PAUSE_BEFORE_EXEC.take() else { return }; + // SAFETY: close, write, and read are async-signal-safe, the only kind of + // call allowed between fork and exec. + unsafe { + command.as_std_mut().pre_exec(move || { + // The child's own copy of the write end would keep it waiting + // forever if the test let go without writing. + libc::close(resume_writer); + let mut byte = 0_u8; + libc::write(forked, (&raw const byte).cast(), 1); + libc::read(resume, (&raw mut byte).cast(), 1); + Ok(()) + }); + } +} + +/// A command started through [`run_process`] and held between fork and exec, +/// the window in which a spawning child holds a copy of every descriptor the +/// harness has open. Dropping it lets the child exec and waits for the +/// command to finish. +#[cfg(all(test, unix))] +pub(crate) struct PausedSpawn { + resume: Option, + spawner: Option>, +} + +#[cfg(all(test, unix))] +impl PausedSpawn { + pub(crate) fn start() -> Self { + use std::io::Read; + use std::os::fd::AsRawFd; + let (mut forked, forked_writer) = std::io::pipe().unwrap(); + let (resume_reader, resume) = std::io::pipe().unwrap(); + let fds = [forked_writer.as_raw_fd(), resume_reader.as_raw_fd(), resume.as_raw_fd()]; + let spawner = std::thread::spawn(move || { + PAUSE_BEFORE_EXEC.set(Some(fds)); + let request = ProcessRequest { + program: "/bin/echo".into(), + args: Vec::new(), + cwd: std::env::temp_dir(), + stdin: StdinMode::Null, + timeout: Duration::from_secs(30), + capture: CaptureSpec { head_bytes: 64, tail_bytes: 64, spill_dir: None, spill_bytes_per_stream: 1024 }, + sandbox: None, + env_allowlist: None, + self_link: false, + }; + let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + let output = runtime.block_on(run_process(request, Arc::new(CancelToken::default()))); + // Held until the spawn returns, so a spawn that never forked + // reads as end of file instead of a wait that never ends. + drop((forked_writer, resume_reader)); + assert!(output.is_ok(), "the held command failed once resumed"); + }); + assert_eq!(forked.read(&mut [0_u8]).unwrap(), 1, "the command never reached its fork"); + Self { resume: Some(resume), spawner: Some(spawner) } + } +} + +#[cfg(all(test, unix))] +impl Drop for PausedSpawn { + fn drop(&mut self) { + use std::io::Write; + if let Some(mut resume) = self.resume.take() { + let _ = resume.write_all(&[0]); + } + if let Some(spawner) = self.spawner.take() { + let finished = spawner.join(); + if !std::thread::panicking() { + finished.unwrap(); + } + } + } +} + #[cfg(unix)] fn send_termination(_: &mut Child, pid: Option) { signal_group(pid, libc::SIGTERM); diff --git a/crates/core/src/ledger.rs b/crates/core/src/ledger.rs index 97f24e5d..a40d9e1f 100644 --- a/crates/core/src/ledger.rs +++ b/crates/core/src/ledger.rs @@ -515,10 +515,8 @@ fn with_lock(dir: &Path, f: impl FnOnce() -> Result) -> Result u64 { @@ -1849,7 +1847,7 @@ pub fn record_usage( .write(true) .open(lock_path(&dir)) .map_err(|e| format!("cannot open ledger lock: {e}"))?; - lock.lock().map_err(|e| format!("cannot lock ledger: {e}"))?; + let lock = crate::sessions::FileLock::wait(lock).map_err(|e| format!("cannot lock ledger: {e}"))?; let result = (|| { // A malformed usage file starts over rather than blocking turns: // usage is telemetry, not policy. @@ -1880,7 +1878,7 @@ pub fn record_usage( let json = serde_json::to_string_pretty(&file).map_err(|e| e.to_string())?; crate::sessions::write_atomic(&usage_path(&dir), json) })(); - let _ = lock.unlock(); + drop(lock); result } diff --git a/crates/core/src/sessions.rs b/crates/core/src/sessions.rs index debff1e1..f9e5441e 100644 --- a/crates/core/src/sessions.rs +++ b/crates/core/src/sessions.rs @@ -69,7 +69,7 @@ pub struct SessionMeta { /// the session it then owns is no longer indexed, and every release drops /// writes to such a session. pub(crate) struct SessionOwner { - _file: std::fs::File, + _file: FileLock, detach: bool, } @@ -99,7 +99,7 @@ fn claim_session(core: &Core, id: &str, validate: bool, reactivate: bool) -> Res let path = lock_path(core, id); let file = std::fs::OpenOptions::new().create(true).write(true).truncate(false) .open(&path).map_err(|e| format!("cannot open session lock {}: {e}", path.display()))?; - file.try_lock().map_err(|e| match e { + let file = FileLock::try_take(file).map_err(|e| match e { std::fs::TryLockError::WouldBlock => format!("session {id} is already open in another process; close it there or start a new session"), std::fs::TryLockError::Error(e) => format!("cannot lock session {id}: {e}"), @@ -540,7 +540,7 @@ fn save_index(core: &Core, metas: &[SessionMeta]) -> Result<(), String> { /// description, so that is what keeps one process from contending with /// itself (see the ledger's `with_lock` note). The lock releases when the /// returned handle drops. -fn lock_index(core: &Core) -> Result { +fn lock_index(core: &Core) -> Result { let dir = sessions_dir(core); let _ = std::fs::create_dir_all(&dir); let path = dir.join("index.lock"); @@ -550,22 +550,57 @@ fn lock_index(core: &Core) -> Result { .truncate(false) .open(&path) .map_err(|e| format!("cannot open {}: {e}", path.display()))?; - file.lock() - .map_err(|e| format!("cannot lock {}: {e}", path.display()))?; - Ok(file) + FileLock::wait(file) + .map_err(|e| format!("cannot lock {}: {e}", path.display())) +} + +/// An exclusive lock held on an open lock file, released by an explicit +/// unlock when dropped, before the file closes. flock(2) belongs to the open +/// file description, not to one descriptor, and a child being spawned holds +/// a copy of every descriptor this process has open until it execs (bash, +/// tools, and hooks all spawn that way, see `configure_process_group`). +/// Closing only this copy would keep the lock held until that child exec'd: +/// a session released and reopened in the window is refused as open in +/// another process, and other processes waiting on the lock stall. The +/// unlock releases it whatever copies remain. The lock taken is unchanged, +/// so older binaries still exclude this one and are excluded by it. +pub(crate) struct FileLock(std::fs::File); + +impl FileLock { + /// Wait until the lock is free, then take it. + pub(crate) fn wait(file: std::fs::File) -> std::io::Result { + file.lock()?; + Ok(Self(file)) + } + + /// Take the lock, or fail at once while another holder has it. + pub(crate) fn try_take(file: std::fs::File) -> Result { + file.try_lock()?; + Ok(Self(file)) + } +} + +impl Drop for FileLock { + fn drop(&mut self) { + // Should the unlock fail, closing the file still releases the lock + // once no spawning child holds a copy. + let _ = self.0.unlock(); + } } /// Take a lock file the way every released binary does: an exclusive flock(2) /// on its own open file description, without blocking. `None` while another /// holder has it. Tests use it to stand in for an older binary sharing the -/// data dir, because std's `File::lock` must keep excluding that binary. +/// data dir, because std's `File::lock` must keep excluding that binary. It +/// is released like the harness's own locks, so a child another test is +/// spawning cannot hold it past the drop. #[cfg(all(test, unix))] -pub(crate) fn raw_flock(path: &std::path::Path) -> Option { +pub(crate) fn raw_flock(path: &std::path::Path) -> Option { use std::os::unix::io::AsRawFd; let file = std::fs::OpenOptions::new().create(true).write(true).truncate(false) .open(path).unwrap(); let held = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) } == 0; - held.then_some(file) + held.then_some(FileLock(file)) } /// Read-modify-write the index under the state lock (concurrent turns in @@ -585,7 +620,7 @@ fn with_index(core: &Core, f: impl FnOnce(&mut Vec) -> R) -> Res Ok(result) } -fn lock_store(core: &Core) -> Result<(std::sync::MutexGuard<'_, ()>, std::fs::File), String> { +fn lock_store(core: &Core) -> Result<(std::sync::MutexGuard<'_, ()>, FileLock), String> { let guard = core.sessions_lock.lock().unwrap(); let flock = lock_index(core)?; Ok((guard, flock)) @@ -2346,6 +2381,51 @@ mod tests { let _ = std::fs::remove_dir_all(dir); } + /// A released session must be free at once, not when the last copy of + /// its lock descriptor closes. flock belongs to the open file + /// description, and a child being spawned holds a copy of every open + /// descriptor until it execs: a release that only closed the harness's + /// copy left the session locked while any bash call, tool, or hook was + /// mid-spawn, and reopening it then was refused as open in another + /// process. + #[cfg(unix)] + #[test] + fn a_released_session_reopens_while_a_child_is_spawning() { + let dir = std::env::temp_dir().join(format!("openmax-spawn-session-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.clone()).unwrap(); + let (other, _other_rx) = Core::new(dir.clone()).unwrap(); + let id = create(&core, "/tmp/p".into()).unwrap().id; + attach(&core, &id).unwrap(); + + let spawning = crate::execution::PausedSpawn::start(); + detach(&core, &id).unwrap(); + attach(&core, &id).expect("a released session reopens while a child is spawning"); + detach(&core, &id).unwrap(); + attach(&other, &id).expect("another process takes a released session while a child is spawning"); + assert!(attach(&core, &id).unwrap_err().contains("another process"), "a held session stays exclusive"); + drop(spawning); + let _ = std::fs::remove_dir_all(dir); + } + + /// The index lock is released the same way, or every other process's + /// index write waits on a child it never started. + #[cfg(unix)] + #[test] + fn a_released_index_lock_is_free_while_a_child_is_spawning() { + let dir = std::env::temp_dir().join(format!("openmax-spawn-index-{}", uuid::Uuid::new_v4())); + let (core, _rx) = Core::new(dir.clone()).unwrap(); + let store = lock_store(&core).unwrap(); + + let spawning = crate::execution::PausedSpawn::start(); + drop(store); + assert!( + raw_flock(&sessions_dir(&core).join("index.lock")).is_some(), + "a released index lock stayed held by a spawning child" + ); + drop(spawning); + let _ = std::fs::remove_dir_all(dir); + } + /// Every claim creates `.lock`. A deleted or discarded session is /// never claimed again, so a lock file it leaves behind is never reused /// or removed: one stray file for every such session, forever. A session diff --git a/crates/core/src/trust.rs b/crates/core/src/trust.rs index a8e58e55..4a5ac23d 100644 --- a/crates/core/src/trust.rs +++ b/crates/core/src/trust.rs @@ -193,14 +193,14 @@ fn update(data_dir: &Path, change: impl FnOnce(&mut TrustFile) -> Result<(), Str .write(true) .open(&lock_path) .map_err(|e| format!("cannot open {}: {e}", lock_path.display()))?; - lock.lock() + // Released on every return, refusals included (see `FileLock`). + let _lock = crate::sessions::FileLock::wait(lock) .map_err(|e| format!("cannot lock {}: {e}", lock_path.display()))?; let mut file = load(data_dir)?; change(&mut file)?; let json = serde_json::to_vec_pretty(&file).map_err(|e| e.to_string())?; crate::sessions::write_atomic(&trust_path(data_dir), json)?; - lock.unlock().map_err(|e| format!("cannot unlock {}: {e}", lock_path.display()))?; Ok(()) } @@ -433,6 +433,28 @@ mod tests { let _ = std::fs::remove_dir_all(project); } + /// A refused change releases trust.lock at once, like a saved one. A + /// child being spawned holds a copy of the lock's descriptor until it + /// execs, so a release that only closed it kept every other process's + /// trust write waiting on that child. + #[cfg(unix)] + #[test] + fn a_refused_trust_change_frees_its_lock_while_a_child_is_spawning() { + let data = temp_dir("spawn-data"); + let mut spawning = None; + let refused = update(&data, |_| { + spawning = Some(crate::execution::PausedSpawn::start()); + Err("refused".into()) + }); + assert_eq!(refused, Err("refused".to_string())); + assert!( + crate::sessions::raw_flock(&trust_lock_path(&data)).is_some(), + "a refused change left trust.lock held by a spawning child" + ); + drop(spawning); + let _ = std::fs::remove_dir_all(data); + } + #[cfg(unix)] #[test] fn symlink_alias_resolves_to_the_same_trust_identity() {