fix(core): release every file lock at once, even while a command is starting

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<Option<[std::os::fd::RawFd; 3]>> =
+        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<std::io::PipeWriter>,
+    spawner: Option<std::thread::JoinHandle<()>>,
+}
+
+#[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<u32>) {
     signal_group(pid, libc::SIGTERM);

crates/core/src/ledger.rs

@@ -515,10 +515,8 @@ fn with_lock<R>(dir: &Path, f: impl FnOnce() -> Result<R, String>) -> Result<R,
         .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 result = f();
-    let _ = lock.unlock();
-    result
+    let _lock = crate::sessions::FileLock::wait(lock).map_err(|e| format!("cannot lock ledger: {e}"))?;
+    f()
 }
 
 fn unix_now() -> 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
 }
 

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<std::fs::File, String> {
+fn lock_index(core: &Core) -> Result<FileLock, String> {
     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<std::fs::File, String> {
         .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<Self> {
+        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<Self, std::fs::TryLockError> {
+        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<std::fs::File> {
+pub(crate) fn raw_flock(path: &std::path::Path) -> Option<FileLock> {
     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<R>(core: &Core, f: impl FnOnce(&mut Vec<SessionMeta>) -> 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 `<id>.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

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