From 1a065db0fc5d07e2dcc1b9dac2fcfc0607b9a23c Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 12 Oct 2026 12:12:50 +0900 Subject: [PATCH] fix(server): retain trust-key revoke publication after cancellation --- crates/workspace-server/src/server.rs | 122 +++++----- .../src/server_db_execution_tests.rs | 222 ++++++++++++++++++ 2 files changed, 290 insertions(+), 54 deletions(-) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index becd6fd80..3283f13da 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -9403,62 +9403,70 @@ impl server_api::ServerApi for ServerApiContractService { "expected_binding_id must contain between 1 and 256 bytes", ))); } - let now = Utc::now().to_rfc3339(); - if let Err(error) = api - .store - .revoke_workspace_runtime_binding_key( - &workspace_id, - &runtime_id, - &request.expected_binding_id, - &actor.account_id, - &now, - ) - .await - { - let kind = match error { - Error::RuntimeBindingIdConflict { .. } => { - Some(server_api::RuntimeTrustConflictKind::StaleBinding) - } - Error::RuntimeBindingFingerprintConflict { .. } => { - Some(server_api::RuntimeTrustConflictKind::FingerprintInUse) - } - _ => None, - }; - if let Some(kind) = kind { - let current = api - .store - .get_workspace_runtime_binding(&workspace_id, &runtime_id) - .await - .map_err(|error| map_error(ApiError::from(error)))?; - return Err(server_api::RuntimeManagementApiError::trust_conflict( - server_api::RuntimeTrustConflictResponse { - error: kind, - message: match kind { - server_api::RuntimeTrustConflictKind::StaleBinding => { - "the Runtime trust binding changed; reload before retrying".to_string() - } - server_api::RuntimeTrustConflictKind::FingerprintInUse => { - "the public key is already bound to another Runtime in this Workspace".to_string() - } + // Own the whole revoke outcome through local publication and response. + // The accepted DB mutation outlives a cancelled HTTP waiter; its Host + // parent must also remove expectations and terminate subscriptions. + // SQL keeps its independent owner, and response reads stay on this Host + // workflow rather than attempting nested admission to the same lane. + run_host_workflow(api.host_executors.host()?, async move { + let now = Utc::now().to_rfc3339(); + if let Err(error) = api + .store + .revoke_workspace_runtime_binding_key( + &workspace_id, + &runtime_id, + &request.expected_binding_id, + &actor.account_id, + &now, + ) + .await + { + let kind = match error { + Error::RuntimeBindingIdConflict { .. } => { + Some(server_api::RuntimeTrustConflictKind::StaleBinding) + } + Error::RuntimeBindingFingerprintConflict { .. } => { + Some(server_api::RuntimeTrustConflictKind::FingerprintInUse) + } + _ => None, + }; + if let Some(kind) = kind { + let current = api + .store + .get_workspace_runtime_binding(&workspace_id, &runtime_id) + .await + .map_err(|error| map_error(ApiError::from(error)))?; + return Err(server_api::RuntimeManagementApiError::trust_conflict( + server_api::RuntimeTrustConflictResponse { + error: kind, + message: match kind { + server_api::RuntimeTrustConflictKind::StaleBinding => { + "the Runtime trust binding changed; reload before retrying".to_string() + } + server_api::RuntimeTrustConflictKind::FingerprintInUse => { + "the public key is already bound to another Runtime in this Workspace".to_string() + } + }, + current_binding_id: current.as_ref().map(|binding| binding.binding_id.clone()), + current_fingerprint: current + .as_ref() + .map(|binding| binding.public_key_fingerprint.clone()), }, - current_binding_id: current.as_ref().map(|binding| binding.binding_id.clone()), - current_fingerprint: current - .as_ref() - .map(|binding| binding.public_key_fingerprint.clone()), - }, - )); + )); + } + return Err(map_error(ApiError::from(error))); } - return Err(map_error(ApiError::from(error))); - } - api.runtime_binding_expectations - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .remove(&(workspace_id.clone(), runtime_id.clone())); - api.runtime_subscription_broker - .unregister_runtime(&runtime_id); - workspace_runtime_detail(&api, &workspace_id, &runtime_id) - .await - .map_err(map_error) + api.runtime_binding_expectations + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .remove(&(workspace_id.clone(), runtime_id.clone())); + api.runtime_subscription_broker + .unregister_runtime(&runtime_id); + workspace_runtime_detail_on_host(&api, &workspace_id, &runtime_id) + .await + .map_err(map_error) + }) + .await } async fn runtime_connection_test( @@ -34715,6 +34723,12 @@ impl From for server_api::RepositoryApiError { } } +impl From for server_api::RuntimeManagementApiError { + fn from(error: Error) -> Self { + Self::from(server_api::RepositoryApiError::from(error)) + } +} + impl From for ApiError { fn from(error: Error) -> Self { let diagnostics = match &error { diff --git a/crates/workspace-server/src/server_db_execution_tests.rs b/crates/workspace-server/src/server_db_execution_tests.rs index 4d287776c..6cfa24529 100644 --- a/crates/workspace-server/src/server_db_execution_tests.rs +++ b/crates/workspace-server/src/server_db_execution_tests.rs @@ -1370,6 +1370,228 @@ async fn runtime_connection_post_ping_sql_waits_off_tokio_while_host_control_pro fixture.executor.drain().await.unwrap(); } +#[tokio::test(flavor = "current_thread")] +async fn saturated_host_rejects_runtime_key_revoke_without_mutating_trust() { + let fixture = Fixture::new().await; + fixture.api.worker_projection.shutdown_async().await; + observed(fixture.executor.drain()).await.unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let binding = install_remote_fixture( + &fixture, + "revoke-runtime", + &format!("http://{}", listener.local_addr().unwrap()), + true, + ); + fixture + .api + .runtime_binding_expectations + .write() + .unwrap() + .insert( + (WORKSPACE.into(), binding.runtime_id.clone()), + binding.clone(), + ); + let initial_registrations = fixture.api.runtime_subscription_broker.runtime_ids(); + // Sender Drop also releases each controlled job if a route assertion panics. + let mut releases: Vec> = Vec::new(); + let (started, running) = oneshot::channel(); + let (release, permit) = mpsc::channel(); + releases.push(release); + fixture + .host_executor + .dispatch(move || { + let _ = started.send(()); + let _ = permit.recv_timeout(WATCHDOG); + }) + .unwrap(); + observed(running).await.unwrap(); + let (release, permit) = mpsc::channel(); + releases.push(release); + fixture + .host_executor + .dispatch(move || { + let _ = permit.recv_timeout(WATCHDOG); + }) + .unwrap(); + let body = response_json( + observed(fixture.dispatch(Fixture::request( + "DELETE", + "/api/w/db-execution/runtimes/revoke-runtime/trust-key", + serde_json::json!({"expected_binding_id": binding.binding_id}), + ))) + .await + .unwrap(), + StatusCode::SERVICE_UNAVAILABLE, + ) + .await; + assert!(body.to_string().contains("db_capacity_exceeded"), "{body}"); + assert_eq!( + fixture + .store + .get_workspace_runtime_binding(WORKSPACE, &binding.runtime_id) + .unwrap() + .unwrap(), + binding + ); + assert!( + fixture + .store + .list_workspace_runtime_binding_audit(WORKSPACE, &binding.runtime_id, 10) + .unwrap() + .is_empty() + ); + assert_eq!( + fixture.api.runtime_subscription_broker.runtime_ids(), + initial_registrations + ); + assert_eq!( + fixture + .api + .runtime_binding_expectations + .read() + .unwrap() + .get(&(WORKSPACE.into(), binding.runtime_id.clone())), + Some(&binding) + ); + drop(releases); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); +} + +#[tokio::test(flavor = "current_thread")] +async fn cancelled_runtime_key_revoke_commits_audit_and_removes_broker_registration() { + let fixture = Fixture::new().await; + // Stop unrelated projection writes before observing this mutation's SQLite + // wait. Keep the original runtime alive through actual workflow cleanup. + fixture.api.worker_projection.shutdown_async().await; + observed(fixture.executor.drain()).await.unwrap(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let binding = install_remote_fixture( + &fixture, + "revoke-runtime", + &format!("http://{}", listener.local_addr().unwrap()), + true, + ); + fixture + .api + .runtime_binding_expectations + .write() + .unwrap() + .insert( + (WORKSPACE.into(), binding.runtime_id.clone()), + binding.clone(), + ); + let mut config = remote_runtime_config_from_binding(&binding).unwrap(); + config.workspace_authorization = Some(WorkspaceRuntimeAuthorization::new( + fixture.store.clone(), + fixture.api.signing_identities.clone(), + "http://localhost:8787", + binding.clone(), + )); + fixture + .api + .runtime_subscription_broker + .register_remote_runtime(config); + let mut subscription = fixture + .api + .runtime_subscription_broker + .subscribe( + &binding.runtime_id, + protocol::subscription::EventSubscriptionSelector::RuntimeWorkers, + ) + .unwrap(); + let (mut busy, waiting) = BusyObservation::install_store(fixture.store.clone()); + let mut lock = WriteLock::acquire(fixture.api.config.database_path.clone()); + let server = fixture.server.clone(); + let request = Fixture::request( + "DELETE", + "/api/w/db-execution/runtimes/revoke-runtime/trust-key", + serde_json::json!({"expected_binding_id": binding.binding_id}), + ); + let revoke = + tokio::spawn(async move { dispatch_workspace_request(State(server), request).await }); + observed(waiting).await.unwrap(); // Actual authenticated revoke reached BEGIN IMMEDIATE. + runtime_progress_while_locked(&lock).await; + revoke.abort(); + assert!(revoke.await.unwrap_err().is_cancelled()); + let admission = fixture.server.admission_for_workspace(WORKSPACE).await; + let mut deletion = Box::pin(admission.lock_deletion()); + assert!( + futures::poll!(&mut deletion).is_pending(), + "cancelled HTTP must retain actual revoke ownership until publication" + ); + assert!( + fixture + .api + .runtime_subscription_broker + .runtime_ids() + .contains(&binding.runtime_id) + ); + lock.release(); + busy.resume(); + observed(fixture.host_executor.drain()).await.unwrap(); + observed(fixture.executor.drain()).await.unwrap(); + let _deletion = observed(deletion).await; + let saved = fixture + .store + .get_workspace_runtime_binding(WORKSPACE, &binding.runtime_id) + .unwrap() + .unwrap(); + assert!(saved.revoked_at.is_some()); + assert_ne!(saved.binding_id, binding.binding_id); + let audit = fixture + .store + .list_workspace_runtime_binding_audit(WORKSPACE, &binding.runtime_id, 10) + .unwrap(); + assert_eq!( + audit.len(), + 1, + "cancelled mutation must commit one audit without retry" + ); + assert_eq!(audit[0].action, "revoked"); + assert_eq!(audit[0].actor_account_id, format!("account-{WORKSPACE}")); + assert_eq!(audit[0].binding_id, saved.binding_id); + assert_eq!(audit[0].at, saved.revoked_at.unwrap()); + assert!( + !fixture + .api + .runtime_subscription_broker + .runtime_ids() + .contains(&binding.runtime_id), + "committed revoke must unregister even after HTTP waiter cancellation" + ); + assert!( + !fixture + .api + .runtime_binding_expectations + .read() + .unwrap() + .contains_key(&(WORKSPACE.into(), binding.runtime_id.clone())) + ); + // Exercise downstream termination at the broker protocol boundary, not a + // product process or complete browser/Runtime E2E flow. + observed(async { + loop { + match subscription.recv().await { + Some(crate::runtime_subscription::BrokerSubscriptionEvent::Closed { + code, .. + }) => { + assert_eq!( + code, + protocol::subscription::SubscriptionTerminationCode::ServerShutdown + ); + break; + } + Some(crate::runtime_subscription::BrokerSubscriptionEvent::Disconnected { + .. + }) => {} + event => panic!("expected revoke shutdown, got {event:?}"), + } + } + }) + .await; +} + #[tokio::test(flavor = "current_thread")] async fn cancelled_runtime_metadata_mutation_keeps_commit_and_registration_publication_owned() { let fixture = Fixture::new().await;