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]