From 2d4c7b383afdff9e1ae3ea301f5b35474b322abb Mon Sep 17 00:00:00 2001 From: Hare Date: Sun, 13 Sep 2026 02:30:15 +0900 Subject: [PATCH] feat: add guarded runtime removal operation --- crates/workspace-api/src/lib.rs | 40 + crates/workspace-server/src/latest_schema.sql | 50 + crates/workspace-server/src/server.rs | 546 ++++++++-- crates/workspace-server/src/store.rs | 977 +++++++++++++++++- .../src/workspace_deletion.rs | 1 + 5 files changed, 1526 insertions(+), 88 deletions(-) diff --git a/crates/workspace-api/src/lib.rs b/crates/workspace-api/src/lib.rs index a053cca3..07eed62e 100644 --- a/crates/workspace-api/src/lib.rs +++ b/crates/workspace-api/src/lib.rs @@ -1810,6 +1810,43 @@ pub struct RevokeRuntimeTrustKeyRequest { pub expected_revision: u64, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] +pub struct RemoveRuntimeRequest { + pub operation_id: String, + #[cfg_attr(feature = "typescript", ts(type = "number"))] + pub expected_binding_revision: u64, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum RuntimeRemovalOperationState { + Pending, + CleanupPending, + Succeeded, + Failed, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(deny_unknown_fields)] +pub struct RuntimeRemovalOperationResponse { + pub operation_id: String, + pub workspace_id: String, + pub runtime_id: String, + pub state: RuntimeRemovalOperationState, + pub binding_removed: bool, + pub runtime_registration_removed: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure_category: Option, + pub created_at: String, + pub updated_at: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub completed_at: Option, +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[serde(rename_all = "snake_case")] @@ -3116,6 +3153,9 @@ pub fn catalog_typescript() -> String { WorkspaceRuntimeDetail::decl(&config), RuntimeTrustKeyRevealResponse::decl(&config), RevokeRuntimeTrustKeyRequest::decl(&config), + RemoveRuntimeRequest::decl(&config), + RuntimeRemovalOperationState::decl(&config), + RuntimeRemovalOperationResponse::decl(&config), RuntimeTrustConflictKind::decl(&config), RuntimeTrustConflictResponse::decl(&config), RuntimePublicIdentityBundle::decl(&config), diff --git a/crates/workspace-server/src/latest_schema.sql b/crates/workspace-server/src/latest_schema.sql index 688d5224..8e00b1ac 100644 --- a/crates/workspace-server/src/latest_schema.sql +++ b/crates/workspace-server/src/latest_schema.sql @@ -1057,6 +1057,56 @@ CREATE UNIQUE INDEX worker_workdir_links_active_worker_unique WHERE unlinked_at IS NULL; CREATE INDEX worker_workdir_links_workdir ON worker_workdir_links(workspace_id, workdir_id); +CREATE TABLE runtime_removal_operations ( + operation_id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + request_fingerprint TEXT NOT NULL, + expected_binding_revision INTEGER NOT NULL, + config_revision INTEGER NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'cleanup_pending', 'succeeded', 'failed')), + failure_category TEXT, + binding_removed INTEGER NOT NULL CHECK (binding_removed IN (0, 1)), + runtime_registration_removed INTEGER CHECK (runtime_registration_removed IS NULL OR runtime_registration_removed IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT, + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE +); + +CREATE UNIQUE INDEX runtime_removal_operations_one_active_runtime +ON runtime_removal_operations(runtime_id) +WHERE state IN ('pending', 'cleanup_pending'); + +CREATE INDEX runtime_removal_operations_workspace_state +ON runtime_removal_operations(workspace_id, state, updated_at); + +CREATE TRIGGER runtime_binding_insert_blocked_by_removal +BEFORE INSERT ON workspace_runtime_bindings +FOR EACH ROW +WHEN EXISTS ( + SELECT 1 + FROM runtime_removal_operations operation + WHERE operation.runtime_id = NEW.runtime_id + AND operation.state IN ('pending', 'cleanup_pending') +) +BEGIN + SELECT RAISE(ABORT, 'runtime_removal_in_progress'); +END; + +CREATE TRIGGER runtime_binding_update_blocked_by_removal +BEFORE UPDATE ON workspace_runtime_bindings +FOR EACH ROW +WHEN EXISTS ( + SELECT 1 + FROM runtime_removal_operations operation + WHERE operation.runtime_id = NEW.runtime_id + AND operation.state IN ('pending', 'cleanup_pending') +) +BEGIN + SELECT RAISE(ABORT, 'runtime_removal_in_progress'); +END; + CREATE TABLE workspace_deletion_operations ( operation_id TEXT PRIMARY KEY, request_fingerprint TEXT NOT NULL, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 2d02304a..e3b63f75 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -83,19 +83,20 @@ use workspace_api::{ PasskeyLoginCompleteRequest, PasskeyLoginOptionsRequest, PasskeyLoginOptionsResponse, PasskeyRegistrationCompleteRequest, PasskeyRegistrationOptionsRequest, PasskeyRegistrationOptionsResponse, ProfileSettingsResponse, PutRepositorySshHostTrustRequest, - RepositoryAccessProjection, RepositoryDetailResponse, RepositoryListResponse, - RepositoryLogResponse, RepositorySshConnectionProbeRequest, + RemoveRuntimeRequest, RepositoryAccessProjection, RepositoryDetailResponse, + RepositoryListResponse, RepositoryLogResponse, RepositorySshConnectionProbeRequest, RepositorySshConnectionProbeResponse, RepositorySshConnectionTrustState, RepositorySshCredential, RepositorySshHostKeyCandidate, RepositorySshHostTrust, RepositorySshPublicKey, RequestActor, RevokeRuntimeTrustKeyRequest, RotateRepositorySshCredentialRequest, RuntimeConnectionDisplayState, RuntimeConnectionTestFailureKind, RuntimeConnectionTestResponse, RuntimeConnectionTestStatus, - RuntimeManagementSummary, RuntimeTrustAuditAction, RuntimeTrustAuditEntry, - RuntimeTrustConflictKind, RuntimeTrustConflictResponse, RuntimeTrustKeyRevealResponse, - RuntimeTrustKeyState, RuntimeTrustKeyStatus, TICKET_ORCHESTRATION_PLANS_QUERY_PATH, - TICKET_RELATIONS_QUERY_PATH, UpdateRemoteRuntimeRequest, UpdateWorkspaceMetadataRequest, - WhoamiResponse, WorkerLaunchOptionsResponse, WorkerLaunchProfileCandidate, - WorkerLaunchRuntimeOption, WorkerLaunchWorkerSummary, + RuntimeManagementSummary, RuntimeRemovalOperationResponse, + RuntimeRemovalOperationState as ApiRuntimeRemovalOperationState, RuntimeTrustAuditAction, + RuntimeTrustAuditEntry, RuntimeTrustConflictKind, RuntimeTrustConflictResponse, + RuntimeTrustKeyRevealResponse, RuntimeTrustKeyState, RuntimeTrustKeyStatus, + TICKET_ORCHESTRATION_PLANS_QUERY_PATH, TICKET_RELATIONS_QUERY_PATH, UpdateRemoteRuntimeRequest, + UpdateWorkspaceMetadataRequest, WhoamiResponse, WorkerLaunchOptionsResponse, + WorkerLaunchProfileCandidate, WorkerLaunchRuntimeOption, WorkerLaunchWorkerSummary, WorkingDirectoryCreateRequest as BrowserWorkingDirectoryCreateRequest, WorkingDirectoryCreateResponse as BrowserWorkingDirectoryCreateResponse, WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse, @@ -171,7 +172,8 @@ use crate::skills; use crate::store::{ AccountRecord, ApiTokenRecord, AuthChallengeRecord, BrowserSessionRecord, ControlPlaneStore, DeviceLoginFlowRecord, FlowSourceRecord, PasskeyCredentialRecord, RepositoryInsertOutcome, - RepositoryRecord, TicketAssignmentPrincipal, TicketAssignmentRole, TicketCoderAssignmentRecord, + RepositoryRecord, RuntimeRemovalOperation, RuntimeRemovalOperationState, + TicketAssignmentPrincipal, TicketAssignmentRole, TicketCoderAssignmentRecord, TicketRoleAssignmentRecord, UserRecord, WorkdirCreateCredentialCandidate, WorkdirCreateCredentialCandidateRole, WorkdirCreateOperationRecord, WorkdirRegistryRecord, WorkerControlGrantRecord, WorkerRegistryRecord, WorkerWorkdirLinkRecord, WorkspaceRecord, @@ -2375,6 +2377,7 @@ impl WorkspaceApi { .map_err(|message| Error::Config(message.to_string()))?; } recover_workdir_removals(&api)?; + recover_runtime_removals(&api).await; Ok(api) } @@ -3422,7 +3425,7 @@ fn build_inner_router(api: WorkspaceApi) -> Router { "/api/w/{workspace_id}/runtimes/{runtime_id}", get(scoped_get_runtime_detail) .post(scoped_update_remote_runtime) - .delete(scoped_delete_remote_runtime), + .delete(scoped_remove_remote_runtime), ) .route( "/api/w/{workspace_id}/runtimes/{runtime_id}/trust-key", @@ -12257,14 +12260,15 @@ async fn runtime_trust_conflict_response( ) } -async fn scoped_delete_remote_runtime( +async fn scoped_remove_remote_runtime( State(api): State, AxumPath(path): AxumPath, Extension(actor): Extension, -) -> ApiResult { + Json(request): Json, +) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; require_workspace_owner(&api, &path.workspace_id, &actor, "Runtime removal").await?; - delete_remote_runtime(State(api), AxumPath(path.runtime_id)).await + remove_remote_runtime(State(api), AxumPath(path.runtime_id), Json(request)).await } async fn scoped_test_runtime_connection( @@ -13873,70 +13877,294 @@ async fn create_remote_runtime( Ok((status, Json(resource))) } -async fn delete_remote_runtime( - State(api): State, - AxumPath(runtime_id): AxumPath, -) -> ApiResult { - if runtime_id == EMBEDDED_WORKER_RUNTIME_ID { - return Err(settings_bad_request( - "embedded_runtime_not_config_managed", - "the embedded Runtime is built in and cannot be deleted", +fn runtime_removal_response(operation: RuntimeRemovalOperation) -> RuntimeRemovalOperationResponse { + RuntimeRemovalOperationResponse { + operation_id: operation.operation_id, + workspace_id: operation.workspace_id, + runtime_id: operation.runtime_id, + state: match operation.state { + RuntimeRemovalOperationState::Pending => ApiRuntimeRemovalOperationState::Pending, + RuntimeRemovalOperationState::CleanupPending => { + ApiRuntimeRemovalOperationState::CleanupPending + } + RuntimeRemovalOperationState::Succeeded => ApiRuntimeRemovalOperationState::Succeeded, + RuntimeRemovalOperationState::Failed => ApiRuntimeRemovalOperationState::Failed, + }, + binding_removed: operation.binding_removed, + runtime_registration_removed: operation.runtime_registration_removed, + failure_category: operation.failure_category, + created_at: operation.created_at, + updated_at: operation.updated_at, + completed_at: operation.completed_at, + } +} + +fn runtime_removal_fingerprint( + workspace_id: &str, + runtime_id: &str, + expected_binding_revision: u64, +) -> String { + let digest = Sha256::digest(format!( + "runtime-removal-v1\0{workspace_id}\0{runtime_id}\0{expected_binding_revision}" + )); + let digest = digest + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + format!("sha256:{digest}") +} + +fn runtime_removal_config_guard( + api: &WorkspaceApi, + operation: &RuntimeRemovalOperation, +) -> Result<()> { + let config_state = api + .config_store + .load_workspace_config(&operation.workspace_id)? + .ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workspace {} has no active configuration", + operation.workspace_id + )) + })?; + if config_state.snapshot.revision != operation.config_revision { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_config_revision_changed".to_string(), )); } - let binding = api - .store - .get_workspace_runtime_binding(&api.config.workspace_id, &runtime_id) - .await? - .ok_or_else(|| Error::UnknownRuntime(runtime_id.clone()))?; - if binding.revoked_at.is_none() { + let projection = crate::runtime_settings::project_runtime_from_workspace_config( + &operation.workspace_id, + &config_state, + )?; + if projection.default_runtime_id.as_deref() == Some(operation.runtime_id.as_str()) { return Err(Error::RuntimeBindingConflict( - "runtime trust is still active; revoke this Workspace's trust key with an expected revision before removing the inactive registration".to_string(), - ) - .into()); + "runtime_removal_config_reference_blocked".to_string(), + )); } - let has_other_active_binding = api - .store - .has_other_active_workspace_runtime_binding(&api.config.workspace_id, &runtime_id) - .await?; - if !has_other_active_binding { + Ok(()) +} + +async fn execute_runtime_removal( + api: &WorkspaceApi, + operation: RuntimeRemovalOperation, +) -> ApiResult { + if operation.state == RuntimeRemovalOperationState::Succeeded { + return Ok(operation); + } + let operation = if operation.state == RuntimeRemovalOperationState::Pending { + if let Err(error) = runtime_removal_config_guard(api, &operation) { + let _ = api + .store + .mark_runtime_removal_failed( + &operation.operation_id, + "runtime_removal_config_guard_failed", + ) + .await; + return Err(error.into()); + } + let binding = api + .store + .get_workspace_runtime_binding(&operation.workspace_id, &operation.runtime_id) + .await? + .ok_or_else(|| Error::UnknownRuntime(operation.runtime_id.clone()))?; match api .runtime - .unregister_if_idle(&runtime_id, api.config.max_records.min(200)) - .map_err(|err| err.into_error())? + .unregister_if_idle(&operation.runtime_id, api.config.max_records.min(200)) + .map_err(|error| error.into_error())? { RuntimeRegistryUnregisterResult::Removed | RuntimeRegistryUnregisterResult::NotFound => {} RuntimeRegistryUnregisterResult::BlockedByWorkers { worker_count, - diagnostics, + mut diagnostics, } => { - let mut diagnostics = diagnostics; + let _ = api + .store + .mark_runtime_removal_failed( + &operation.operation_id, + "runtime_removal_active_worker_blocked", + ) + .await; diagnostics.push(settings_diagnostic( - "remote_runtime_delete_blocked", + "runtime_removal_active_worker_blocked", DiagnosticSeverity::Error, format!( - "Remote Runtime '{runtime_id}' has {worker_count} active worker(s); stop or move them before deleting its final Workspace registration." + "Remote Runtime '{}' has {worker_count} active Worker(s); stop and remove them before removing the Runtime.", + operation.runtime_id ), )); return Err(ApiError::with_diagnostics( - Error::RuntimeOperationFailed { - runtime_id, - code: "remote_runtime_delete_blocked".to_string(), - message: "Remote Runtime has active workers".to_string(), - }, + Error::RuntimeBindingConflict( + "runtime_removal_active_worker_blocked".to_string(), + ), diagnostics, )); } } + match api + .store + .commit_runtime_binding_removal(&operation.operation_id) + .await + { + Ok(operation) => operation, + Err(error) => { + let _ = api + .store + .mark_runtime_removal_failed( + &operation.operation_id, + "runtime_removal_commit_failed", + ) + .await; + let restore_result = api.register_workspace_runtime_binding(binding, true); + if let Err(restore_error) = restore_result { + tracing::warn!( + workspace_id = %operation.workspace_id, + runtime_id = %operation.runtime_id, + operation_id = %operation.operation_id, + error = %restore_error, + "failed to restore Runtime registration after removal commit failure" + ); + } + return Err(error.into()); + } + } + } else { + operation + }; + + if operation.state != RuntimeRemovalOperationState::CleanupPending { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_operation_not_cleanup_pending".to_string(), + ) + .into()); } - if !api + api.runtime_binding_expectations + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&(operation.workspace_id.clone(), operation.runtime_id.clone())); + api.runtime_subscription_broker + .unregister_runtime(&operation.runtime_id); + api.store + .complete_runtime_removal(&operation.operation_id, true) + .await + .map_err(Into::into) +} + +async fn recover_runtime_removals(api: &WorkspaceApi) { + let operations = match api .store - .delete_workspace_runtime_binding(&api.config.workspace_id, &runtime_id) - .await? + .list_resumable_runtime_removals(&api.config.workspace_id) + .await { - return Err(Error::UnknownRuntime(runtime_id).into()); + Ok(operations) => operations, + Err(error) => { + tracing::warn!( + workspace_id = %api.config.workspace_id, + error = %error, + "failed to list resumable Runtime removal operations" + ); + return; + } + }; + for operation in operations { + let operation_id = operation.operation_id.clone(); + let runtime_id = operation.runtime_id.clone(); + if let Err(error) = execute_runtime_removal(api, operation).await { + tracing::warn!( + workspace_id = %api.config.workspace_id, + runtime_id = %runtime_id, + operation_id = %operation_id, + error = ?error, + "Runtime removal recovery remains incomplete" + ); + } } - Ok(StatusCode::NO_CONTENT) +} + +async fn remove_remote_runtime( + State(api): State, + AxumPath(runtime_id): AxumPath, + Json(request): Json, +) -> ApiResult> { + if runtime_id == EMBEDDED_WORKER_RUNTIME_ID { + return Err(settings_bad_request( + "embedded_runtime_not_config_managed", + "the embedded Runtime is built in and cannot be removed", + )); + } + if request.operation_id.is_empty() || request.operation_id.len() > 128 { + return Err(Error::InvalidInput( + "operation_id must contain between 1 and 128 bytes".to_string(), + ) + .into()); + } + let request_fingerprint = runtime_removal_fingerprint( + &api.config.workspace_id, + &runtime_id, + request.expected_binding_revision, + ); + if let Some(existing) = api.store.get_runtime_removal(&request.operation_id).await? { + if existing.workspace_id != api.config.workspace_id + || existing.runtime_id != runtime_id + || existing.request_fingerprint != request_fingerprint + || existing.expected_binding_revision != request.expected_binding_revision + { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_operation_id_reused".to_string(), + ) + .into()); + } + let existing = if existing.state == RuntimeRemovalOperationState::Failed { + api.store + .reserve_runtime_removal( + &existing.workspace_id, + &existing.runtime_id, + &existing.operation_id, + &existing.request_fingerprint, + existing.expected_binding_revision, + existing.config_revision, + ) + .await? + .operation + } else { + existing + }; + let operation = execute_runtime_removal(&api, existing).await?; + return Ok(Json(runtime_removal_response(operation))); + } + + let config_state = api + .config_store + .load_workspace_config(&api.config.workspace_id)? + .ok_or_else(|| { + Error::RegistryInconsistency(format!( + "Workspace {} has no active configuration", + api.config.workspace_id + )) + })?; + let runtime_projection = crate::runtime_settings::project_runtime_from_workspace_config( + &api.config.workspace_id, + &config_state, + )?; + if runtime_projection.default_runtime_id.as_deref() == Some(runtime_id.as_str()) { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_config_reference_blocked".to_string(), + ) + .into()); + } + let reservation = api + .store + .reserve_runtime_removal( + &api.config.workspace_id, + &runtime_id, + &request.operation_id, + &request_fingerprint, + request.expected_binding_revision, + config_state.snapshot.revision, + ) + .await?; + let operation = execute_runtime_removal(&api, reservation.operation).await?; + Ok(Json(runtime_removal_response(operation))) } async fn perform_workspace_runtime_verification( @@ -25778,6 +26006,38 @@ mod tests { .unwrap(); } + async fn register_test_runtime(api: &WorkspaceApi, runtime_id: &str) { + let identity = RuntimeIdentityMaterial::generate(runtime_id).unwrap(); + let binding = WorkspaceRuntimeBinding { + workspace_id: TEST_WORKSPACE_ID.to_string(), + runtime_id: runtime_id.to_string(), + display_name: format!("{runtime_id} display"), + base_url: "https://runtime.example.invalid".to_string(), + public_key: identity.public_key, + public_key_fingerprint: String::new(), + binding_revision: 1, + state: StoredRuntimeBindingState::Verified, + authentication_mode: StoredRuntimeAuthenticationMode::LegacyServerIssuer, + workspace_key_id: None, + workspace_key_generation: None, + created_at: "1".to_string(), + updated_at: "1".to_string(), + revoked_at: None, + }; + api.store + .upsert_workspace_runtime_binding_record(binding.clone(), false) + .await + .unwrap(); + api.runtime.register_or_replace( + RemoteWorkerRuntime::new( + remote_runtime_config_from_binding(&binding).unwrap(), + TEST_WORKSPACE_ID.to_string(), + "http://127.0.0.1:8787".to_string(), + ) + .unwrap(), + ); + } + fn assign_test_orchestrator(api: &WorkspaceApi, ticket_id: &str) { api.store .set_current_ticket_role_assignment( @@ -28920,7 +29180,10 @@ mod tests { app.clone(), "DELETE", &format!("{runtimes_uri}/{EMBEDDED_WORKER_RUNTIME_ID}"), - None, + Some(serde_json::json!({ + "operation_id": "remove-embedded", + "expected_binding_revision": 1 + })), StatusCode::BAD_REQUEST, ) .await; @@ -28988,25 +29251,30 @@ mod tests { .expect("team runtime launch option"); assert_eq!(team_runtime["working_directory_required"], true); - api.store - .revoke_workspace_runtime_binding_key( - TEST_WORKSPACE_ID, - "team-runtime", - 1, - &format!("account-{TEST_WORKSPACE_ID}"), - &Utc::now().to_rfc3339(), - ) - .await - .unwrap(); + let removal_request = serde_json::json!({ + "operation_id": "remove-team-runtime", + "expected_binding_revision": 1 + }); let deleted = request_json( app.clone(), "DELETE", &format!("{runtimes_uri}/team-runtime"), - None, - StatusCode::NO_CONTENT, + Some(removal_request.clone()), + StatusCode::OK, ) .await; - assert_eq!(deleted["message"], ""); + assert_eq!(deleted["state"], "succeeded"); + assert_eq!(deleted["binding_removed"], true); + assert_eq!(deleted["runtime_registration_removed"], true); + let replay = request_json( + app.clone(), + "DELETE", + &format!("{runtimes_uri}/team-runtime"), + Some(removal_request), + StatusCode::OK, + ) + .await; + assert_eq!(replay, deleted); let launch_options = get_json(app.clone(), "/api/workers/launch-options").await; assert!( !launch_options["runtimes"] @@ -29022,6 +29290,120 @@ mod tests { assert!(persisted.is_none()); } + #[tokio::test] + async fn runtime_removal_config_and_revision_guards_preserve_active_trust() { + let root = tempfile::tempdir().unwrap(); + let api = test_api(root.path()).await; + register_test_runtime(&api, "guarded-runtime").await; + let app = build_inner_router(api.clone()).layer(Extension(test_owner_actor())); + let runtimes_uri = format!("/api/w/{TEST_WORKSPACE_ID}/runtimes"); + + let stale = request_json( + app.clone(), + "DELETE", + &format!("{runtimes_uri}/guarded-runtime"), + Some(serde_json::json!({ + "operation_id": "remove-guarded-stale", + "expected_binding_revision": 2 + })), + StatusCode::CONFLICT, + ) + .await; + assert!( + stale.to_string().contains("revision conflict"), + "unexpected stale-revision response: {stale}" + ); + let binding = api + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, "guarded-runtime") + .await + .unwrap() + .expect("stale removal must preserve binding"); + assert_eq!(binding.binding_revision, 1); + assert!(binding.revoked_at.is_none()); + + set_test_default_runtime(&api, "guarded-runtime"); + let referenced = request_json( + app, + "DELETE", + &format!("{runtimes_uri}/guarded-runtime"), + Some(serde_json::json!({ + "operation_id": "remove-guarded-referenced", + "expected_binding_revision": 1 + })), + StatusCode::CONFLICT, + ) + .await; + assert!( + referenced + .to_string() + .contains("runtime_removal_config_reference_blocked"), + "unexpected config-reference response: {referenced}" + ); + let binding = api + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, "guarded-runtime") + .await + .unwrap() + .expect("config-blocked removal must preserve binding"); + assert_eq!(binding.binding_revision, 1); + assert!(binding.revoked_at.is_none()); + } + + #[tokio::test] + async fn startup_recovers_runtime_removal_after_binding_checkpoint() { + let root = tempfile::tempdir().unwrap(); + let api = test_api(root.path()).await; + register_test_runtime(&api, "recover-runtime").await; + let config_revision = api + .config_store + .load_workspace_config(TEST_WORKSPACE_ID) + .unwrap() + .unwrap() + .snapshot + .revision; + api.store + .reserve_runtime_removal( + TEST_WORKSPACE_ID, + "recover-runtime", + "remove-recover-runtime", + &runtime_removal_fingerprint(TEST_WORKSPACE_ID, "recover-runtime", 1), + 1, + config_revision, + ) + .await + .unwrap(); + let checkpoint = api + .store + .commit_runtime_binding_removal("remove-recover-runtime") + .await + .unwrap(); + assert_eq!( + checkpoint.state, + RuntimeRemovalOperationState::CleanupPending + ); + drop(api); + + let recovered = test_api(root.path()).await; + let operation = recovered + .store + .get_runtime_removal("remove-recover-runtime") + .await + .unwrap() + .expect("recovered removal operation must remain auditable"); + assert_eq!(operation.state, RuntimeRemovalOperationState::Succeeded); + assert!(operation.binding_removed); + assert_eq!(operation.runtime_registration_removed, Some(true)); + assert!( + recovered + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, "recover-runtime") + .await + .unwrap() + .is_none() + ); + } + #[tokio::test(flavor = "multi_thread")] async fn runtime_connection_delete_rejects_active_remote_workers() { let (runtime, _worker_ref) = runtime_with_worker(); @@ -29069,17 +29451,7 @@ mod tests { ) .unwrap(), ); - api.store - .revoke_workspace_runtime_binding_key( - TEST_WORKSPACE_ID, - "busy-runtime", - 1, - &format!("account-{TEST_WORKSPACE_ID}"), - &Utc::now().to_rfc3339(), - ) - .await - .unwrap(); - let app = build_inner_router(api).layer(Extension(test_owner_actor())); + let app = build_inner_router(api.clone()).layer(Extension(test_owner_actor())); let workers = get_json(app.clone(), "/api/workers").await; assert!( workers["items"] @@ -29094,7 +29466,10 @@ mod tests { app, "DELETE", &format!("/api/w/{TEST_WORKSPACE_ID}/runtimes/busy-runtime"), - None, + Some(serde_json::json!({ + "operation_id": "remove-busy-runtime", + "expected_binding_revision": 1 + })), StatusCode::CONFLICT, ) .await; @@ -29102,15 +29477,16 @@ mod tests { response["message"] .as_str() .unwrap() - .contains("remote_runtime_delete_blocked") - ); - assert!( - response["diagnostics"] - .as_array() - .unwrap() - .iter() - .any(|diagnostic| { diagnostic["code"] == "remote_runtime_delete_blocked" }) + .contains("runtime_removal_active_worker_blocked") ); + let persisted = api + .store + .get_workspace_runtime_binding(TEST_WORKSPACE_ID, "busy-runtime") + .await + .unwrap() + .expect("busy Runtime binding must remain after rejected removal"); + assert_eq!(persisted.binding_revision, 1); + assert!(persisted.revoked_at.is_none()); } async fn run_runtime_connection_test( diff --git a/crates/workspace-server/src/store.rs b/crates/workspace-server/src/store.rs index daba7aea..800e7d50 100644 --- a/crates/workspace-server/src/store.rs +++ b/crates/workspace-server/src/store.rs @@ -18,7 +18,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore; use crate::{Error, Result}; const OLDEST_SCHEMA_VERSION: i64 = 50; -const LATEST_SCHEMA_VERSION: i64 = 59; +const LATEST_SCHEMA_VERSION: i64 = 60; const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline"; const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings"; const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit"; @@ -34,6 +34,7 @@ const REMOVE_WORKDIR_CACHE_GENERATION_MIGRATION_NAME: &str = "remove obsolete Workdir Repository cache generation"; const WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME: &str = "Workdir create credential candidate snapshots"; +const RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME: &str = "guarded Runtime removal operations"; const MIGRATIONS: &[Migration] = &[ Migration { @@ -81,6 +82,11 @@ const MIGRATIONS: &[Migration] = &[ name: WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME, apply: migrate_workdir_credential_candidate_snapshots_v58_to_v59, }, + Migration { + version: 60, + name: RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME, + apply: migrate_runtime_removal_operations_v59_to_v60, + }, ]; #[derive(Clone, Copy)] @@ -271,6 +277,51 @@ pub struct WorkspaceRuntimeBinding { pub revoked_at: Option, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RuntimeRemovalOperation { + pub operation_id: String, + pub workspace_id: String, + pub runtime_id: String, + pub request_fingerprint: String, + pub expected_binding_revision: u64, + pub config_revision: u64, + pub state: RuntimeRemovalOperationState, + pub failure_category: Option, + pub binding_removed: bool, + pub runtime_registration_removed: Option, + pub created_at: String, + pub updated_at: String, + pub completed_at: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RuntimeRemovalOperationState { + Pending, + CleanupPending, + Succeeded, + Failed, +} + +impl RuntimeRemovalOperationState { + fn parse(value: &str) -> Result { + match value { + "pending" => Ok(Self::Pending), + "cleanup_pending" => Ok(Self::CleanupPending), + "succeeded" => Ok(Self::Succeeded), + "failed" => Ok(Self::Failed), + other => Err(Error::Store(format!( + "unknown Runtime removal operation state `{other}`" + ))), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RuntimeRemovalReservation { + pub operation: RuntimeRemovalOperation, + pub replay: bool, +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum WorkspaceRuntimeBindingState { @@ -857,6 +908,37 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore { workspace_id: &str, runtime_id: &str, ) -> Result; + async fn reserve_runtime_removal( + &self, + workspace_id: &str, + runtime_id: &str, + operation_id: &str, + request_fingerprint: &str, + expected_binding_revision: u64, + config_revision: u64, + ) -> Result; + async fn get_runtime_removal( + &self, + operation_id: &str, + ) -> Result>; + async fn mark_runtime_removal_failed( + &self, + operation_id: &str, + failure_category: &str, + ) -> Result; + async fn commit_runtime_binding_removal( + &self, + operation_id: &str, + ) -> Result; + async fn complete_runtime_removal( + &self, + operation_id: &str, + runtime_registration_removed: bool, + ) -> Result; + async fn list_resumable_runtime_removals( + &self, + workspace_id: &str, + ) -> Result>; async fn upsert_workspace_runtime_binding_record( &self, record: WorkspaceRuntimeBinding, @@ -2008,6 +2090,255 @@ impl SqliteWorkspaceStore { }) } + pub fn reserve_runtime_removal( + &self, + workspace_id: &str, + runtime_id: &str, + operation_id: &str, + request_fingerprint: &str, + expected_binding_revision: u64, + config_revision: u64, + ) -> Result { + validate_identifier("workspace_id", workspace_id)?; + validate_identifier("runtime_id", runtime_id)?; + validate_identifier("operation_id", operation_id)?; + validate_non_empty("request_fingerprint", request_fingerprint)?; + if operation_id.len() > 128 || request_fingerprint.len() > 128 { + return Err(Error::InvalidInput( + "Runtime removal operation identity is too long".to_string(), + )); + } + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + if let Some(existing) = load_runtime_removal_operation(&tx, operation_id)? { + if existing.workspace_id != workspace_id + || existing.runtime_id != runtime_id + || existing.request_fingerprint != request_fingerprint + || existing.expected_binding_revision != expected_binding_revision + || existing.config_revision != config_revision + { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_operation_id_reused".to_string(), + )); + } + if existing.state != RuntimeRemovalOperationState::Failed { + tx.commit()?; + return Ok(RuntimeRemovalReservation { + operation: existing, + replay: true, + }); + } + runtime_removal_preflight( + &tx, + workspace_id, + runtime_id, + expected_binding_revision, + )?; + let now = chrono::Utc::now().to_rfc3339(); + tx.execute( + "UPDATE runtime_removal_operations \ + SET state = 'pending', failure_category = NULL, updated_at = ?2, \ + completed_at = NULL \ + WHERE operation_id = ?1 AND state = 'failed'", + params![operation_id, now], + )?; + let operation = load_runtime_removal_operation(&tx, operation_id)?.ok_or_else(|| { + Error::Store("Runtime removal operation disappeared".to_string()) + })?; + tx.commit()?; + return Ok(RuntimeRemovalReservation { + operation, + replay: true, + }); + } + + let active_operation_id = tx + .query_row( + "SELECT operation_id FROM runtime_removal_operations \ + WHERE runtime_id = ?1 AND state IN ('pending', 'cleanup_pending')", + params![runtime_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if active_operation_id.is_some() { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_already_in_progress".to_string(), + )); + } + runtime_removal_preflight( + &tx, + workspace_id, + runtime_id, + expected_binding_revision, + )?; + let now = chrono::Utc::now().to_rfc3339(); + tx.execute( + "INSERT INTO runtime_removal_operations(\ + operation_id, workspace_id, runtime_id, request_fingerprint, \ + expected_binding_revision, config_revision, state, failure_category, \ + binding_removed, runtime_registration_removed, created_at, updated_at, completed_at\ + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'pending', NULL, 0, NULL, ?7, ?7, NULL)", + params![ + operation_id, + workspace_id, + runtime_id, + request_fingerprint, + i64::try_from(expected_binding_revision).map_err(|_| { + Error::InvalidInput("binding revision is too large".to_string()) + })?, + i64::try_from(config_revision).map_err(|_| { + Error::InvalidInput("config revision is too large".to_string()) + })?, + now, + ], + )?; + let operation = load_runtime_removal_operation(&tx, operation_id)?.ok_or_else(|| { + Error::Store("Runtime removal operation was not persisted".to_string()) + })?; + tx.commit()?; + Ok(RuntimeRemovalReservation { + operation, + replay: false, + }) + }) + } + + pub fn get_runtime_removal( + &self, + operation_id: &str, + ) -> Result> { + validate_identifier("operation_id", operation_id)?; + self.with_conn(|conn| load_runtime_removal_operation(conn, operation_id)) + } + + pub fn mark_runtime_removal_failed( + &self, + operation_id: &str, + failure_category: &str, + ) -> Result { + validate_identifier("operation_id", operation_id)?; + validate_non_empty("failure_category", failure_category)?; + let failure_category = failure_category.chars().take(256).collect::(); + self.with_conn(|conn| { + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "UPDATE runtime_removal_operations \ + SET state = 'failed', failure_category = ?2, updated_at = ?3, completed_at = ?3 \ + WHERE operation_id = ?1 AND state = 'pending'", + params![operation_id, failure_category, now], + )?; + load_runtime_removal_operation(conn, operation_id)? + .ok_or_else(|| Error::Store("Runtime removal operation not found".to_string())) + }) + } + + pub fn commit_runtime_binding_removal( + &self, + operation_id: &str, + ) -> Result { + validate_identifier("operation_id", operation_id)?; + self.with_conn_mut(|conn| { + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?; + let operation = load_runtime_removal_operation(&tx, operation_id)? + .ok_or_else(|| Error::Store("Runtime removal operation not found".to_string()))?; + if matches!( + operation.state, + RuntimeRemovalOperationState::CleanupPending + | RuntimeRemovalOperationState::Succeeded + ) { + tx.commit()?; + return Ok(operation); + } + if operation.state != RuntimeRemovalOperationState::Pending { + return Err(Error::RuntimeBindingConflict( + "runtime_removal_operation_not_pending".to_string(), + )); + } + runtime_removal_preflight( + &tx, + &operation.workspace_id, + &operation.runtime_id, + operation.expected_binding_revision, + )?; + tx.execute( + "DELETE FROM worker_mutation_source_proof_jtis \ + WHERE workspace_id = ?1 AND runtime_id = ?2", + params![operation.workspace_id, operation.runtime_id], + )?; + tx.execute( + "DELETE FROM workspace_runtime_binding_audit \ + WHERE workspace_id = ?1 AND runtime_id = ?2", + params![operation.workspace_id, operation.runtime_id], + )?; + let removed = tx.execute( + "DELETE FROM workspace_runtime_bindings \ + WHERE workspace_id = ?1 AND runtime_id = ?2 AND binding_revision = ?3", + params![ + operation.workspace_id, + operation.runtime_id, + i64::try_from(operation.expected_binding_revision).map_err(|_| { + Error::InvalidInput("binding revision is too large".to_string()) + })?, + ], + )?; + if removed != 1 { + return Err(Error::RuntimeBindingRevisionConflict { + expected: Some(operation.expected_binding_revision), + actual: None, + }); + } + let now = chrono::Utc::now().to_rfc3339(); + tx.execute( + "UPDATE runtime_removal_operations \ + SET state = 'cleanup_pending', failure_category = NULL, binding_removed = 1, \ + updated_at = ?2, completed_at = NULL \ + WHERE operation_id = ?1 AND state = 'pending'", + params![operation_id, now], + )?; + let operation = load_runtime_removal_operation(&tx, operation_id)? + .ok_or_else(|| Error::Store("Runtime removal operation disappeared".to_string()))?; + tx.commit()?; + Ok(operation) + }) + } + + pub fn complete_runtime_removal( + &self, + operation_id: &str, + runtime_registration_removed: bool, + ) -> Result { + validate_identifier("operation_id", operation_id)?; + self.with_conn(|conn| { + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "UPDATE runtime_removal_operations \ + SET state = 'succeeded', runtime_registration_removed = ?2, \ + failure_category = NULL, updated_at = ?3, completed_at = ?3 \ + WHERE operation_id = ?1 AND state = 'cleanup_pending'", + params![operation_id, i64::from(runtime_registration_removed), now], + )?; + load_runtime_removal_operation(conn, operation_id)? + .ok_or_else(|| Error::Store("Runtime removal operation not found".to_string())) + }) + } + + pub fn list_resumable_runtime_removals( + &self, + workspace_id: &str, + ) -> Result> { + validate_identifier("workspace_id", workspace_id)?; + self.with_conn(|conn| { + let mut stmt = conn.prepare(&format!( + "{RUNTIME_REMOVAL_OPERATION_SELECT} \ + WHERE workspace_id = ?1 AND state IN ('pending', 'cleanup_pending') \ + ORDER BY created_at ASC, operation_id ASC" + ))?; + stmt.query_map(params![workspace_id], read_runtime_removal_operation)? + .collect::, _>>() + .map_err(Error::from) + }) + } + pub fn upsert_workspace_runtime_binding( &self, mut record: WorkspaceRuntimeBinding, @@ -3572,6 +3903,67 @@ impl ControlPlaneStore for SqliteWorkspaceStore { SqliteWorkspaceStore::delete_workspace_runtime_binding(self, workspace_id, runtime_id) } + async fn reserve_runtime_removal( + &self, + workspace_id: &str, + runtime_id: &str, + operation_id: &str, + request_fingerprint: &str, + expected_binding_revision: u64, + config_revision: u64, + ) -> Result { + SqliteWorkspaceStore::reserve_runtime_removal( + self, + workspace_id, + runtime_id, + operation_id, + request_fingerprint, + expected_binding_revision, + config_revision, + ) + } + + async fn get_runtime_removal( + &self, + operation_id: &str, + ) -> Result> { + SqliteWorkspaceStore::get_runtime_removal(self, operation_id) + } + + async fn mark_runtime_removal_failed( + &self, + operation_id: &str, + failure_category: &str, + ) -> Result { + SqliteWorkspaceStore::mark_runtime_removal_failed(self, operation_id, failure_category) + } + + async fn commit_runtime_binding_removal( + &self, + operation_id: &str, + ) -> Result { + SqliteWorkspaceStore::commit_runtime_binding_removal(self, operation_id) + } + + async fn complete_runtime_removal( + &self, + operation_id: &str, + runtime_registration_removed: bool, + ) -> Result { + SqliteWorkspaceStore::complete_runtime_removal( + self, + operation_id, + runtime_registration_removed, + ) + } + + async fn list_resumable_runtime_removals( + &self, + workspace_id: &str, + ) -> Result> { + SqliteWorkspaceStore::list_resumable_runtime_removals(self, workspace_id) + } + async fn upsert_workspace_runtime_binding_record( &self, record: WorkspaceRuntimeBinding, @@ -7136,6 +7528,155 @@ fn validate_workspace_runtime_verification( Ok(()) } +fn read_runtime_removal_operation( + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + let state_value = row.get::<_, String>(6)?; + let state = RuntimeRemovalOperationState::parse(&state_value).map_err(|error| { + rusqlite::Error::FromSqlConversionFailure( + 6, + rusqlite::types::Type::Text, + error.to_string().into(), + ) + })?; + let expected_binding_revision = u64::try_from(row.get::<_, i64>(4)?).map_err(|_| { + rusqlite::Error::FromSqlConversionFailure( + 4, + rusqlite::types::Type::Integer, + "invalid Runtime removal binding revision".into(), + ) + })?; + let config_revision = u64::try_from(row.get::<_, i64>(5)?).map_err(|_| { + rusqlite::Error::FromSqlConversionFailure( + 5, + rusqlite::types::Type::Integer, + "invalid Runtime removal config revision".into(), + ) + })?; + Ok(RuntimeRemovalOperation { + operation_id: row.get(0)?, + workspace_id: row.get(1)?, + runtime_id: row.get(2)?, + request_fingerprint: row.get(3)?, + expected_binding_revision, + config_revision, + state, + failure_category: row.get(7)?, + binding_removed: row.get::<_, i64>(8)? != 0, + runtime_registration_removed: row.get::<_, Option>(9)?.map(|value| value != 0), + created_at: row.get(10)?, + updated_at: row.get(11)?, + completed_at: row.get(12)?, + }) +} + +const RUNTIME_REMOVAL_OPERATION_SELECT: &str = "SELECT operation_id, workspace_id, runtime_id, request_fingerprint, \ + expected_binding_revision, config_revision, state, failure_category, \ + binding_removed, runtime_registration_removed, created_at, updated_at, completed_at \ + FROM runtime_removal_operations"; + +fn load_runtime_removal_operation( + conn: &Connection, + operation_id: &str, +) -> Result> { + conn.query_row( + &format!("{RUNTIME_REMOVAL_OPERATION_SELECT} WHERE operation_id = ?1"), + params![operation_id], + read_runtime_removal_operation, + ) + .optional() + .map_err(Error::from) +} + +fn runtime_removal_preflight( + conn: &Connection, + workspace_id: &str, + runtime_id: &str, + expected_binding_revision: u64, +) -> Result<()> { + let binding_revision = conn + .query_row( + "SELECT binding_revision FROM workspace_runtime_bindings \ + WHERE workspace_id = ?1 AND runtime_id = ?2", + params![workspace_id, runtime_id], + |row| row.get::<_, i64>(0), + ) + .optional()?; + let Some(binding_revision) = binding_revision else { + return Err(Error::RuntimeBindingNotFound { + runtime_id: runtime_id.to_string(), + }); + }; + let binding_revision = u64::try_from(binding_revision) + .map_err(|_| Error::Store("Runtime binding revision is invalid".to_string()))?; + if binding_revision != expected_binding_revision { + return Err(Error::RuntimeBindingRevisionConflict { + expected: Some(expected_binding_revision), + actual: Some(binding_revision), + }); + } + + let guards = [ + ( + "other_workspace_binding", + "SELECT EXISTS(SELECT 1 FROM workspace_runtime_bindings \ + WHERE runtime_id = ?1 AND workspace_id <> ?2 AND state <> 'revoked')", + ), + ( + "active_worker", + "SELECT EXISTS(SELECT 1 FROM worker_registry WHERE runtime_id = ?1)", + ), + ( + "active_worker_assignment", + "SELECT EXISTS(SELECT 1 FROM ticket_current_worker_assignments WHERE runtime_id = ?1)", + ), + ( + "active_workdir", + "SELECT EXISTS(SELECT 1 FROM workdir_registry WHERE runtime_id = ?1)", + ), + ( + "active_workdir_attachment", + "SELECT EXISTS(SELECT 1 FROM worker_workdir_links \ + WHERE runtime_id = ?1 AND unlinked_at IS NULL)", + ), + ( + "worker_create_or_restore", + "SELECT EXISTS(SELECT 1 FROM worker_create_reservations \ + WHERE runtime_id = ?1 AND state <> 'removed')", + ), + ( + "workdir_create", + "SELECT EXISTS(SELECT 1 FROM workdir_create_operations \ + WHERE resolved_runtime_id = ?1 AND state = 'pending')", + ), + ( + "worker_removal", + "SELECT EXISTS(SELECT 1 FROM worker_removal_operations \ + WHERE runtime_id = ?1 AND state IN ('planned', 'executing', 'failed'))", + ), + ( + "workdir_removal", + "SELECT EXISTS(SELECT 1 FROM workdir_removal_operations \ + WHERE runtime_id = ?1 AND state IN ('pending', 'failed') AND retryable = 1)", + ), + ]; + for (category, sql) in guards { + let blocked = if category == "other_workspace_binding" { + conn.query_row(sql, params![runtime_id, workspace_id], |row| { + row.get::<_, bool>(0) + })? + } else { + conn.query_row(sql, params![runtime_id], |row| row.get::<_, bool>(0))? + }; + if blocked { + return Err(Error::RuntimeBindingConflict(format!( + "runtime_removal_{category}_blocked" + ))); + } + } + Ok(()) +} + fn read_workspace_runtime_binding( row: &rusqlite::Row<'_>, ) -> rusqlite::Result { @@ -9045,6 +9586,62 @@ fn verify_canonical_workspace_runtime_binding(binding: WorkspaceRuntimeBinding) Ok(()) } +fn migrate_runtime_removal_operations_v59_to_v60(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#"CREATE TABLE runtime_removal_operations ( + operation_id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + request_fingerprint TEXT NOT NULL, + expected_binding_revision INTEGER NOT NULL, + config_revision INTEGER NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending', 'cleanup_pending', 'succeeded', 'failed')), + failure_category TEXT, + binding_removed INTEGER NOT NULL CHECK (binding_removed IN (0, 1)), + runtime_registration_removed INTEGER CHECK (runtime_registration_removed IS NULL OR runtime_registration_removed IN (0, 1)), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT, + FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE + ); + CREATE UNIQUE INDEX runtime_removal_operations_one_active_runtime + ON runtime_removal_operations(runtime_id) + WHERE state IN ('pending', 'cleanup_pending'); + CREATE INDEX runtime_removal_operations_workspace_state + ON runtime_removal_operations(workspace_id, state, updated_at); + CREATE TRIGGER runtime_binding_insert_blocked_by_removal + BEFORE INSERT ON workspace_runtime_bindings + FOR EACH ROW + WHEN EXISTS ( + SELECT 1 FROM runtime_removal_operations operation + WHERE operation.runtime_id = NEW.runtime_id + AND operation.state IN ('pending', 'cleanup_pending') + ) + BEGIN + SELECT RAISE(ABORT, 'runtime_removal_in_progress'); + END; + CREATE TRIGGER runtime_binding_update_blocked_by_removal + BEFORE UPDATE ON workspace_runtime_bindings + FOR EACH ROW + WHEN EXISTS ( + SELECT 1 FROM runtime_removal_operations operation + WHERE operation.runtime_id = NEW.runtime_id + AND operation.state IN ('pending', 'cleanup_pending') + ) + BEGIN + SELECT RAISE(ABORT, 'runtime_removal_in_progress'); + END;"#, + )?; + conn.execute( + "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)", + params![ + LATEST_SCHEMA_VERSION, + RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME + ], + )?; + Ok(()) +} + fn create_latest_workspace_schema(conn: &Connection) -> Result<()> { conn.execute_batch(include_str!("latest_schema.sql"))?; Ok(()) @@ -9679,6 +10276,369 @@ fn migrate_workdir_credential_candidate_snapshots_v58_to_v59(conn: &Connection) mod tests { use super::*; + fn runtime_removal_test_store() -> (tempfile::TempDir, SqliteWorkspaceStore) { + let temp = tempfile::tempdir().unwrap(); + let store = SqliteWorkspaceStore::open(temp.path().join("server.db")).unwrap(); + store + .with_conn(|conn| { + conn.execute_batch( + "INSERT INTO accounts(account_id, kind, handle, display_name, created_at, updated_at) \ + VALUES ('owner', 'user', 'owner', 'Owner', '1', '1'); \ + INSERT INTO workspaces(workspace_id, owner_account_id, display_name, state, created_at, updated_at) \ + VALUES ('workspace-a', 'owner', 'Workspace A', 'active', '1', '1'); \ + INSERT INTO workspace_runtime_bindings(\ + workspace_id, runtime_id, display_name, base_url, public_key, \ + public_key_fingerprint, binding_revision, state, authentication_mode, \ + created_at, updated_at\ + ) VALUES (\ + 'workspace-a', 'runtime-a', 'Runtime A', 'https://runtime.invalid', \ + 'key-a', 'fingerprint-a', 3, 'verified', 'legacy_server_issuer', '1', '1'\ + );", + )?; + Ok(()) + }) + .unwrap(); + (temp, store) + } + + fn assert_runtime_removal_binding_unchanged(store: &SqliteWorkspaceStore) { + let binding = store + .get_workspace_runtime_binding("workspace-a", "runtime-a") + .unwrap() + .expect("Runtime binding must remain"); + assert_eq!(binding.binding_revision, 3); + assert_eq!(binding.state, WorkspaceRuntimeBindingState::Verified); + assert!(binding.revoked_at.is_none()); + } + + #[test] + fn runtime_removal_preflight_rejects_stale_revision_and_active_resources_without_revoking_trust() + { + let (_temp, store) = runtime_removal_test_store(); + let stale = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-stale", + "fingerprint-stale", + 2, + 1, + ) + .unwrap_err(); + assert!(matches!( + stale, + Error::RuntimeBindingRevisionConflict { .. } + )); + assert_runtime_removal_binding_unchanged(&store); + + store + .with_conn(|conn| { + conn.execute( + "INSERT INTO worker_registry(\ + workspace_id, worker_id, runtime_id, display_name, retention_state, created_at, updated_at\ + ) VALUES ('workspace-a', 'worker-a', 'runtime-a', 'Worker A', 'normal', '1', '1')", + [], + )?; + Ok(()) + }) + .unwrap(); + let worker = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-worker", + "fingerprint-worker", + 3, + 1, + ) + .unwrap_err(); + assert!( + worker + .to_string() + .contains("runtime_removal_active_worker_blocked") + ); + assert_runtime_removal_binding_unchanged(&store); + } + + #[test] + fn runtime_removal_preflight_rejects_workdir_and_pending_create_without_revoking_trust() { + let (_temp, store) = runtime_removal_test_store(); + store + .with_conn(|conn| { + conn.execute_batch( + "INSERT INTO repositories(\ + workspace_id, repository_id, repository_key, kind, uri, created_at, updated_at, \ + source_kind, source_uri, source_revision, source_fingerprint, observed_status\ + ) VALUES (\ + 'workspace-a', 'repository-a', 'repository-a', 'git', '/repo', '1', '1', \ + 'local', '/repo', 1, 'source-a', 'unverified'\ + ); \ + INSERT INTO workdir_registry(\ + workspace_id, workdir_id, runtime_id, repository_id, materialization_status, \ + cleanliness, created_at, updated_at\ + ) VALUES (\ + 'workspace-a', 'workdir-a', 'runtime-a', 'repository-a', 'present', \ + 'clean', '1', '1'\ + );", + )?; + Ok(()) + }) + .unwrap(); + let workdir = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-workdir", + "fingerprint-workdir", + 3, + 1, + ) + .unwrap_err(); + assert!( + workdir + .to_string() + .contains("runtime_removal_active_workdir_blocked") + ); + assert_runtime_removal_binding_unchanged(&store); + + store + .with_conn(|conn| { + conn.execute("DELETE FROM workdir_registry", [])?; + conn.execute( + "INSERT INTO workdir_create_operations(\ + workspace_id, operation_id, request_fingerprint, repository_id, \ + resolved_runtime_id, config_revision, config_projection_digest, \ + working_directory_id, state, created_at, updated_at\ + ) VALUES (\ + 'workspace-a', 'create-workdir', 'request-a', 'repository-a', \ + 'runtime-a', 1, 'projection-a', 'workdir-b', 'pending', '1', '1'\ + )", + [], + )?; + Ok(()) + }) + .unwrap(); + let pending = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-pending", + "fingerprint-pending", + 3, + 1, + ) + .unwrap_err(); + assert!( + pending + .to_string() + .contains("runtime_removal_workdir_create_blocked") + ); + assert_runtime_removal_binding_unchanged(&store); + } + + #[test] + fn runtime_removal_rejects_shared_binding_and_fences_new_binding_until_cleanup() { + let (_temp, store) = runtime_removal_test_store(); + store + .with_conn(|conn| { + conn.execute_batch( + "INSERT INTO accounts(account_id, kind, handle, display_name, created_at, updated_at) \ + VALUES ('owner-b', 'user', 'owner-b', 'Owner B', '1', '1'); \ + INSERT INTO workspaces(workspace_id, owner_account_id, display_name, state, created_at, updated_at) \ + VALUES ('workspace-b', 'owner-b', 'Workspace B', 'active', '1', '1'); \ + INSERT INTO workspace_runtime_bindings(\ + workspace_id, runtime_id, display_name, base_url, public_key, \ + public_key_fingerprint, binding_revision, state, authentication_mode, \ + created_at, updated_at\ + ) VALUES (\ + 'workspace-b', 'runtime-a', 'Runtime A', 'https://runtime.invalid', \ + 'key-a', 'fingerprint-b', 1, 'verified', 'legacy_server_issuer', '1', '1'\ + );", + )?; + Ok(()) + }) + .unwrap(); + let shared = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-shared", + "fingerprint-shared", + 3, + 1, + ) + .unwrap_err(); + assert!( + shared + .to_string() + .contains("runtime_removal_other_workspace_binding_blocked") + ); + assert_runtime_removal_binding_unchanged(&store); + assert!( + store + .get_workspace_runtime_binding("workspace-b", "runtime-a") + .unwrap() + .is_some() + ); + + store + .with_conn(|conn| { + conn.execute( + "DELETE FROM workspace_runtime_bindings WHERE workspace_id = 'workspace-b'", + [], + )?; + Ok(()) + }) + .unwrap(); + store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-fenced", + "fingerprint-fenced", + 3, + 1, + ) + .unwrap(); + let blocked_insert = store.with_conn(|conn| { + conn.execute( + "INSERT INTO workspace_runtime_bindings(\ + workspace_id, runtime_id, display_name, base_url, public_key, \ + public_key_fingerprint, binding_revision, state, authentication_mode, \ + created_at, updated_at\ + ) VALUES (\ + 'workspace-b', 'runtime-a', 'Runtime A', 'https://runtime.invalid', \ + 'key-a', 'fingerprint-b', 1, 'verified', 'legacy_server_issuer', '1', '1'\ + )", + [], + )?; + Ok(()) + }); + assert!( + blocked_insert + .unwrap_err() + .to_string() + .contains("runtime_removal_in_progress") + ); + } + + #[test] + fn runtime_removal_checkpoint_replays_and_recovers_after_reopen() { + let (temp, store) = runtime_removal_test_store(); + let reserved = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-runtime-a", + "fingerprint-a", + 3, + 7, + ) + .unwrap(); + assert!(!reserved.replay); + let failed = store + .mark_runtime_removal_failed("remove-runtime-a", "injected_cleanup_failure") + .unwrap(); + assert_eq!(failed.state, RuntimeRemovalOperationState::Failed); + assert_runtime_removal_binding_unchanged(&store); + let retried = store + .reserve_runtime_removal( + "workspace-a", + "runtime-a", + "remove-runtime-a", + "fingerprint-a", + 3, + 7, + ) + .unwrap(); + assert!(retried.replay); + let checkpoint = store + .commit_runtime_binding_removal("remove-runtime-a") + .unwrap(); + assert_eq!( + checkpoint.state, + RuntimeRemovalOperationState::CleanupPending + ); + assert!(checkpoint.binding_removed); + drop(store); + + let reopened = Connection::open(temp.path().join("server.db")).unwrap(); + let (state, binding_removed): (String, i64) = reopened + .query_row( + "SELECT state, binding_removed FROM runtime_removal_operations \ + WHERE operation_id = 'remove-runtime-a'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!(state, "cleanup_pending"); + assert_eq!(binding_removed, 1); + reopened + .execute( + "UPDATE runtime_removal_operations \ + SET state = 'succeeded', runtime_registration_removed = 1, completed_at = '2', updated_at = '2' \ + WHERE operation_id = 'remove-runtime-a' AND state = 'cleanup_pending'", + [], + ) + .unwrap(); + let completed: (String, i64) = reopened + .query_row( + "SELECT state, runtime_registration_removed FROM runtime_removal_operations \ + WHERE operation_id = 'remove-runtime-a'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!(completed, ("succeeded".to_string(), 1)); + } + + #[test] + fn schema_v59_upgrade_adds_runtime_removal_authority_atomically() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("server.db"); + let store = SqliteWorkspaceStore::open(&path).unwrap(); + store + .with_conn(|conn| { + conn.execute_batch( + "DROP TRIGGER runtime_binding_insert_blocked_by_removal; \ + DROP TRIGGER runtime_binding_update_blocked_by_removal; \ + DROP TABLE runtime_removal_operations; \ + DELETE FROM __yoi_schema_migrations; \ + INSERT INTO __yoi_schema_migrations(version, name) \ + VALUES (59, 'workspace schema baseline');", + )?; + Ok(()) + }) + .unwrap(); + drop(store); + + let migrated = SqliteWorkspaceStore::open(&path).unwrap(); + migrated + .with_conn(|conn| { + let table_count: i64 = conn.query_row( + "SELECT COUNT(*) FROM sqlite_master \ + WHERE type = 'table' AND name = 'runtime_removal_operations'", + [], + |row| row.get(0), + )?; + let trigger_count: i64 = conn.query_row( + "SELECT COUNT(*) FROM sqlite_master \ + WHERE type = 'trigger' AND name LIKE 'runtime_binding_%_blocked_by_removal'", + [], + |row| row.get(0), + )?; + let foreign_key_failures: i64 = + conn.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| { + row.get(0) + })?; + assert_eq!(table_count, 1); + assert_eq!(trigger_count, 2); + assert_eq!(foreign_key_failures, 0); + Ok(()) + }) + .unwrap(); + } + #[test] fn current_schema_accepts_every_retained_canonical_provenance() { for baseline_version in OLDEST_SCHEMA_VERSION..=LATEST_SCHEMA_VERSION { @@ -9745,6 +10705,9 @@ mod tests { r#" DROP TABLE workdir_create_credential_revision_retentions; DROP TABLE workdir_create_credential_candidates; + DROP TRIGGER runtime_binding_insert_blocked_by_removal; + DROP TRIGGER runtime_binding_update_blocked_by_removal; + DROP TABLE runtime_removal_operations; DROP INDEX workspace_signing_identity_audit_workspace_idx; DROP TABLE workspace_signing_identity_audit; DROP TABLE workspace_signing_identity_provisioning_operations; @@ -9890,6 +10853,10 @@ mod tests { version: 59, name: WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME.to_string(), }, + WorkspaceSchemaMigrationStep { + version: 60, + name: RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME.to_string(), + }, ] ); @@ -9932,6 +10899,10 @@ mod tests { 59, WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME.to_string(), ), + ( + 60, + RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME.to_string(), + ), ] ); assert!(!table_exists(conn, "trusted_runtime_records")?); @@ -10002,7 +10973,7 @@ mod tests { .iter() .map(|migration| migration.version) .collect::>(), - vec![52, 53, 54, 55, 56, 57, 58, 59] + vec![52, 53, 54, 55, 56, 57, 58, 59, 60] ); SqliteWorkspaceStore::migrate_database(&path).unwrap(); let conn = Connection::open(&path).unwrap(); @@ -10010,7 +10981,7 @@ mod tests { current_schema_version(&conn).unwrap(), LATEST_SCHEMA_VERSION ); - assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 10); + assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 11); } #[test] diff --git a/crates/workspace-server/src/workspace_deletion.rs b/crates/workspace-server/src/workspace_deletion.rs index 83af4ad7..a74f2c51 100644 --- a/crates/workspace-server/src/workspace_deletion.rs +++ b/crates/workspace-server/src/workspace_deletion.rs @@ -32,6 +32,7 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[ "objective_ticket_links", "objectives", "repositories", + "runtime_removal_operations", "repository_secret_audit_events", "repository_secret_operations", "repository_ssh_credential_revisions",