diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 900dd16f1d..9fd8933654 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -324,6 +324,8 @@ By default, `thread/resume` includes the reconstructed turn history in `thread.t Paginated threads keep the same resume contract as legacy threads. A default resume materializes the full projected history into `thread.turns`; `excludeTurns: true` keeps that array empty and includes `turnsBackwardsCursor` and `itemsBackwardsCursor` for the durable history visible at the resume boundary. Pass each cursor directly to its matching list API with `sortDirection: "desc"`; the first page includes the cursor's head row, while newer records arrive through live notifications. Either cursor is `null` when there is no durable row yet. +Only one app-server process can hold a paginated thread open for writing at a time. If another process already owns the thread, `thread/resume` fails with JSON-RPC error `-32600`. Read-only requests remain available without resuming the thread. + Experimental clients that want the live resume subscription plus a turns page in one round trip can pass `initialTurnsPage`. It accepts the same `limit`, `sortDirection`, and `itemsView` controls as `thread/turns/list`; omitted controls use its defaults. The response includes `initialTurnsPage` with `nextCursor` and `backwardsCursor` for follow-up pagination. By default, resume uses the latest persisted `model` and `reasoningEffort` values associated with the thread. Supplying any of `model`, `modelProvider`, `config.model`, or `config.model_reasoning_effort` disables that persisted fallback and uses the explicit overrides plus normal config resolution instead. diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 607b71f8bb..5d33dad6b3 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -3399,7 +3399,10 @@ impl ThreadRequestProcessor { .await; } Err(err) => { - let error = internal_error(format!("error resuming thread: {err}")); + let error = match err { + CodexErr::InvalidRequest(message) => invalid_request(message), + err => internal_error(format!("error resuming thread: {err}")), + }; self.outgoing.send_error(request_id, error).await; } } diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index f83096f9d4..f348a8daf1 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -55,6 +55,8 @@ use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStatus; +use codex_app_server_protocol::ThreadTurnsListParams; +use codex_app_server_protocol::ThreadTurnsListResponse; use codex_app_server_protocol::ThreadUnsubscribeParams; use codex_app_server_protocol::TurnItemsView; use codex_app_server_protocol::TurnStartParams; @@ -167,6 +169,114 @@ async fn thread_resume_paginated_metadata_only_uses_model_context() -> Result<() Ok(()) } +#[tokio::test] +async fn thread_resume_rejects_paginated_writer_owned_by_another_process() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + let codex_home = TempDir::new()?; + mock_responses_config(&server.uri()).write(codex_home.path())?; + + let mut primary = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized() + .await?; + let ThreadStartResponse { thread, .. } = primary + .start_thread(ThreadStartParams { + model: Some("gpt-5.4".to_string()), + history_mode: Some(ThreadHistoryMode::Paginated), + ..Default::default() + }) + .await?; + timeout( + DEFAULT_READ_TIMEOUT, + primary.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "first writer".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), + ) + .await??; + + let secondary_sqlite_home = TempDir::new()?; + let secondary_sqlite_home_path = secondary_sqlite_home.path().to_string_lossy(); + let mut secondary = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_env_overrides(&[( + "CODEX_SQLITE_HOME", + Some(secondary_sqlite_home_path.as_ref()), + )]) + .build_initialized() + .await?; + let read_id = secondary + .send_thread_read_request(ThreadReadParams { + thread_id: thread.id.clone(), + include_turns: false, + }) + .await?; + let ThreadReadResponse { thread: read, .. } = + timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(read_id)).await??; + assert_eq!(read.id, thread.id); + + let resume_id = secondary + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + ..Default::default() + }) + .await?; + let error = timeout( + DEFAULT_READ_TIMEOUT, + secondary.read_stream_until_error_message(RequestId::Integer(resume_id)), + ) + .await??; + assert_eq!(error.error.code, -32600); + assert_eq!( + error.error.message, + format!("thread {} already has an active writer", thread.id) + ); + + timeout(DEFAULT_READ_TIMEOUT, primary.shutdown_gracefully()).await??; + + let next_resume_id = secondary + .send_thread_resume_request(ThreadResumeParams { + thread_id: thread.id.clone(), + ..Default::default() + }) + .await?; + let _: ThreadResumeResponse = timeout( + DEFAULT_READ_TIMEOUT, + secondary.read_response(next_resume_id), + ) + .await??; + timeout( + DEFAULT_READ_TIMEOUT, + secondary.start_turn_and_wait_for_completion(TurnStartParams { + thread_id: thread.id.clone(), + input: vec![UserInput::Text { + text: "second writer".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }), + ) + .await??; + + let list_id = secondary + .send_thread_turns_list_request(ThreadTurnsListParams { + thread_id: thread.id, + cursor: None, + limit: None, + sort_direction: None, + items_view: None, + }) + .await?; + let ThreadTurnsListResponse { data, .. } = + timeout(DEFAULT_READ_TIMEOUT, secondary.read_response(list_id)).await??; + assert_eq!(data.len(), 2); + Ok(()) +} + fn normalized_existing_path(path: impl AsRef) -> Result { Ok(AbsolutePathBuf::from_absolute_path(path.as_ref().canonicalize()?)?.into_path_buf()) } diff --git a/codex-rs/core/src/session_rollout_init_error.rs b/codex-rs/core/src/session_rollout_init_error.rs index db71dfcd37..14307a9df3 100644 --- a/codex-rs/core/src/session_rollout_init_error.rs +++ b/codex-rs/core/src/session_rollout_init_error.rs @@ -6,11 +6,21 @@ use codex_protocol::error::CodexErr; use codex_thread_store::ThreadStoreError; pub(crate) fn map_session_init_error(err: &anyhow::Error, codex_home: &Path) -> CodexErr { - if let Some(ThreadStoreError::Unsupported { operation }) = err + if let Some(store_error) = err .chain() .find_map(|cause| cause.downcast_ref::()) { - return CodexErr::UnsupportedOperation(format!("{operation} is not supported yet")); + match store_error { + ThreadStoreError::Unsupported { operation } => { + return CodexErr::UnsupportedOperation(format!("{operation} is not supported yet")); + } + ThreadStoreError::Conflict { message } => { + return CodexErr::InvalidRequest(message.clone()); + } + ThreadStoreError::ThreadNotFound { .. } + | ThreadStoreError::InvalidRequest { .. } + | ThreadStoreError::Internal { .. } => {} + } } if let Some(mapped) = err diff --git a/codex-rs/thread-store/src/local/delete_thread.rs b/codex-rs/thread-store/src/local/delete_thread.rs index 5b3e3ecfde..df75df56c5 100644 --- a/codex-rs/thread-store/src/local/delete_thread.rs +++ b/codex-rs/thread-store/src/local/delete_thread.rs @@ -144,9 +144,13 @@ async fn delete_thread_after_reference_check( }); } } - // Stop the live writer before removing files. The per-thread lock keeps new writes and - // replacements out while we find paths and clean up rollout files and history rows. - store.live_recorders.lock().await.remove(&thread_id); + // Drop the recorder before removing files, but retain its writer lock until cleanup finishes. + let _writer_lock = store + .live_recorders + .lock() + .await + .remove(&thread_id) + .and_then(|entry| entry._writer_lock); let found_rollout_path = !rollout_paths.is_empty(); for rollout_path in rollout_paths { delete_rollout_file(store, rollout_path.as_path(), thread_id)?; @@ -218,11 +222,14 @@ mod tests { use codex_protocol::ThreadId; use codex_protocol::protocol::HistoryPosition; use codex_protocol::protocol::ThreadHistoryMode; + use codex_protocol::protocol::ThreadMemoryMode; use pretty_assertions::assert_eq; use tempfile::TempDir; use uuid::Uuid; use super::*; + use crate::ResumeThreadParams; + use crate::ThreadPersistenceMetadata; use crate::ThreadStore; use crate::local::LocalThreadStore; use crate::local::test_support::test_config; @@ -406,7 +413,7 @@ mod tests { let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); let uuid = Uuid::from_u128(306); let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id"); - write_session_file_with_history_mode( + let rollout_path = write_session_file_with_history_mode( home.path(), "2025-01-03T12-00-00", uuid, @@ -439,10 +446,31 @@ mod tests { .await .expect("insert projection state"); + store + .resume_thread(ResumeThreadParams { + thread_id, + rollout_path: Some(rollout_path), + history: None, + include_archived: false, + metadata: ThreadPersistenceMetadata { + cwd: Some(home.path().to_path_buf()), + model_provider: "test-provider".to_string(), + memory_mode: ThreadMemoryMode::Enabled, + }, + }) + .await + .expect("resume paginated writer before deletion"); + let lock_path = home + .path() + .join("thread-writer-locks") + .join(format!("{thread_id}.lock")); + assert!(lock_path.exists()); + store .delete_thread(DeleteThreadParams { thread_id }) .await .expect("delete thread"); + assert!(!lock_path.exists()); let counts = sqlx::query_as::<_, (i64, i64, i64)>( r#" diff --git a/codex-rs/thread-store/src/local/live_writer.rs b/codex-rs/thread-store/src/local/live_writer.rs index e4d49f33d3..6202c8fc83 100644 --- a/codex-rs/thread-store/src/local/live_writer.rs +++ b/codex-rs/thread-store/src/local/live_writer.rs @@ -30,9 +30,14 @@ pub(super) async fn create_thread( let _live_writer_guard = store.live_writer_locks.lock(thread_id).await; let history_mode = params.history_mode; store.ensure_live_recorder_absent(thread_id).await?; + let writer_lock = if matches!(history_mode, ThreadHistoryMode::Paginated) { + Some(store.writer_lock_coordinator.acquire(thread_id)?) + } else { + None + }; let recorder = create_thread::create_thread(store, params).await?; store - .insert_live_recorder(thread_id, recorder, history_mode) + .insert_live_recorder(thread_id, recorder, history_mode, writer_lock) .await } @@ -98,13 +103,18 @@ pub(super) async fn resume_thread( model_provider_id: params.metadata.model_provider.clone(), generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled), }; + let writer_lock = if matches!(history_mode, ThreadHistoryMode::Paginated) { + Some(store.writer_lock_coordinator.acquire(params.thread_id)?) + } else { + None + }; let recorder = RolloutRecorder::new(&config, RolloutRecorderParams::resume(rollout_path)) .await .map_err(|err| ThreadStoreError::Internal { message: format!("failed to resume local thread recorder: {err}"), })?; store - .insert_live_recorder(params.thread_id, recorder, history_mode) + .insert_live_recorder(params.thread_id, recorder, history_mode, writer_lock) .await } diff --git a/codex-rs/thread-store/src/local/mod.rs b/codex-rs/thread-store/src/local/mod.rs index 3ebc837829..8a2cad525f 100644 --- a/codex-rs/thread-store/src/local/mod.rs +++ b/codex-rs/thread-store/src/local/mod.rs @@ -14,6 +14,7 @@ mod thread_history; mod thread_history_materialization; mod unarchive_thread; mod update_thread_metadata; +mod writer_lock; #[cfg(test)] mod test_support; @@ -59,6 +60,8 @@ use crate::ThreadStoreFuture; use crate::ThreadStoreResult; use crate::TurnPage; use crate::UpdateThreadMetadataParams; +use crate::local::writer_lock::WriterLockCoordinator; +use crate::local::writer_lock::WriterLockGuard; /// Local filesystem/SQLite-backed implementation of [`ThreadStore`]. /// @@ -78,6 +81,7 @@ pub struct LocalThreadStore { pub(super) config: LocalThreadStoreConfig, live_recorders: Arc>>, live_writer_locks: Arc, + writer_lock_coordinator: Arc, state_db: Option, thread_history_db: Arc>, } @@ -88,6 +92,7 @@ struct LiveRecorderEntry { // canonical SessionMeta is durable. Retain the mode captured when live persistence was opened // so missing SQLite rows can still be seeded. history_mode: ThreadHistoryMode, + _writer_lock: Option, } #[derive(Default)] @@ -150,10 +155,12 @@ impl std::fmt::Debug for LocalThreadStore { impl LocalThreadStore { /// Create a local store using an already initialized state DB handle. pub fn new(config: LocalThreadStoreConfig, state_db: Option) -> Self { + let writer_lock_coordinator = Arc::new(WriterLockCoordinator::new(&config.codex_home)); Self { config, live_recorders: Arc::new(Mutex::new(HashMap::new())), live_writer_locks: Arc::new(LiveWriterLocks::default()), + writer_lock_coordinator, state_db, thread_history_db: Arc::new(OnceCell::new()), } @@ -208,11 +215,12 @@ impl LocalThreadStore { Ok(()) } - pub(super) async fn insert_live_recorder( + async fn insert_live_recorder( &self, thread_id: ThreadId, recorder: RolloutRecorder, history_mode: ThreadHistoryMode, + writer_lock: Option, ) -> ThreadStoreResult<()> { match self.live_recorders.lock().await.entry(thread_id) { Entry::Occupied(entry) => Err(ThreadStoreError::InvalidRequest { @@ -222,6 +230,7 @@ impl LocalThreadStore { entry.insert(LiveRecorderEntry { recorder, history_mode, + _writer_lock: writer_lock, }); Ok(()) } @@ -963,56 +972,77 @@ mod tests { async fn create_thread_rejects_missing_cwd() { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); - let thread_id = ThreadId::default(); - let mut params = create_thread_params(thread_id); - params.metadata.cwd = None; + for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] { + let thread_id = ThreadId::default(); + let mut params = create_thread_params(thread_id); + params.history_mode = history_mode; + params.metadata.cwd = None; - let err = store - .create_thread(params) - .await - .expect_err("local thread store should require cwd"); + let err = store + .create_thread(params) + .await + .expect_err("local thread store should require cwd"); - assert!(matches!( - err, - ThreadStoreError::InvalidRequest { message } - if message == "local thread store requires a cwd" - )); + assert!(matches!( + err, + ThreadStoreError::InvalidRequest { message } + if message == "local thread store requires a cwd" + )); + + let mut valid_params = create_thread_params(thread_id); + valid_params.history_mode = history_mode; + store + .create_thread(valid_params) + .await + .expect("failed initialization should release writer ownership"); + } } #[tokio::test] async fn discard_thread_drops_unmaterialized_live_writer() { let home = TempDir::new().expect("temp dir"); let store = LocalThreadStore::new(test_config(home.path()), /*state_db*/ None); - let thread_id = ThreadId::default(); + for history_mode in [ThreadHistoryMode::Legacy, ThreadHistoryMode::Paginated] { + let thread_id = ThreadId::default(); + let mut params = create_thread_params(thread_id); + params.history_mode = history_mode; - store - .create_thread(create_thread_params(thread_id)) - .await - .expect("create live thread"); - let rollout_path = store - .live_rollout_path(thread_id) - .await - .expect("load rollout path"); - store - .discard_thread(thread_id) - .await - .expect("discard live thread"); - - assert!( - !tokio::fs::try_exists(rollout_path.as_path()) + store + .create_thread(params) .await - .expect("check rollout path") - ); - let err = store - .append_items(AppendThreadItemsParams { - thread_id, - items: vec![user_message_item("write after discard")], - }) - .await - .expect_err("discard should remove the live thread writer"); - assert!( - matches!(err, ThreadStoreError::ThreadNotFound { thread_id: missing } if missing == thread_id) - ); + .expect("create live thread"); + let rollout_path = store + .live_rollout_path(thread_id) + .await + .expect("load rollout path"); + assert!(!rollout_path.exists()); + + let lock_path = home + .path() + .join("thread-writer-locks") + .join(format!("{thread_id}.lock")); + assert_eq!( + lock_path.exists(), + matches!(history_mode, ThreadHistoryMode::Paginated) + ); + store + .discard_thread(thread_id) + .await + .expect("discard live thread"); + + assert!(!rollout_path.exists()); + assert!(!lock_path.exists()); + let err = store + .append_items(AppendThreadItemsParams { + thread_id, + items: vec![user_message_item("write after discard")], + }) + .await + .expect_err("discard should remove the live thread writer"); + assert!( + matches!(err, ThreadStoreError::ThreadNotFound { thread_id: missing } if missing == thread_id) + ); + } } #[tokio::test] diff --git a/codex-rs/thread-store/src/local/writer_lock.rs b/codex-rs/thread-store/src/local/writer_lock.rs new file mode 100644 index 0000000000..f6b2c8564b --- /dev/null +++ b/codex-rs/thread-store/src/local/writer_lock.rs @@ -0,0 +1,194 @@ +use std::fs; +use std::fs::File; +use std::fs::OpenOptions; +use std::io; +use std::path::Path; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::Ordering; + +use codex_protocol::ThreadId; +use tracing::warn; + +use crate::ThreadStoreError; +use crate::ThreadStoreResult; + +const WRITER_LOCK_DIR: &str = "thread-writer-locks"; +const COORDINATION_LOCK_FILE: &str = ".coordination.lock"; + +pub(super) struct WriterLockCoordinator { + directory: PathBuf, + cleanup_attempted: AtomicBool, +} + +pub(super) struct WriterLockGuard { + coordinator: Arc, + path: PathBuf, + file: Option, +} + +impl WriterLockCoordinator { + pub(super) fn new(codex_home: &Path) -> Self { + Self { + directory: codex_home.join(WRITER_LOCK_DIR), + cleanup_attempted: AtomicBool::new(false), + } + } + + pub(super) fn acquire( + self: &Arc, + thread_id: ThreadId, + ) -> ThreadStoreResult { + 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 path = self.directory.join(format!("{thread_id}.lock")); + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to open thread writer lock {}: {err}", + path.display() + ), + })?; + + match file.try_lock() { + Ok(()) => {} + Err(std::fs::TryLockError::WouldBlock) => { + return Err(ThreadStoreError::Conflict { + message: format!("thread {thread_id} already has an active writer"), + }); + } + Err(std::fs::TryLockError::Error(err)) => { + return Err(ThreadStoreError::Internal { + message: format!( + "failed to acquire thread writer lock {}: {err}", + path.display() + ), + }); + } + } + + drop(coordination_lock); + Ok(WriterLockGuard { + coordinator: Arc::clone(self), + path, + file: Some(file), + }) + } + + fn lock_coordination(&self) -> ThreadStoreResult { + fs::create_dir_all(&self.directory).map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to create thread writer lock directory {}: {err}", + self.directory.display() + ), + })?; + let path = self.directory.join(COORDINATION_LOCK_FILE); + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to open thread writer coordination lock {}: {err}", + path.display() + ), + })?; + file.lock().map_err(|err| ThreadStoreError::Internal { + message: format!( + "failed to acquire thread writer coordination lock {}: {err}", + path.display() + ), + })?; + Ok(file) + } + + fn remove_stale_thread_locks(&self) -> io::Result<()> { + for entry in fs::read_dir(&self.directory)? { + let entry = entry?; + let Some(file_name) = entry.file_name().to_str().map(str::to_owned) else { + continue; + }; + let Some(thread_id) = file_name.strip_suffix(".lock") else { + continue; + }; + if ThreadId::from_string(thread_id).is_err() { + continue; + } + + let path = entry.path(); + let file = match OpenOptions::new().read(true).write(true).open(&path) { + Ok(file) => file, + Err(err) if err.kind() == io::ErrorKind::NotFound => continue, + Err(err) => { + warn!( + "failed to inspect thread writer lock {}: {err}", + path.display() + ); + continue; + } + }; + match file.try_lock() { + Ok(()) => { + drop(file); + if let Err(err) = fs::remove_file(&path) + && err.kind() != io::ErrorKind::NotFound + { + warn!( + "failed to remove stale thread writer lock {}: {err}", + path.display() + ); + } + } + Err(std::fs::TryLockError::WouldBlock) => {} + Err(std::fs::TryLockError::Error(err)) => { + warn!( + "failed to inspect thread writer lock {}: {err}", + path.display() + ); + } + } + } + Ok(()) + } +} + +impl Drop for WriterLockGuard { + fn drop(&mut self) { + let coordination_lock = match self.coordinator.lock_coordination() { + Ok(lock) => lock, + Err(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) + && err.kind() != io::ErrorKind::NotFound + { + warn!( + "failed to remove thread writer lock {}: {err}", + self.path.display() + ); + } + drop(coordination_lock); + } +} + +#[cfg(test)] +#[path = "writer_lock_tests.rs"] +mod tests; diff --git a/codex-rs/thread-store/src/local/writer_lock_tests.rs b/codex-rs/thread-store/src/local/writer_lock_tests.rs new file mode 100644 index 0000000000..6406314225 --- /dev/null +++ b/codex-rs/thread-store/src/local/writer_lock_tests.rs @@ -0,0 +1,81 @@ +use std::fs; +use std::sync::Arc; + +use codex_protocol::ThreadId; +use tempfile::TempDir; + +use super::COORDINATION_LOCK_FILE; +use super::WRITER_LOCK_DIR; +use super::WriterLockCoordinator; +use crate::ThreadStoreError; + +#[test] +fn writer_locks_reject_competing_owners_and_release_their_files() { + let home = TempDir::new().expect("temp dir"); + let primary = Arc::new(WriterLockCoordinator::new(home.path())); + let secondary = Arc::new(WriterLockCoordinator::new(home.path())); + let thread_id = ThreadId::default(); + let other_thread_id = ThreadId::default(); + + let owner = primary.acquire(thread_id).expect("acquire writer lock"); + let lock_path = home + .path() + .join(WRITER_LOCK_DIR) + .join(format!("{thread_id}.lock")); + assert!(lock_path.exists()); + + let err = match secondary.acquire(thread_id) { + Ok(_) => panic!("competing owner should fail"), + Err(err) => err, + }; + assert!(matches!(err, ThreadStoreError::Conflict { .. })); + let other_owner = secondary + .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"); + drop(next_owner); + drop(other_owner); + + let entries = fs::read_dir(home.path().join(WRITER_LOCK_DIR)) + .expect("read lock directory") + .map(|entry| entry.expect("lock directory entry").file_name()) + .collect::>(); + assert_eq!(entries, vec![COORDINATION_LOCK_FILE]); +} + +#[test] +fn first_acquisition_removes_stale_locks_without_removing_active_locks() { + let home = TempDir::new().expect("temp dir"); + let primary = Arc::new(WriterLockCoordinator::new(home.path())); + let active_thread_id = ThreadId::default(); + let active_owner = primary + .acquire(active_thread_id) + .expect("acquire active writer lock"); + + let stale_thread_id = ThreadId::default(); + let stale_path = home + .path() + .join(WRITER_LOCK_DIR) + .join(format!("{stale_thread_id}.lock")); + fs::File::create(&stale_path).expect("create stale writer lock"); + + let secondary = Arc::new(WriterLockCoordinator::new(home.path())); + let secondary_owner = secondary + .acquire(ThreadId::default()) + .expect("acquire writer lock after cleanup"); + + assert!(!stale_path.exists()); + let err = match secondary.acquire(active_thread_id) { + Ok(_) => panic!("active writer should survive cleanup"), + Err(err) => err, + }; + assert!(matches!(err, ThreadStoreError::Conflict { .. })); + + drop(secondary_owner); + drop(active_owner); +}