From d2ffbf2c401ee8c3ee74aaeea29e5c6cca161347 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 2 Sep 2026 01:40:48 +0900 Subject: [PATCH] fix: guard Workdir removal recovery ownership --- crates/workspace-server/src/server.rs | 158 +++++++++++++++++- crates/workspace-server/src/store.rs | 147 +++++++++++----- .../src/workdir_create_operations.rs | 84 +++++++++- .../workspace-server/src/workdir_removal.rs | 142 ++++++++++++++-- docs/design/durable-operations.md | 4 +- 5 files changed, 479 insertions(+), 56 deletions(-) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index f8576579..9c08f871 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -142,8 +142,8 @@ use crate::store::{ WorkspaceResourceKind, }; use crate::workdir_removal::{ - WorkdirRemovalDisposition, WorkdirRemovalOperation, WorkdirRemovalOperationState, - workdir_removal_intent, + WorkdirRemovalAttemptOwner, WorkdirRemovalDisposition, WorkdirRemovalOperation, + WorkdirRemovalOperationState, workdir_removal_intent, }; use crate::workspace_catalog::{WorkspaceCatalogService, WorkspaceCreateRequest}; use crate::{Error, Result}; @@ -519,6 +519,7 @@ pub struct WorkspaceApi { workdir_session_locks: Arc>>>>, worker_remove_locks: Arc>>>>, workdir_remove_locks: Arc>>>>, + workdir_remove_attempt_owner: WorkdirRemovalAttemptOwner, worker_control_locks: Arc>>>>, } @@ -1640,6 +1641,7 @@ impl WorkspaceApi { workdir_session_locks: Arc::new(Mutex::new(HashMap::new())), worker_remove_locks: Arc::new(Mutex::new(HashMap::new())), workdir_remove_locks: Arc::new(Mutex::new(HashMap::new())), + workdir_remove_attempt_owner: current_workdir_removal_attempt_owner()?, worker_control_locks: Arc::new(Mutex::new(HashMap::new())), }; if let Some(dispatcher) = worker_remove_dispatcher { @@ -9320,6 +9322,16 @@ async fn create_workspace_working_directory( ) }); } + let reserved = if reserved.state == "failed" { + api.config_store.begin_failed_workdir_create_retry( + workspace_id, + &operation_id, + &request_fingerprint, + &now_registry_timestamp(), + )? + } else { + reserved + }; let runtime = match api .runtime @@ -9589,6 +9601,91 @@ fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static st } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum WorkdirRemovalOwnerObservation { + Running { process_start_marker: u64 }, + Missing, + Unobservable, +} + +fn current_workdir_removal_attempt_owner() -> Result { + let process_id = std::process::id(); + match observe_workdir_removal_owner(process_id) { + WorkdirRemovalOwnerObservation::Running { + process_start_marker, + } => Ok(WorkdirRemovalAttemptOwner { + process_id, + process_start_marker, + }), + WorkdirRemovalOwnerObservation::Missing | WorkdirRemovalOwnerObservation::Unobservable => { + Err(Error::Config( + "current Server process identity is unavailable for durable Workdir removal" + .to_string(), + )) + } + } +} + +fn workdir_removal_attempt_is_orphaned(operation: &WorkdirRemovalOperation) -> Result { + let Some(owner) = operation.attempt_owner else { + return Ok(operation.attempt_count == 0); + }; + match observe_workdir_removal_owner(owner.process_id) { + WorkdirRemovalOwnerObservation::Running { + process_start_marker, + } if process_start_marker == owner.process_start_marker => Ok(false), + WorkdirRemovalOwnerObservation::Running { .. } + | WorkdirRemovalOwnerObservation::Missing => Ok(true), + WorkdirRemovalOwnerObservation::Unobservable => Err(Error::RegistryInconsistency( + "prior Workdir removal attempt owner liveness is unobservable".to_string(), + )), + } +} + +#[cfg(target_os = "linux")] +fn observe_workdir_removal_owner(process_id: u32) -> WorkdirRemovalOwnerObservation { + let stat = match std::fs::read_to_string(format!("/proc/{process_id}/stat")) { + Ok(stat) => stat, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + return if process_id != std::process::id() + && std::fs::read_to_string("/proc/self/stat") + .ok() + .and_then(|stat| parse_linux_process_start_marker(&stat)) + .is_some() + { + WorkdirRemovalOwnerObservation::Missing + } else { + WorkdirRemovalOwnerObservation::Unobservable + }; + } + Err(_) => return WorkdirRemovalOwnerObservation::Unobservable, + }; + parse_linux_process_start_marker(&stat) + .map( + |process_start_marker| WorkdirRemovalOwnerObservation::Running { + process_start_marker, + }, + ) + .unwrap_or(WorkdirRemovalOwnerObservation::Unobservable) +} + +#[cfg(target_os = "linux")] +fn parse_linux_process_start_marker(stat: &str) -> Option { + let (_, tail) = stat.rsplit_once(") ")?; + tail.split_whitespace().nth(19)?.parse().ok() +} + +#[cfg(not(target_os = "linux"))] +fn observe_workdir_removal_owner(process_id: u32) -> WorkdirRemovalOwnerObservation { + if process_id == std::process::id() { + WorkdirRemovalOwnerObservation::Running { + process_start_marker: 0, + } + } else { + WorkdirRemovalOwnerObservation::Unobservable + } +} + trait WorkdirRemovalRuntimeProvider: Send + Sync { fn observe_workdir( &self, @@ -9641,17 +9738,25 @@ fn execute_reserved_workdir_removal_with_provider( 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 + }; 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, )? } else { api.config_store.begin_workdir_removal_attempt( &operation.workspace_id, &operation.operation_id, &operation.request_fingerprint, + api.workdir_remove_attempt_owner, )? }; let guards = match api.config_store.workdir_removal_guards(&operation) { @@ -22703,6 +22808,51 @@ mod tests { ); } + #[tokio::test] + async fn recovery_does_not_reclaim_live_attempt_owner() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + seed_cleanup_workdir(&api, "live-owner-workdir", "present", "clean"); + let record = api + .store + .get_workdir_registry(&api.config.workspace_id, "live-owner-workdir") + .unwrap() + .unwrap(); + let intent = workdir_removal_intent( + &record, + "account:owner", + "do not steal a live provider call", + ) + .unwrap(); + let reserved = api + .config_store + .reserve_workdir_removal_operation(&intent) + .unwrap(); + api.config_store + .begin_workdir_removal_attempt( + &reserved.workspace_id, + &reserved.operation_id, + &reserved.request_fingerprint, + api.workdir_remove_attempt_owner, + ) + .unwrap(); + + recover_workdir_removals(&api).unwrap(); + + let operation = api + .config_store + .get_workdir_removal_operation(&api.config.workspace_id, &intent.operation_id) + .unwrap() + .unwrap(); + assert_eq!(operation.state, WorkdirRemovalOperationState::Pending); + assert_eq!(operation.attempt_count, 1); + assert_eq!( + operation.attempt_owner, + Some(api.workdir_remove_attempt_owner) + ); + } + #[tokio::test] async fn recovery_retries_same_operation_and_retains_unknown_provider_result() { let workspace = tempfile::tempdir().unwrap(); @@ -22726,6 +22876,10 @@ mod tests { &reserved.workspace_id, &reserved.operation_id, &reserved.request_fingerprint, + WorkdirRemovalAttemptOwner { + process_id: u32::MAX, + process_start_marker: 1, + }, ) .unwrap(); assert_eq!(interrupted_attempt.attempt_count, 1); diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index de2cd372..137487b6 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -10104,48 +10104,117 @@ mod tests { } #[test] - fn schema_v49_upgrades_v48_with_durable_workdir_removal_authority() { - let conn = Connection::open_in_memory().unwrap(); - configure_sqlite(&conn).unwrap(); - apply_migrations_through(&conn, 48).unwrap(); - assert_eq!(current_schema_version(&conn).unwrap(), 48); - assert!(!table_exists(&conn, "workdir_removal_operations").unwrap()); - - apply_migrations(&conn).unwrap(); - - assert_eq!(current_schema_version(&conn).unwrap(), 49); - assert!(table_exists(&conn, "workdir_removal_operations").unwrap()); - let columns = table_columns(&conn, "workdir_removal_operations").unwrap(); - for required in [ - "workspace_id", - "operation_id", - "request_fingerprint", - "workdir_id", - "runtime_id", - "repository_id", - "materialization_fingerprint", - "source_actor", - "reason", - "state", - "attempt_count", - "retryable", - "disposition", - "failure_category", - "created_at", - "updated_at", - "completed_at", - ] { - assert!( - columns.iter().any(|column| column == required), - "missing {required}" - ); + fn schema_v49_upgrades_persisted_v48_workdir_fixture() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server-v48.db"); + { + let conn = Connection::open(&path).unwrap(); + configure_sqlite(&conn).unwrap(); + apply_migrations_through(&conn, 48).unwrap(); + assert_eq!(current_schema_version(&conn).unwrap(), 48); + assert!(!table_exists(&conn, "workdir_removal_operations").unwrap()); + conn.execute_batch( + r#" + INSERT INTO accounts ( + account_id, kind, handle, display_name, created_at, updated_at + ) VALUES ('owner-account', 'user', 'owner', 'Owner', '1', '1'); + INSERT INTO workspaces ( + workspace_id, owner_account_id, display_name, state, created_at, updated_at + ) VALUES ('workspace-a', 'owner-account', 'Workspace A', 'active', '1', '1'); + INSERT INTO repositories ( + workspace_id, repository_id, name, kind, provider, uri, + source_kind, source_uri, default_ref, source_revision, + source_fingerprint, observed_status, observed_at, created_at, updated_at + ) VALUES ( + 'workspace-a', 'repository-a', 'Repository A', 'git', 'local', '/repo-a', + 'local_path', '/repo-a', 'develop', 1, + 'sha256:source-a', 'unverified', NULL, '1', '1' + ); + INSERT INTO workdir_registry ( + workspace_id, workdir_id, runtime_id, repository_id, + creation_selector, creation_ref, creation_tree, + current_selector, current_ref, current_tree, + observed_at_epoch_seconds, materialization_status, cleanliness, + created_at, updated_at + ) VALUES ( + 'workspace-a', 'workdir-a', 'runtime-a', 'repository-a', + 'refs/heads/develop', 'abc', 'tree-a', + 'refs/heads/work', 'def', 'tree-b', + 1, 'present', 'clean', '1', '1' + ); + "#, + ) + .unwrap(); } - let foreign_key_failures: i64 = conn - .query_row("SELECT count(*) FROM pragma_foreign_key_check", [], |row| { - row.get(0) + + let store = SqliteWorkspaceStore::open(&path).unwrap(); + store + .with_conn(|conn| { + assert_eq!(current_schema_version(conn)?, 49); + assert!(table_exists(conn, "workdir_removal_operations")?); + let columns = table_columns(conn, "workdir_removal_operations")?; + for required in [ + "workspace_id", + "operation_id", + "request_fingerprint", + "workdir_id", + "runtime_id", + "repository_id", + "materialization_fingerprint", + "source_actor", + "reason", + "state", + "attempt_count", + "retryable", + "disposition", + "failure_category", + "attempt_owner_pid", + "attempt_owner_start_marker", + "created_at", + "updated_at", + "completed_at", + ] { + assert!( + columns.iter().any(|column| column == required), + "missing {required}" + ); + } + let preserved: (String, String, String) = conn.query_row( + "SELECT workspace_id, repository_id, materialization_status FROM workdir_registry WHERE workdir_id='workdir-a'", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + )?; + assert_eq!( + preserved, + ( + "workspace-a".to_string(), + "repository-a".to_string(), + "present".to_string(), + ) + ); + let foreign_key_failures: i64 = conn.query_row( + "SELECT count(*) FROM pragma_foreign_key_check", + [], + |row| row.get(0), + )?; + assert_eq!(foreign_key_failures, 0); + Ok(()) }) .unwrap(); - assert_eq!(foreign_key_failures, 0); + + let workdir = store + .get_workdir_registry("workspace-a", "workdir-a") + .unwrap() + .unwrap(); + let intent = crate::workdir_removal::workdir_removal_intent( + &workdir, + "migration-test", + "remove migrated Workdir", + ) + .unwrap(); + let operation = store.reserve_workdir_removal_operation(&intent).unwrap(); + assert_eq!(operation.workspace_id, "workspace-a"); + assert_eq!(operation.working_directory_id, "workdir-a"); } #[test] diff --git a/crates/workspace-server/src/workdir_create_operations.rs b/crates/workspace-server/src/workdir_create_operations.rs index 3ad73ab3..25fd910d 100644 --- a/crates/workspace-server/src/workdir_create_operations.rs +++ b/crates/workspace-server/src/workdir_create_operations.rs @@ -1,4 +1,4 @@ -use rusqlite::{OptionalExtension, params}; +use rusqlite::{OptionalExtension, TransactionBehavior, params}; use sha2::{Digest, Sha256}; use crate::store::WorkdirCreateOperationRecord; @@ -103,6 +103,65 @@ impl SqliteWorkspaceStore { }) } + pub fn begin_failed_workdir_create_retry( + &self, + workspace_id: &str, + operation_id: &str, + request_fingerprint: &str, + updated_at: &str, + ) -> Result { + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let operation = read_workdir_create_operation(&tx, workspace_id, operation_id)? + .ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workdir create operation `{operation_id}` disappeared before retry" + )) + })?; + if operation.request_fingerprint != request_fingerprint { + return Err(Error::InvalidInput(format!( + "Workdir create operation `{operation_id}` was reused with different input" + ))); + } + if operation.state != "failed" { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir create operation `{operation_id}` is not a failed retry" + ))); + } + let removal_pending: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM workdir_removal_operations WHERE workspace_id=?1 AND workdir_id=?2 AND state='pending')", + params![workspace_id, operation.working_directory_id], + |row| row.get(0), + )?; + if removal_pending { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir {} has a pending durable removal operation", + operation.working_directory_id + ))); + } + let changed = tx.execute( + r#"UPDATE workdir_create_operations + SET state='pending', failure=NULL, updated_at=?1 + WHERE workspace_id=?2 AND operation_id=?3 + AND request_fingerprint=?4 AND state='failed'"#, + params![updated_at, workspace_id, operation_id, request_fingerprint], + )?; + if changed != 1 { + return Err(Error::WorkdirAttachmentConflict(format!( + "Workdir create operation `{operation_id}` retry was claimed concurrently" + ))); + } + let updated = read_workdir_create_operation(&tx, workspace_id, operation_id)? + .ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workdir create operation `{operation_id}` disappeared after retry claim" + )) + })?; + tx.commit()?; + Ok(updated) + }) + } + pub fn bind_workdir_create_repository_access( &self, workspace_id: &str, @@ -406,11 +465,32 @@ mod tests { .unwrap(); assert_eq!(replayed, bound); assert_eq!(replayed.source_uri.as_deref(), Some("/tmp/repo")); + let failed = store + .finish_workdir_create_operation( + "workspace", + "call-1", + &record.request_fingerprint, + false, + Some("provider failed"), + "2026-08-24T00:00:03Z", + ) + .unwrap(); + assert_eq!(failed.state, "failed"); + let retry = store + .begin_failed_workdir_create_retry( + "workspace", + "call-1", + &record.request_fingerprint, + "2026-08-24T00:00:04Z", + ) + .unwrap(); + assert_eq!(retry.state, "pending"); + assert_eq!(retry.failure, None); assert_eq!( store .load_workdir_create_operation("workspace", "call-1") .unwrap(), - Some(bound.clone()) + Some(retry.clone()) ); let mut changed_input = record.clone(); changed_input.request_fingerprint = diff --git a/crates/workspace-server/src/workdir_removal.rs b/crates/workspace-server/src/workdir_removal.rs index e5eb0cf6..ae8cfa04 100644 --- a/crates/workspace-server/src/workdir_removal.rs +++ b/crates/workspace-server/src/workdir_removal.rs @@ -44,6 +44,12 @@ impl WorkdirRemovalDisposition { } } +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +pub struct WorkdirRemovalAttemptOwner { + pub process_id: u32, + pub process_start_marker: u64, +} + #[derive(Debug, Clone, PartialEq, Eq)] pub struct WorkdirRemovalIntent { pub operation_id: String, @@ -73,6 +79,7 @@ pub struct WorkdirRemovalOperation { pub retryable: bool, pub disposition: Option, pub failure_category: Option, + pub attempt_owner: Option, pub created_at: String, pub updated_at: String, pub completed_at: Option, @@ -102,6 +109,8 @@ CREATE TABLE workdir_removal_operations ( retryable INTEGER NOT NULL CHECK (retryable IN (0, 1)), disposition TEXT CHECK (disposition IN ('removed', 'retained', 'attention_required')), failure_category TEXT, + attempt_owner_pid INTEGER CHECK (attempt_owner_pid > 0), + attempt_owner_start_marker INTEGER CHECK (attempt_owner_start_marker >= 0), created_at TEXT NOT NULL, updated_at TEXT NOT NULL, completed_at TEXT, @@ -252,11 +261,13 @@ impl SqliteWorkspaceStore { workspace_id: &str, operation_id: &str, request_fingerprint: &str, + owner: WorkdirRemovalAttemptOwner, ) -> Result { self.begin_workdir_removal_attempt_inner( workspace_id, operation_id, request_fingerprint, + owner, false, ) } @@ -266,12 +277,15 @@ impl SqliteWorkspaceStore { workspace_id: &str, operation_id: &str, request_fingerprint: &str, + owner: WorkdirRemovalAttemptOwner, + prior_owner_is_orphaned: bool, ) -> Result { self.begin_workdir_removal_attempt_inner( workspace_id, operation_id, request_fingerprint, - true, + owner, + prior_owner_is_orphaned, ) } @@ -280,7 +294,8 @@ impl SqliteWorkspaceStore { workspace_id: &str, operation_id: &str, request_fingerprint: &str, - recovery: bool, + owner: WorkdirRemovalAttemptOwner, + prior_owner_is_orphaned: bool, ) -> Result { self.with_conn_mut(|conn| { let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; @@ -297,7 +312,7 @@ impl SqliteWorkspaceStore { } if operation.state == WorkdirRemovalOperationState::Pending && operation.attempt_count > 0 - && !recovery + && !prior_owner_is_orphaned { return Err(Error::WorkdirAttachmentConflict(format!( "Workdir removal operation `{operation_id}` already has an active attempt" @@ -305,8 +320,19 @@ impl SqliteWorkspaceStore { } let now = Utc::now().to_rfc3339(); tx.execute( - "UPDATE workdir_removal_operations SET state='pending', attempt_count=attempt_count+1, retryable=1, failure_category=NULL, disposition=NULL, updated_at=?1, completed_at=NULL WHERE workspace_id=?2 AND operation_id=?3 AND request_fingerprint=?4", - params![now, workspace_id, operation_id, request_fingerprint], + "UPDATE workdir_removal_operations SET state='pending', attempt_count=attempt_count+1, retryable=1, failure_category=NULL, disposition=NULL, attempt_owner_pid=?1, attempt_owner_start_marker=?2, updated_at=?3, completed_at=NULL WHERE workspace_id=?4 AND operation_id=?5 AND request_fingerprint=?6", + params![ + owner.process_id, + i64::try_from(owner.process_start_marker).map_err(|_| { + Error::InvalidInput( + "process start marker is out of SQLite range".to_string(), + ) + })?, + now, + workspace_id, + operation_id, + request_fingerprint, + ], )?; let operation = require_operation(&tx, workspace_id, operation_id, request_fingerprint)?; @@ -456,7 +482,7 @@ impl SqliteWorkspaceStore { } let now = Utc::now().to_rfc3339(); tx.execute( - "UPDATE workdir_removal_operations SET state='failed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7", + "UPDATE workdir_removal_operations SET state='failed', retryable=?1, disposition=?2, failure_category=?3, attempt_owner_pid=NULL, attempt_owner_start_marker=NULL, updated_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7", params![ retryable, WorkdirRemovalDisposition::AttentionRequired.as_str(), @@ -538,7 +564,7 @@ impl SqliteWorkspaceStore { } let now = Utc::now().to_rfc3339(); tx.execute( - "UPDATE workdir_removal_operations SET state='completed', retryable=?1, disposition=?2, failure_category=?3, updated_at=?4, completed_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7", + "UPDATE workdir_removal_operations SET state='completed', retryable=?1, disposition=?2, failure_category=?3, attempt_owner_pid=NULL, attempt_owner_start_marker=NULL, updated_at=?4, completed_at=?4 WHERE workspace_id=?5 AND operation_id=?6 AND request_fingerprint=?7", params![ retryable, disposition.as_str(), @@ -620,13 +646,16 @@ fn require_no_removal_blockers( UNION ALL SELECT 1 FROM worker_workdir_attachment_reservations WHERE workspace_id=?1 AND workdir_id=?2 + UNION ALL + SELECT 1 FROM workdir_create_operations + WHERE workspace_id=?1 AND working_directory_id=?2 AND state='pending' )"#, params![workspace_id, workdir_id], |row| row.get(0), )?; if blocked { return Err(Error::WorkdirAttachmentConflict(format!( - "Workdir {workdir_id} acquired an active or pending attachment during removal" + "Workdir {workdir_id} acquired active attachment or materialization authority during removal" ))); } Ok(()) @@ -784,6 +813,7 @@ fn operation_select_sql() -> &'static str { r#"SELECT operation_id, request_fingerprint, workspace_id, workdir_id, runtime_id, repository_id, materialization_fingerprint, source_actor, reason, state, attempt_count, retryable, disposition, failure_category, + attempt_owner_pid, attempt_owner_start_marker, created_at, updated_at, completed_at FROM workdir_removal_operations"# } @@ -795,6 +825,26 @@ fn read_operation(row: &rusqlite::Row<'_>) -> rusqlite::Result(10)?; + let attempt_owner_pid = row.get::<_, Option>(14)?; + let attempt_owner_start_marker = row.get::<_, Option>(15)?; + let attempt_owner = match (attempt_owner_pid, attempt_owner_start_marker) { + (None, None) => None, + (Some(process_id), Some(process_start_marker)) => Some(WorkdirRemovalAttemptOwner { + process_id: process_id + .try_into() + .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(14, process_id))?, + process_start_marker: process_start_marker + .try_into() + .map_err(|_| rusqlite::Error::IntegralValueOutOfRange(15, process_start_marker))?, + }), + _ => { + return Err(rusqlite::Error::FromSqlConversionFailure( + 14, + rusqlite::types::Type::Integer, + "incomplete Workdir removal attempt owner".into(), + )); + } + }; Ok(WorkdirRemovalOperation { operation_id: row.get(0)?, request_fingerprint: row.get(1)?, @@ -812,9 +862,10 @@ fn read_operation(row: &rusqlite::Row<'_>) -> rusqlite::Result WorkdirRemovalAttemptOwner { + WorkdirRemovalAttemptOwner { + process_id: 100, + process_start_marker: 200, + } + } + async fn seeded_store() -> (SqliteWorkspaceStore, WorkdirRegistryRecord) { let store = SqliteWorkspaceStore::in_memory().unwrap(); store @@ -950,6 +1008,7 @@ mod tests { &reserved.workspace_id, &reserved.operation_id, &reserved.request_fingerprint, + attempt_owner(), ) .unwrap(); assert_eq!(first.attempt_count, 1); @@ -962,6 +1021,7 @@ mod tests { &failed.workspace_id, &failed.operation_id, &failed.request_fingerprint, + attempt_owner(), ) .unwrap(); assert_eq!(retry.attempt_count, 2); @@ -981,6 +1041,7 @@ mod tests { &completed.workspace_id, &completed.operation_id, &completed.request_fingerprint, + attempt_owner(), ) .unwrap(); assert_eq!(replay, completed); @@ -1009,6 +1070,7 @@ mod tests { &operation.workspace_id, &operation.operation_id, &operation.request_fingerprint, + attempt_owner(), ); if claim.is_ok() { provider_calls.fetch_add(1, Ordering::SeqCst); @@ -1039,6 +1101,63 @@ mod tests { ); } + #[tokio::test] + async fn failed_workdir_create_retry_cannot_start_after_removal_claim() { + use crate::store::WorkdirCreateOperationRecord; + + let (store, workdir) = seeded_store().await; + let create = WorkdirCreateOperationRecord { + workspace_id: "workspace-a".to_string(), + operation_id: "create-a".to_string(), + request_fingerprint: "create-fingerprint".to_string(), + repository_id: "repository-a".to_string(), + selector: Some("develop".to_string()), + requested_runtime_id: Some("runtime-a".to_string()), + resolved_runtime_id: "runtime-a".to_string(), + config_revision: 1, + config_projection_digest: "projection-a".to_string(), + source_kind: Some("local_path".to_string()), + source_uri: Some("/repository-a".to_string()), + source_revision: Some(1), + source_fingerprint: Some("source-a".to_string()), + credential_id: None, + credential_revision: None, + host_trust_id: None, + host_trust_revision: None, + repository_access_mode: None, + cache_generation: 0, + working_directory_id: "workdir-a".to_string(), + state: "pending".to_string(), + failure: None, + created_at: "1".to_string(), + updated_at: "1".to_string(), + }; + store.reserve_workdir_create_operation(&create).unwrap(); + store + .finish_workdir_create_operation( + "workspace-a", + "create-a", + "create-fingerprint", + false, + Some("provider failed"), + "2", + ) + .unwrap(); + let intent = + workdir_removal_intent(&workdir, "workspace-api", "remove clean Workdir").unwrap(); + store.reserve_workdir_removal_operation(&intent).unwrap(); + + let error = store + .begin_failed_workdir_create_retry("workspace-a", "create-a", "create-fingerprint", "3") + .unwrap_err(); + assert!(matches!(error, Error::WorkdirAttachmentConflict(_))); + let create = store + .load_workdir_create_operation("workspace-a", "create-a") + .unwrap() + .unwrap(); + assert_eq!(create.state, "failed"); + } + #[tokio::test] async fn pending_removal_fences_new_attachment_and_retry_rereads_live_reservation() { let (store, workdir) = seeded_store().await; @@ -1062,6 +1181,7 @@ mod tests { &failed.workspace_id, &failed.operation_id, &failed.request_fingerprint, + attempt_owner(), ) .unwrap(); let guards = store.workdir_removal_guards(&retry).unwrap(); diff --git a/docs/design/durable-operations.md b/docs/design/durable-operations.md index 7098b0f3..3455aab2 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. +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. `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. @@ -30,7 +30,7 @@ Workdir removal is one durable side-effect operation in the Workspace Server DB. Each attempt: 1. resolves or revalidates the persisted same-Workspace Workdir, Runtime, Repository, and materialization identity; -2. checks current attachments, attachment reservations, current assignment occupancy, retention/cleanup holds, and pending materialization authority; +2. checks current attachments, attachment reservations, current assignment occupancy, retention/cleanup holds, and pending materialization authority; a failed Workdir-create retry must atomically return to `pending` before provider work and is rejected while removal is pending; 3. retains dirty, occupied, blocked, or otherwise unknown Workdirs without detaching a Worker or forcing deletion; 4. observes the owning Runtime/provider and calls its existing Workdir cleanup only for an eligible clean Workdir; 5. treats only successful provider cleanup or exact `working_directory_not_found` as removal evidence;