Serialize session index mutations across processes (#57358)

## Why

The process-local mutex cannot coordinate session name appends and removals from multiple Codex processes sharing the same home directory.

## What changed

- Guard index mutations with a persistent `session_index.jsonl.lock` file, retaining the lock until the blocking worker finishes even if its caller is cancelled.
- Return a timeout error after 30 seconds of lock contention without modifying the index.
- Rewrite removals through a unique temporary file while preserving index permissions, and skip removal when the index is absent.

## Testing

Update regression tests to check lock retention after cancellation, timeout without writing, successful writes after contention clears, and removal with a missing home directory. Retain coverage for cancelled queued updates preserving the next write.

GitOrigin-RevId: 5231c9ff85c6b34de5ee92931d119b6f46bd583b

codex-rs/rollout/src/session_index.rs

@@ -1,3 +1,6 @@
+//! Store session names in an append-only index and serialize index rewrites.
+//! Writers must share the filesystem lock namespace and use this locking protocol.
+
 use std::collections::HashMap;
 use std::collections::HashSet;
 use std::fs::File;
@@ -5,8 +8,8 @@ use std::io::ErrorKind;
 use std::io::Write;
 use std::path::Path;
 use std::path::PathBuf;
-use std::sync::Arc;
-use std::sync::LazyLock;
+use std::time::Duration;
+use std::time::Instant;
 use std::time::SystemTime;
 
 use crate::reverse_jsonl_scanner::ReverseJsonlScanner;
@@ -18,8 +21,7 @@ use serde::Deserialize;
 use serde::Serialize;
 
 const SESSION_INDEX_FILE: &str = "session_index.jsonl";
-static SESSION_INDEX_LOCK: LazyLock<Arc<tokio::sync::Mutex<()>>> =
-    LazyLock::new(|| Arc::new(tokio::sync::Mutex::new(())));
+const SESSION_INDEX_LOCK_FILE: &str = "session_index.jsonl.lock";
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct SessionIndexEntry {
@@ -56,7 +58,7 @@ pub async fn append_session_index_entry(
     entry: SessionIndexEntry,
 ) -> std::io::Result<()> {
     let path = session_index_path(codex_home);
-    with_session_index_lock(&SESSION_INDEX_LOCK, move || {
+    with_session_index_lock(codex_home, move || {
         let mut file = std::fs::OpenOptions::new()
             .create(true)
             .append(true)
@@ -76,7 +78,10 @@ pub async fn remove_thread_name_entries(
     thread_id: ThreadId,
 ) -> std::io::Result<()> {
     let path = session_index_path(codex_home);
-    with_session_index_lock(&SESSION_INDEX_LOCK, move || {
+    if !path.try_exists()? {
+        return Ok(());
+    }
+    with_session_index_lock(codex_home, move || {
         let contents = match std::fs::read_to_string(&path) {
             Ok(contents) => contents,
             Err(err) if err.kind() == ErrorKind::NotFound => return Ok(()),
@@ -97,21 +102,50 @@ pub async fn remove_thread_name_entries(
         if !removed {
             return Ok(());
         }
-        let temp_path = path.with_extension("jsonl.tmp");
-        std::fs::write(&temp_path, remaining)?;
-        std::fs::rename(temp_path, path)
+        let mut temporary = tempfile::Builder::new()
+            .prefix("session-index-")
+            .tempfile_in(path.parent().unwrap_or(Path::new(".")))?;
+        temporary
+            .as_file()
+            .set_permissions(std::fs::metadata(&path)?.permissions())?;
+        temporary.write_all(remaining.as_bytes())?;
+        temporary.persist(&path).map_err(|error| error.error)?;
+        Ok(())
     })
     .await
 }
 
-/// Acquire the ordered index lock before dispatch and retain it through cancelled updates.
+/// Serialize index mutations within and across processes, including cancelled updates.
 async fn with_session_index_lock(
-    lock: &Arc<tokio::sync::Mutex<()>>,
+    codex_home: &Path,
     update: impl FnOnce() -> std::io::Result<()> + Send + 'static,
 ) -> std::io::Result<()> {
-    let guard = Arc::clone(lock).lock_owned().await;
+    let lock_path = codex_home.join(SESSION_INDEX_LOCK_FILE);
     tokio::task::spawn_blocking(move || {
-        let _guard = guard;
+        // Keep this file in place. If we replace it, another process can lock a new
+        // file while this update still holds the old file's lock.
+        let file_lock = std::fs::OpenOptions::new()
+            .create(true)
+            .truncate(false)
+            .read(true)
+            .write(true)
+            .open(lock_path)?;
+        let deadline = Instant::now() + Duration::from_secs(/*secs*/ 30);
+        loop {
+            match file_lock.try_lock() {
+                Ok(()) => break,
+                Err(std::fs::TryLockError::WouldBlock) if Instant::now() < deadline => {
+                    std::thread::sleep(Duration::from_millis(/*millis*/ 50));
+                }
+                Err(std::fs::TryLockError::WouldBlock) => {
+                    return Err(std::io::Error::new(
+                        ErrorKind::TimedOut,
+                        "timed out waiting for the session index lock; another Codex process may be suspended on this or another host",
+                    ));
+                }
+                Err(std::fs::TryLockError::Error(err)) => return Err(err),
+            }
+        }
         update()
     })
     .await

codex-rs/rollout/src/session_index_tests.rs

@@ -487,6 +487,8 @@ async fn find_thread_names_by_ids_skips_unusable_names_and_reads_across_chunks()
 
 #[tokio::test]
 async fn removal_preserves_other_names_and_malformed_lines() -> std::io::Result<()> {
+    let absent_home = TempDir::new()?;
+    remove_thread_name_entries(&absent_home.path().join("missing"), ThreadId::new()).await?;
     let temp = TempDir::new()?;
     let removed_id = ThreadId::new();
     let kept_id = ThreadId::new();
@@ -514,26 +516,54 @@ async fn removal_preserves_other_names_and_malformed_lines() -> std::io::Result<
 }
 
 #[tokio::test]
-async fn cancelled_index_update_holds_lock_until_worker_finishes() -> std::io::Result<()> {
+async fn index_lock_survives_cancellation_and_times_out_without_writing() -> std::io::Result<()> {
+    let home = TempDir::new()?;
+    let update_home = home.path().to_path_buf();
     let (reached_tx, reached_rx) = tokio::sync::oneshot::channel();
     let (resume_tx, resume_rx) = std::sync::mpsc::channel();
-    let update = tokio::spawn(with_session_index_lock(&SESSION_INDEX_LOCK, move || {
-        let _ = reached_tx.send(());
-        // Dropping the sender on a test failure also releases the blocking worker.
-        let _ = resume_rx.recv();
-        Ok(())
-    }));
+    let update = tokio::spawn(async move {
+        with_session_index_lock(&update_home, move || {
+            let _ = reached_tx.send(());
+            // Dropping the sender on a test failure also releases the blocking worker.
+            let _ = resume_rx.recv();
+            Ok(())
+        })
+        .await
+    });
     reached_rx.await.unwrap();
     update.abort();
     assert!(update.await.unwrap_err().is_cancelled());
 
-    let lock_is_held = SESSION_INDEX_LOCK.try_lock().is_err();
+    let independent_file = std::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .open(home.path().join(SESSION_INDEX_LOCK_FILE))?;
+    let file_lock_is_held = match independent_file.try_lock() {
+        Ok(()) => {
+            independent_file.unlock()?;
+            false
+        }
+        Err(std::fs::TryLockError::WouldBlock) => true,
+        Err(std::fs::TryLockError::Error(error)) => return Err(error),
+    };
     resume_tx.send(()).unwrap();
-    with_session_index_lock(&SESSION_INDEX_LOCK, || Ok(())).await?;
+    with_session_index_lock(home.path(), || Ok(())).await?;
     assert!(
-        lock_is_held,
-        "cancellation released the running worker's lock"
+        file_lock_is_held,
+        "cancellation released a running index mutation's lock"
     );
+    independent_file
+        .try_lock()
+        .expect("the completed mutation must release its file lock");
+    // A busy file lock must not allow an append to run without ownership.
+    let error = append_thread_name(home.path(), ThreadId::new(), "blocked")
+        .await
+        .unwrap_err();
+    assert_eq!(error.kind(), std::io::ErrorKind::TimedOut);
+    assert!(!session_index_path(home.path()).exists());
+    drop(independent_file);
+    append_thread_name(home.path(), ThreadId::new(), "after contention").await?;
+    assert!(session_index_path(home.path()).is_file());
     Ok(())
 }
 
@@ -546,8 +576,6 @@ fn cancelled_queued_index_update_preserves_the_next_write() -> std::io::Result<(
     runtime.block_on(async {
         let home = TempDir::new()?;
         let path = home.path().join("thread-name");
-        // A private lock ensures shared-process test runners cannot supply another owner's guard.
-        let lock = Arc::new(tokio::sync::Mutex::new(()));
 
         let (reached_tx, reached_rx) = tokio::sync::oneshot::channel();
         let (resume_tx, resume_rx) = std::sync::mpsc::channel::<()>();
@@ -559,7 +587,7 @@ fn cancelled_queued_index_update_preserves_the_next_write() -> std::io::Result<(
         reached_rx.await.unwrap();
 
         let queued_path = path.clone();
-        let mut update = Box::pin(with_session_index_lock(&lock, move || {
+        let mut update = Box::pin(with_session_index_lock(home.path(), move || {
             std::fs::write(queued_path, "obsolete")
         }));
         let state = std::future::poll_fn(|cx| {
@@ -570,21 +598,13 @@ fn cancelled_queued_index_update_preserves_the_next_write() -> std::io::Result<(
             state.is_pending(),
             "the blocking pool should hold the update queued"
         );
-        assert!(
-            lock.try_lock().is_err(),
-            "this update must own the lock before cancellation"
-        );
         drop(update);
-        let queued_update_holds_lock = lock.try_lock().is_err();
 
         drop(resume_tx);
         blocker.await.unwrap();
         let next_path = path.clone();
-        with_session_index_lock(&lock, move || std::fs::write(next_path, "latest")).await?;
-        assert_eq!(
-            (queued_update_holds_lock, std::fs::read_to_string(path)?),
-            (true, "latest".to_string())
-        );
+        with_session_index_lock(home.path(), move || std::fs::write(next_path, "latest")).await?;
+        assert_eq!(std::fs::read_to_string(path)?, "latest".to_string());
         Ok(())
     })
 }