diff --git a/codex-rs/rollout/src/writer_lock.rs b/codex-rs/rollout/src/writer_lock.rs index 31068707a2..7ad0f8e619 100644 --- a/codex-rs/rollout/src/writer_lock.rs +++ b/codex-rs/rollout/src/writer_lock.rs @@ -1,5 +1,6 @@ //! Shared cross-process ownership for local thread writers and rollout publication. -//! Thread lock-file creation and removal uses a bounded wait for the home coordination lock. +//! Opening a thread lock file uses a bounded wait for home coordination. Cleanup +//! removes the file only if it can acquire exclusive coordination immediately. use std::fs; use std::fs::File; @@ -43,12 +44,22 @@ impl WriterLockCoordinator { /// Acquires exclusive writer ownership, returning `WouldBlock` for an active writer. pub fn acquire(self: &Arc, thread_id: ThreadId) -> io::Result { - let _coordination_lock = self.lock_coordination()?; - if !self.cleanup_attempted.swap(true, Ordering::Relaxed) - && let Err(err) = self.remove_stale_thread_locks() - { - warn!("failed to clean up stale thread writer locks: {err}"); + let file = self.open_coordination()?; + // Never queue a directory sweep behind another writer. Removal remains exclusive. + let have_exclusive = !self.cleanup_attempted.swap(true, Ordering::Relaxed) + && match file.try_lock() { + Ok(()) => true, + Err(fs::TryLockError::WouldBlock) => false, + Err(fs::TryLockError::Error(err)) => return Err(err), + }; + if have_exclusive { + if let Err(err) = self.remove_stale_thread_locks() { + warn!("failed to clean up stale thread writer locks: {err}"); + } + } else { + Self::wait_for_coordination(|| file.try_lock_shared())?; } + let _coordination_lock = file; let path = self.directory.join(format!("{thread_id}.lock")); let file = OpenOptions::new() @@ -107,18 +118,29 @@ impl WriterLockCoordinator { } } - fn lock_coordination(&self) -> io::Result { + fn open_coordination(&self) -> io::Result { fs::create_dir_all(&self.directory)?; - let file = OpenOptions::new() + OpenOptions::new() .read(true) .write(true) .create(true) .truncate(false) - .open(self.directory.join(COORDINATION_LOCK_FILE))?; + .open(self.directory.join(COORDINATION_LOCK_FILE)) + } + + fn lock_coordination(&self) -> io::Result { + let file = self.open_coordination()?; + Self::wait_for_coordination(|| file.try_lock())?; + Ok(file) + } + + fn wait_for_coordination( + try_lock: impl Fn() -> Result<(), fs::TryLockError>, + ) -> io::Result<()> { let deadline = Instant::now() + Duration::from_secs(/*secs*/ 30); loop { - match file.try_lock() { - Ok(()) => break, + match try_lock() { + Ok(()) => return Ok(()), Err(fs::TryLockError::WouldBlock) if Instant::now() < deadline => { std::thread::sleep(Duration::from_millis(/*millis*/ 50)); } @@ -131,7 +153,6 @@ impl WriterLockCoordinator { Err(fs::TryLockError::Error(err)) => return Err(err), } } - Ok(file) } fn remove_stale_thread_locks(&self) -> io::Result<()> { @@ -186,7 +207,7 @@ impl WriterLockCoordinator { impl Drop for WriterLockGuard { fn drop(&mut self) { - let coordination_lock = match self.coordinator.lock_coordination() { + let coordination_lock = match self.coordinator.open_coordination() { Ok(lock) => lock, Err(err) => { warn!("failed to coordinate thread writer lock cleanup: {err}"); @@ -194,6 +215,16 @@ impl Drop for WriterLockGuard { } }; + match coordination_lock.try_lock() { + Ok(()) => {} + // Leave the file for the next writer. Dropping the guard still closes it. + Err(fs::TryLockError::WouldBlock) => return, + Err(fs::TryLockError::Error(err)) => { + warn!("failed to coordinate thread writer lock cleanup: {err}"); + return; + } + } + // Close the writer lock before deleting it so cleanup works on Windows too. drop(self.file.take()); if let Err(err) = fs::remove_file(&self.path) diff --git a/codex-rs/rollout/src/writer_lock_tests.rs b/codex-rs/rollout/src/writer_lock_tests.rs index 89e42c1f40..1b60d86c50 100644 --- a/codex-rs/rollout/src/writer_lock_tests.rs +++ b/codex-rs/rollout/src/writer_lock_tests.rs @@ -1,5 +1,7 @@ use std::fs; use std::sync::Arc; +use std::sync::mpsc; +use std::time::Duration; use codex_protocol::ThreadId; use tempfile::TempDir; @@ -25,6 +27,18 @@ fn writer_locks_reject_competing_owners_and_release_their_files() { .join(format!("{thread_id}.lock")); assert!(lock_path.exists()); + // Another process is opening a writer. Distinct threads can start together, + // but even during this overlap the same thread must still have one owner. + let opening = fs::OpenOptions::new() + .read(true) + .write(true) + .open( + home.path() + .join(WRITER_LOCK_DIR) + .join(COORDINATION_LOCK_FILE), + ) + .expect("open coordination"); + opening.lock_shared().expect("another writer is opening"); let err = match secondary.acquire(thread_id) { Ok(_) => panic!("competing owner should fail"), Err(err) => err, @@ -34,11 +48,21 @@ fn writer_locks_reject_competing_owners_and_release_their_files() { .acquire(other_thread_id) .expect("other thread should acquire its own lock"); - drop(owner); - assert!(!lock_path.exists()); - let next_owner = secondary - .acquire(thread_id) - .expect("released thread should accept another owner"); + let next_owner = std::thread::scope(|scope| { + let _opening = opening; + let (closed, wait_for_close) = mpsc::channel(); + scope.spawn(move || { + drop(owner); + let _ = closed.send(()); + }); + wait_for_close + .recv_timeout(Duration::from_secs(/*secs*/ 10)) + .expect("writer should close while another writer is opening"); + assert!(lock_path.exists()); + secondary + .acquire(thread_id) + .expect("released thread should accept another owner") + }); drop(next_owner); drop(other_owner);