Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix(server): retain trust-key revoke publication after cancellation

Browse Source
This commit is contained in:
Hare
2026-10-12 12:12:50 +09:00
parent 6d8c49b72d
commit 1a065db0fc
2 changed files with 290 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/server.rs
+68 -54
View File
@@ -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<Error> for server_api::RepositoryApiError {
}
}
impl From<Error> for server_api::RuntimeManagementApiError {
fn from(error: Error) -> Self {
Self::from(server_api::RepositoryApiError::from(error))
}
}
impl From<Error> for ApiError {
fn from(error: Error) -> Self {
let diagnostics = match &error {
crates/workspace-server/src/server_db_execution_tests.rs
+222
View File
@@ -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<mpsc::Sender<()>> = 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;