From faa727965b53d1a037c731b0029afccbba3722b6 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 2 Sep 2026 01:59:32 +0900 Subject: [PATCH] fix: fence orphan recovery claims --- crates/workspace-server/src/server.rs | 96 +++++++++++++++++-- .../workspace-server/src/workdir_removal.rs | 21 ++-- docs/design/durable-operations.md | 2 +- 3 files changed, 105 insertions(+), 14 deletions(-) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 9c08f871..26bf67cb 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -9737,19 +9737,21 @@ fn execute_reserved_workdir_removal_with_provider( if operation.state == WorkdirRemovalOperationState::Completed { return Ok(operation); } - let operation = if recovery { - let prior_owner_is_orphaned = if operation.state == WorkdirRemovalOperationState::Pending { - workdir_removal_attempt_is_orphaned(&operation)? - } else { - false - }; + let operation = if recovery && operation.state == WorkdirRemovalOperationState::Pending { + if !workdir_removal_attempt_is_orphaned(&operation)? { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir removal operation `{}` still has a live attempt owner", + operation.operation_id + ))); + } api.config_store .reclaim_workdir_removal_attempt_for_recovery( &operation.workspace_id, &operation.operation_id, &operation.request_fingerprint, api.workdir_remove_attempt_owner, - prior_owner_is_orphaned, + operation.attempt_owner, + operation.attempt_count, )? } else { api.config_store.begin_workdir_removal_attempt( @@ -22471,6 +22473,7 @@ mod tests { >, >, cleanup_calls: std::sync::atomic::AtomicUsize, + observation_delay: std::time::Duration, } impl FakeWorkdirRemovalProvider { @@ -22482,9 +22485,15 @@ mod tests { observation: Mutex::new(Some(Ok(observation))), cleanup: Mutex::new(Some(Ok(cleanup))), cleanup_calls: std::sync::atomic::AtomicUsize::new(0), + observation_delay: std::time::Duration::ZERO, } } + fn with_observation_delay(mut self, delay: std::time::Duration) -> Self { + self.observation_delay = delay; + self + } + fn cleanup_calls(&self) -> usize { self.cleanup_calls.load(std::sync::atomic::Ordering::SeqCst) } @@ -22497,6 +22506,7 @@ mod tests { _working_directory_id: &str, ) -> std::result::Result { + std::thread::sleep(self.observation_delay); self.observation .lock() .unwrap() @@ -22700,6 +22710,78 @@ mod tests { assert_eq!(unsupported_provider.cleanup_calls(), 1); } + #[tokio::test] + async fn concurrent_orphan_recovery_runs_delayed_provider_cleanup_once() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let (reserved, clean_summary) = reserve_removal_fixture(&api, "orphan-race"); + let interrupted = api + .config_store + .begin_workdir_removal_attempt( + &reserved.workspace_id, + &reserved.operation_id, + &reserved.request_fingerprint, + WorkdirRemovalAttemptOwner { + process_id: u32::MAX, + process_start_marker: 1, + }, + ) + .unwrap(); + let provider = Arc::new( + FakeWorkdirRemovalProvider::new( + workdir_removal_result( + WorkerOperationState::Accepted, + Some(clean_summary), + Vec::new(), + ), + workdir_removal_result(WorkerOperationState::Accepted, None, Vec::new()), + ) + .with_observation_delay(std::time::Duration::from_millis(100)), + ); + let barrier = Arc::new(std::sync::Barrier::new(3)); + let mut callers = Vec::new(); + for _ in 0..2 { + let api = api.clone(); + let operation = interrupted.clone(); + let provider = provider.clone(); + let barrier = barrier.clone(); + callers.push(std::thread::spawn(move || { + barrier.wait(); + execute_reserved_workdir_removal_with_provider( + &api, + operation, + true, + provider.as_ref(), + ) + })); + } + barrier.wait(); + let results = callers + .into_iter() + .map(|caller| caller.join().unwrap()) + .collect::>(); + assert_eq!( + results + .iter() + .filter(|result| { + result.as_ref().is_ok_and(|operation| { + operation.disposition == Some(WorkdirRemovalDisposition::Removed) + }) + }) + .count(), + 1 + ); + assert_eq!( + results + .iter() + .filter(|result| matches!(result, Err(Error::WorkdirAttachmentConflict(_)))) + .count(), + 1 + ); + assert_eq!(provider.cleanup_calls(), 1); + } + #[test] fn only_exact_provider_not_found_is_removal_evidence() { let not_found = crate::hosts::RuntimeWorkingDirectoryResult { diff --git a/crates/workspace-server/src/workdir_removal.rs b/crates/workspace-server/src/workdir_removal.rs index ae8cfa04..e8c104f4 100644 --- a/crates/workspace-server/src/workdir_removal.rs +++ b/crates/workspace-server/src/workdir_removal.rs @@ -268,7 +268,7 @@ impl SqliteWorkspaceStore { operation_id, request_fingerprint, owner, - false, + None, ) } @@ -278,14 +278,15 @@ impl SqliteWorkspaceStore { operation_id: &str, request_fingerprint: &str, owner: WorkdirRemovalAttemptOwner, - prior_owner_is_orphaned: bool, + expected_prior_owner: Option, + expected_attempt_count: u64, ) -> Result { self.begin_workdir_removal_attempt_inner( workspace_id, operation_id, request_fingerprint, owner, - prior_owner_is_orphaned, + Some((expected_prior_owner, expected_attempt_count)), ) } @@ -295,7 +296,7 @@ impl SqliteWorkspaceStore { operation_id: &str, request_fingerprint: &str, owner: WorkdirRemovalAttemptOwner, - prior_owner_is_orphaned: bool, + recovery_expected: Option<(Option, u64)>, ) -> Result { self.with_conn_mut(|conn| { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; @@ -310,9 +311,17 @@ impl SqliteWorkspaceStore { "Workdir removal operation `{operation_id}` is not retryable" ))); } - if operation.state == WorkdirRemovalOperationState::Pending + if let Some((expected_owner, expected_attempt_count)) = recovery_expected { + if operation.state != WorkdirRemovalOperationState::Pending + || operation.attempt_owner != expected_owner + || operation.attempt_count != expected_attempt_count + { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir removal operation `{operation_id}` changed after orphan proof" + ))); + } + } else if operation.state == WorkdirRemovalOperationState::Pending && operation.attempt_count > 0 - && !prior_owner_is_orphaned { return Err(Error::WorkdirAttachmentConflict(format!( "Workdir removal operation `{operation_id}` already has an active attempt" diff --git a/docs/design/durable-operations.md b/docs/design/durable-operations.md index 3455aab2..a582e63f 100644 --- a/docs/design/durable-operations.md +++ b/docs/design/durable-operations.md @@ -13,7 +13,7 @@ The record stores only facts that affect identity, authorization, replay, or the - explicit retryability, bounded failure category, and bounded disposition; - a factual checkpoint only when a non-idempotent provider effect cannot be safely re-observed or repeated. -A fingerprint excludes Server-generated result identifiers, attempt data, diagnostics, and fresh observations. Reusing one operation identity with a different fingerprint is an error. A completed exact retry replays the committed bounded result. A durable one-pending-operation constraint plus an atomic attempt claim prevents concurrent callers from entering the provider side effect for the same Workdir; the in-process resource lock is an additional serialization layer, not the sole authority. Each active attempt persists the Server process ID and process-start marker. Recovery reclaims only an owner proven missing or replaced; a live or unobservable owner is never stolen. +A fingerprint excludes Server-generated result identifiers, attempt data, diagnostics, and fresh observations. Reusing one operation identity with a different fingerprint is an error. A completed exact retry replays the committed bounded result. A durable one-pending-operation constraint plus an atomic attempt claim prevents concurrent callers from entering the provider side effect for the same Workdir; the in-process resource lock is an additional serialization layer, not the sole authority. Each active attempt persists the Server process ID and process-start marker. Recovery reclaims only an owner proven missing or replaced; a live or unobservable owner is never stolen. The reclaim transaction CAS-checks the exact proved owner snapshot and attempt count so a stale orphan proof cannot overwrite a newer live claim. `pending` means only that the intent remains open. Function names, validation steps, and provider-call positions are not persisted as lifecycle stages. `failed` records the latest terminal attempt outcome; retryability remains separate metadata. `completed` means the required domain result and disposition are durably committed.