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() {