From 4d3ed3f3b47d39f8fe0b60df74a20f528b6fc618 Mon Sep 17 00:00:00 2001 From: jif Date: Tue, 27 Oct 2026 18:25:09 +0000 Subject: [PATCH] Release empty PID probe locks explicitly on Unix (#57533) ## Why A concurrent fork can inherit a PID reservation probe's file descriptor and keep its lock held after the probe closes it, making a stale reservation appear active. ## What changed Add a drop guard that explicitly unlocks empty PID reservation probes on Unix, including when inspection fails or is cancelled. ## Testing Extend the duplicated-handle regression test to cover empty PID files, read errors, and cancellation alongside missing PID files. Verify that cancellation occurs while the lock is held and that subsequent status checks do not report an active reservation. GitOrigin-RevId: 793569c9bcaca02900662a66c33aa48f2bebfd5e --- codex-rs/app-server-daemon/src/backend/pid.rs | 29 +++- .../src/backend/pid_tests.rs | 131 ++++++++++++++---- 2 files changed, 132 insertions(+), 28 deletions(-) diff --git a/codex-rs/app-server-daemon/src/backend/pid.rs b/codex-rs/app-server-daemon/src/backend/pid.rs index 7a196e0604..97738dd8e2 100644 --- a/codex-rs/app-server-daemon/src/backend/pid.rs +++ b/codex-rs/app-server-daemon/src/backend/pid.rs @@ -1,5 +1,5 @@ //! PID reservations serialize detached launches; process identities protect stale-record cleanup. -//! Missing-PID probes explicitly unlock so concurrent forks cannot retain their transient reservations. +//! PID probes explicitly unlock so concurrent forks cannot retain their transient reservations. #[path = "pid_identity.rs"] mod identity; @@ -750,9 +750,36 @@ async fn inspect_empty_pid_reservation( }); } }; + inspect_empty_pid_reservation_with_file(pid_path, file).await +} + +#[cfg(unix)] +struct EmptyPidProbeLock(fs::File); + +#[cfg(unix)] +impl Drop for EmptyPidProbeLock { + fn drop(&mut self) { + use std::os::fd::AsRawFd; + + // Closing alone leaves the lock held by any concurrent fork. Release it + // on every exit, including errors and cancellation while inspecting the PID. + if unsafe { libc::flock(self.0.as_raw_fd(), libc::LOCK_UN) } != 0 { + let error = std::io::Error::last_os_error(); + tracing::warn!(%error, "failed to unlock empty pid reservation probe"); + } + } +} + +#[cfg(any(unix, windows))] +async fn inspect_empty_pid_reservation_with_file( + pid_path: &Path, + file: fs::File, +) -> Result { if !try_lock_file(&file)? { return Ok(EmptyPidReservation::Active); } + #[cfg(unix)] + let _probe_lock = EmptyPidProbeLock(file); let contents = match fs::read_to_string(pid_path).await { Ok(contents) => contents, diff --git a/codex-rs/app-server-daemon/src/backend/pid_tests.rs b/codex-rs/app-server-daemon/src/backend/pid_tests.rs index 0e11df230e..3c47d12496 100644 --- a/codex-rs/app-server-daemon/src/backend/pid_tests.rs +++ b/codex-rs/app-server-daemon/src/backend/pid_tests.rs @@ -101,35 +101,112 @@ async fn unlocked_empty_pid_file_is_treated_as_stale_reservation() { } #[cfg(unix)] -#[tokio::test] -async fn reservation_probe_releases_lock_with_duplicated_handle() { - let temp_dir = TempDir::new().expect("temp dir"); - let backend = PidBackend::new( - temp_dir.path().join("codex"), - temp_dir.path().join("app-server.pid"), - /*remote_control_enabled*/ false, - ); - let file = std::fs::OpenOptions::new() - .create(true) - .truncate(false) - .write(true) - .open(&backend.lock_file) - .expect("open pid lock file"); - // Share the open file description as fork would, without inheriting - // unrelated descriptors from other tests. - let inherited = file.try_clone().expect("clone pid lock file"); +#[test] +fn reservation_probe_releases_lock_with_duplicated_handle() { + #[derive(Debug, Clone, Copy)] + enum Probe { + Missing, + Empty, + ReadError, + Cancelled, + } - let probe = super::probe_reservation_lock(tokio::fs::File::from_std(file)); - let running = backend.is_starting_or_running().await; + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .max_blocking_threads(/*val*/ 1) + .build() + .expect("runtime"); + runtime.block_on(async { + for case in [ + Probe::Missing, + Probe::Empty, + Probe::ReadError, + Probe::Cancelled, + ] { + let temp_dir = TempDir::new().expect("temp dir"); + let backend = PidBackend::new( + temp_dir.path().join("codex"), + temp_dir.path().join("app-server.pid"), + /*remote_control_enabled*/ false, + ); + match case { + Probe::Missing => {} + Probe::Empty | Probe::Cancelled => { + std::fs::write(&backend.pid_file, "").expect("empty pid file"); + } + Probe::ReadError => { + std::fs::create_dir(&backend.pid_file).expect("unreadable pid file"); + } + } + let file = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .write(true) + .open(&backend.lock_file) + .expect("open pid lock file"); + // Share the open file description as fork would, without inheriting + // unrelated descriptors from other tests. + let inherited = file.try_clone().expect("clone pid lock file"); - drop(inherited); - assert_eq!( - ( - probe.expect("probe reservation"), - running.expect("read daemon status"), - ), - (false, false), - ); + let file = tokio::fs::File::from_std(file); + let mut cancellation = None; + let probe = match case { + Probe::Missing => super::probe_reservation_lock(file), + Probe::Empty | Probe::ReadError => { + super::inspect_empty_pid_reservation_with_file(&backend.pid_file, file) + .await + .map(|state| state != super::EmptyPidReservation::Stale) + } + Probe::Cancelled => { + // Occupy the sole blocking worker so the PID read cannot finish + // before cancellation. No timing assumptions or sleeps are needed. + let (started_tx, started_rx) = std::sync::mpsc::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let blocker = tokio::task::spawn_blocking(move || { + started_tx.send(()).expect("worker started"); + release_rx.recv().expect("release worker"); + }); + started_rx.recv().expect("wait for worker"); + let mut inspection = Box::pin(super::inspect_empty_pid_reservation_with_file( + &backend.pid_file, + file, + )); + let pending = futures::poll!(&mut inspection).is_pending(); + let contender = tokio::fs::File::from_std( + std::fs::OpenOptions::new() + .write(true) + .open(&backend.lock_file) + .expect("open competing reservation"), + ); + cancellation = Some((pending, try_lock_file(&contender))); + drop(contender); + drop(inspection); + release_tx.send(()).expect("unblock PID read"); + blocker.await.expect("blocking worker"); + Ok(false) + } + }; + if matches!(case, Probe::ReadError) { + std::fs::remove_dir(&backend.pid_file).expect("remove unreadable pid file"); + } + let running = backend.is_starting_or_running().await; + + drop(inherited); + match case { + Probe::ReadError => assert!(probe.is_err(), "{case:?}"), + Probe::Missing | Probe::Empty | Probe::Cancelled => { + assert!(!probe.expect("probe reservation"), "{case:?}"); + } + } + if let Some((pending, acquired)) = cancellation { + assert_eq!( + (pending, acquired.expect("competing reservation")), + (true, false) + ); + } + assert!(!running.expect("read daemon status"), "{case:?}"); + } + }); } #[tokio::test]