From 9f13446afd40da5618ada0f4fff401fb2719acdc Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 30 Oct 2026 05:02:21 +0900 Subject: [PATCH] fix: interrupt revoked Plugin transports and enforce full control-plane exclusions --- crates/plugin-runtime/src/owner.rs | 6 + .../src/plugin_https/test_support.rs | 27 +- crates/workspace-server/src/server.rs | 27 ++ .../src/server_db_execution_tests.rs | 1 + .../src/server_plugin_host.rs | 164 ++++++--- .../src/server_plugin_host_tests.rs | 321 ++++++++++++++++++ .../src/worker_source/runtime_request.rs | 1 + 7 files changed, 506 insertions(+), 41 deletions(-) diff --git a/crates/plugin-runtime/src/owner.rs b/crates/plugin-runtime/src/owner.rs index cf74bc53a..2bfbb449b 100644 --- a/crates/plugin-runtime/src/owner.rs +++ b/crates/plugin-runtime/src/owner.rs @@ -44,6 +44,12 @@ pub trait PluginHostOwner: Send + Sync { } } +/// Finite generated-client response budget for the supported one-MiB binary +/// body, its worst-case JSON numeric-array expansion, headers and envelope. +/// Set `response_body_limit` on the injected client before constructing a Remote +/// owner; a smaller operator-selected limit may reject otherwise valid replies. +pub const PLUGIN_HOST_RESPONSE_BODY_LIMIT: usize = 4 * 1024 * 1024 + 512 * 1024; + /// Uses an injected, already configured generated client (including its existing /// Runtime request authorizer). This adapter never creates or owns credentials. pub struct RemotePluginHostOwner { diff --git a/crates/workspace-server/src/plugin_https/test_support.rs b/crates/workspace-server/src/plugin_https/test_support.rs index ba45b1fc4..8f471b2ac 100644 --- a/crates/workspace-server/src/plugin_https/test_support.rs +++ b/crates/workspace-server/src/plugin_https/test_support.rs @@ -108,6 +108,7 @@ pub(crate) struct Peer { pub(super) root: reqwest::Certificate, pub(super) observations: mpsc::Receiver>, pub(super) connections: Arc, + closed: mpsc::Receiver<()>, task: JoinHandle<()>, _directory: tempfile::TempDir, } @@ -164,6 +165,7 @@ impl Peer { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let address = listener.local_addr().unwrap(); let (observed, observations) = mpsc::channel(16); + let (closed_sender, closed) = mpsc::channel(16); let connections = Arc::new(AtomicUsize::new(0)); let count = connections.clone(); let mut replies = VecDeque::from(replies); @@ -176,6 +178,7 @@ impl Peer { count.fetch_add(1, Ordering::SeqCst); let acceptor = acceptor.clone(); let observed = observed.clone(); + let closed_sender = closed_sender.clone(); let reply = replies.pop_front().unwrap_or(Reply::Close); handlers.spawn(async move { let Ok(mut socket) = acceptor.accept(socket).await else { return; }; @@ -185,7 +188,7 @@ impl Peer { let Ok(length) = socket.read(&mut buffer).await else { return; }; if length == 0 { return; } bytes.extend_from_slice(&buffer[..length]); - assert!(bytes.len() <= 64 * 1024, "test request exceeded fixture bound"); + assert!(bytes.len() <= 2 * 1024 * 1024, "test request exceeded fixture bound"); if let Some(end) = bytes.windows(4).position(|part| part == b"\r\n\r\n") { let text = String::from_utf8_lossy(&bytes[..end]).to_lowercase(); let length = text.lines().find_map(|line| line.strip_prefix("content-length: ")) @@ -197,7 +200,13 @@ impl Peer { match reply { Reply::Bytes(bytes) => { let _ = socket.write_all(&bytes).await; let _ = socket.shutdown().await; } Reply::StallAfter(bytes) => { let _ = socket.write_all(&bytes).await; pending::<()>().await; } - Reply::Stall => pending::<()>().await, + Reply::Stall => { + let mut byte = [0]; + match socket.read(&mut byte).await { + Ok(0) | Err(_) => { let _ = closed_sender.send(()).await; } + Ok(_) => panic!("unexpected bytes after complete request"), + } + }, Reply::Close => {}, Reply::Gated(release) => { if let Ok(bytes) = release.await { @@ -217,6 +226,7 @@ impl Peer { root: reqwest::Certificate::from_pem(&certificate).unwrap(), observations, connections, + closed, task, _directory: directory, } @@ -232,6 +242,19 @@ impl Peer { self.connections.load(Ordering::SeqCst) } + pub(crate) async fn closed(&mut self) { + tokio::time::timeout(Duration::from_secs(2), self.closed.recv()) + .await + .unwrap() + .unwrap(); + } + + pub(crate) fn transport_with_policy(&self, policy: TransportPolicy) -> HttpsTransport { + let mut transport = self.transport_for_requests(4); + transport.policy = policy; + transport + } + pub(crate) fn transport(&self) -> HttpsTransport { self.transport_for_requests(1) } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 353db3fd6..d5894b789 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -307,6 +307,9 @@ pub struct ServerConfig { pub runtime_event_sources: Vec, pub remote_runtime_sources: Vec, pub backend_base_url: Option, + // Operational deny inventory survives Workspace capability filtering. It is + // never deserialized from Workspace configuration or returned to a guest. + pub(crate) control_plane_endpoints: std::collections::BTreeSet, } impl ServerConfig { @@ -344,6 +347,7 @@ impl ServerConfig { runtime_event_sources: Vec::new(), remote_runtime_sources: Vec::new(), backend_base_url: None, + control_plane_endpoints: Default::default(), } } @@ -414,6 +418,28 @@ impl ServerConfig { self } + fn plugin_control_plane_endpoints(&self) -> std::collections::BTreeSet { + let AuthConfig::Passkey { + public_base_url, .. + } = &self.auth; + self.control_plane_endpoints + .iter() + .chain(std::iter::once(public_base_url)) + .chain(self.backend_base_url.iter()) + .chain( + self.remote_runtime_sources + .iter() + .map(|source| &source.base_url), + ) + .chain( + self.runtime_event_sources + .iter() + .map(|source| &source.endpoint), + ) + .cloned() + .collect() + } + fn for_catalog_workspace( &self, workspace: &WorkspaceRecord, @@ -456,6 +482,7 @@ impl ServerConfig { scoped.repositories = repositories; // Runtime trust is server-global. Only explicitly assigned sources enter // this Workspace's registry and receive Workspace-scoped capabilities. + scoped.control_plane_endpoints = self.plugin_control_plane_endpoints(); scoped.remote_runtime_sources.retain(|runtime| { runtime.workspace_id.as_deref() == Some(workspace.workspace_id.as_str()) }); diff --git a/crates/workspace-server/src/server_db_execution_tests.rs b/crates/workspace-server/src/server_db_execution_tests.rs index 44a9fbb12..ecf74eb06 100644 --- a/crates/workspace-server/src/server_db_execution_tests.rs +++ b/crates/workspace-server/src/server_db_execution_tests.rs @@ -77,6 +77,7 @@ impl Fixture { runtime_event_sources: Vec::new(), remote_runtime_sources: Vec::new(), backend_base_url: Some("http://localhost:8787".into()), + control_plane_endpoints: Default::default(), }; let store = Arc::new( SqliteWorkspaceStore::open_for_test(&config.database_path) diff --git a/crates/workspace-server/src/server_plugin_host.rs b/crates/workspace-server/src/server_plugin_host.rs index 7ce444e5f..627ede4e1 100644 --- a/crates/workspace-server/src/server_plugin_host.rs +++ b/crates/workspace-server/src/server_plugin_host.rs @@ -77,7 +77,14 @@ use tokio_util::sync::CancellationToken; const RECEIPT_RETENTION_MS: i64 = 30 * 24 * 60 * 60 * 1000; pub(super) fn router(service: Arc) -> Router { - server_api_axum::plugin_host(service).layer(DefaultBodyLimit::max(2 * 1024 * 1024)) + // Vec uses JSON's numeric array representation (up to four wire bytes + // per decoded byte). Reserve bounded room for IDs, URL/query and escaped + // headers; the shared owner still enforces every decoded field/body limit. + let body_ceiling = crate::plugin_binding::PluginConnectionPolicyCeiling::default() + .limits + .max_request_bytes as usize; + server_api_axum::plugin_host(service) + .layer(DefaultBodyLimit::max(4 * body_ceiling + 512 * 1024)) } fn denied() -> PluginHostError { @@ -623,32 +630,41 @@ async fn execute_with_transport( None => Arc::new(transport(&api).await?), }; let cancellation = CancellationToken::new(); - let result = transport - .execute_with_guard( - binding, - prepared, - secret.as_deref().map(|v| v.as_str()), - &cancellation, - || async { - invocation( - api.clone(), - &context, - &workspace, - &invocation_id, - operation.clone(), - ) - .await - .map(|_| ()) - .map_err(|_| https::TransportError::Denied) - }, - ) - .await - .map(|response| PluginHttpResponse { + let sending = transport.execute_with_guard( + binding, + prepared, + secret.as_deref().map(|v| v.as_str()), + &cancellation, + || async { + invocation( + api.clone(), + &context, + &workspace, + &invocation_id, + operation.clone(), + ) + .await + .map(|_| ()) + .map_err(|_| https::TransportError::Denied) + }, + ); + // Monitor the existing durable owners rather than maintaining + // another revocation registry. Dropping the transport future + // shuts down its single-use client/socket. A stalled authority + // owner also closes the attempt; it is never permission to run + // until the HTTP deadline. Admission remains consumed even if + // the shutdown acknowledgement or completion write is lost. + let result = tokio::select! { + result = sending => result.map(|response| PluginHttpResponse { status: response.status, headers: response.headers, body: response.body, - }) - .map_err(sent_transport_error); + }).map_err(sent_transport_error), + error = monitor_invocation(api.clone(), &context, &workspace, &invocation_id, operation.clone()) => { + cancellation.cancel(); + Err(error) + } + }; let store = api.config_store.clone(); let r = record; let fp = fingerprint; @@ -699,25 +715,92 @@ async fn execute_with_transport( } } +async fn monitor_invocation( + api: Arc, + context: &ServerRequestContext, + workspace: &str, + invocation_id: &str, + operation: PluginHostOperation, +) -> PluginHostError { + loop { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let checked = tokio::time::timeout( + std::time::Duration::from_secs(1), + invocation( + api.clone(), + context, + workspace, + invocation_id, + operation.clone(), + ), + ) + .await; + let error = match checked { + Ok(Ok(_)) => continue, + Ok(Err(error)) => error, + Err(_) => policy::map_error(policy::failure(PluginHostErrorCode::Unavailable)), + }; + // After admission the request may have reached the peer. Neither a + // failed check nor a local connection shutdown proves remote rollback. + return PluginHostError { + effect: PluginEffectObservation::Indeterminate, + ..error + }; + } +} + async fn transport( api: &WorkspaceApi, ) -> std::result::Result { + let policy = transport_policy(api, |host, port| async move { + tokio::net::lookup_host((host, port)) + .await + .map(|addresses| addresses.take(65).collect::>()) + .map_err(|_| https::TransportError::Dns) + }) + .await?; + Ok(https::HttpsTransport::new(policy)) +} + +pub(super) async fn transport_policy( + api: &WorkspaceApi, + resolve: F, +) -> std::result::Result +where + F: Fn(String, u16) -> Fut, + Fut: std::future::Future< + Output = std::result::Result, https::TransportError>, + >, +{ let mut hosts = std::collections::BTreeSet::new(); let mut ips = std::collections::BTreeSet::new(); - let AuthConfig::Passkey { - public_base_url, .. - } = &api.config.auth; - for address in std::iter::once(public_base_url) - .chain(api.config.backend_base_url.iter()) - .chain( - api.config - .remote_runtime_sources - .iter() - .map(|source| &source.base_url), - ) + // Current server-owned bindings include dynamically registered endpoints in + // other Workspaces, even when they were absent from the startup config. + let store = api.config_store.clone(); + let registered = store + .db_executor() + .run(move || { + store.with_conn(|conn| { + let mut statement = + conn.prepare("SELECT DISTINCT base_url FROM workspace_runtime_bindings")?; + let rows = statement.query_map([], |row| row.get::<_, String>(0))?; + Ok(rows.collect::, _>>()?) + }) + }) + .await + .map_err(policy::map_error)?; + for address in api + .config + .plugin_control_plane_endpoints() + .iter() + .chain(registered.iter()) { if let Ok(url) = url::Url::parse(address) { + if !matches!(url.scheme(), "http" | "https" | "ws" | "wss") { + continue; + } if let Some(host) = url.host_str() { + let host = host.trim_start_matches('[').trim_end_matches(']'); hosts.insert(host.to_ascii_lowercase()); if let Ok(ip) = host.parse() { ips.insert(ip); @@ -725,12 +808,15 @@ async fn transport( let port = url.port_or_known_default().ok_or_else(denied)?; let resolved = tokio::time::timeout( std::time::Duration::from_secs(5), - tokio::net::lookup_host((host, port)), + resolve(host.to_owned(), port), ) .await .map_err(|_| transport_error(https::TransportError::Dns))? .map_err(|_| transport_error(https::TransportError::Dns))?; - for (index, address) in resolved.enumerate() { + if resolved.is_empty() { + return Err(transport_error(https::TransportError::Dns)); + } + for (index, address) in resolved.into_iter().enumerate() { if index >= 64 { return Err(transport_error(https::TransportError::Dns)); } @@ -740,10 +826,10 @@ async fn transport( } } } - Ok(https::HttpsTransport::new(https::TransportPolicy { + Ok(https::TransportPolicy { denied_addresses: ips, denied_hosts: hosts, - })) + }) } fn sent_transport_error(error: https::TransportError) -> PluginHostError { match error { diff --git a/crates/workspace-server/src/server_plugin_host_tests.rs b/crates/workspace-server/src/server_plugin_host_tests.rs index 7ba43c5d3..5954c04f2 100644 --- a/crates/workspace-server/src/server_plugin_host_tests.rs +++ b/crates/workspace-server/src/server_plugin_host_tests.rs @@ -1258,6 +1258,7 @@ async fn remote_owner_for( })); let client = server_api::ServerApiClient::builder(format!("http://{address}")) .unwrap() + .response_body_limit(yoi_plugin_runtime::owner::PLUGIN_HOST_RESPONSE_BODY_LIMIT) .client( reqwest::Client::builder() .no_proxy() @@ -1557,3 +1558,323 @@ async fn reserved_job_without_canonical_job_binding_cannot_fall_back_to_ordinary assert_eq!(f.count("plugin_effect_receipts"), 0); f.shutdown().await; } + +#[tokio::test] +async fn active_remote_and_embedded_sends_close_on_cancel_revoke_permission_loss_and_owner_failure() +{ + use crate::plugin_https::test_support::Reply; + use server_api::PluginEffectObservation; + for remote in [false, true] { + for action in ["cancel", "revoke", "permission", "unavailable"] { + let mut peer = Peer::new(HOST, vec![Reply::Stall]).await; + let mut f = Fixture::new(Some(peer_binding(&peer))).await; + f.api.plugin_https_transport = Some(Arc::new(peer.transport_for_requests(4))); + let (remote_owner, server) = + remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await; + let owner = if remote { + remote_owner + } else { + f.owner(Some(peer.transport_for_requests(4))).await + }; + let host = f.host(owner.clone(), "active-interruption").await; + let id = host.context().invocation_id.clone(); + let request = send(&id, PluginHttpMethod::Post, "once"); + let mut recovered_api = f.api.clone(); + recovered_api.config_store = + Arc::new(SqliteWorkspaceStore::open_for_test(&f.api.config.database_path).unwrap()); + let attempt = owner.call(request.clone()); + tokio::pin!(attempt); + tokio::select! { + result = &mut attempt => panic!("request must reach the stalled peer: {result:?}"), + _ = peer.observed() => {}, + } + match action { + "cancel" => host.cancel().await.unwrap(), + "revoke" => f.revoke(), + "permission" => f.deny(), + "unavailable" => f + .api + .config_store + .with_conn(|conn| { + conn.execute_batch( + "ALTER TABLE plugin_execution_plans RENAME TO unavailable_plugin_plans", + )?; + Ok(()) + }) + .unwrap(), + _ => unreachable!(), + } + // No response is released. HTTP has a five-second timeout; authority + // loss must close the owner transport substantially before that. + let error = tokio::time::timeout(std::time::Duration::from_secs(2), &mut attempt) + .await + .expect("active transport did not stop") + .unwrap_err(); + assert_eq!( + error.effect, + PluginEffectObservation::Indeterminate, + "{remote}/{action}" + ); + peer.closed().await; + if action == "unavailable" { + f.api + .config_store + .with_conn(|conn| { + conn.execute_batch( + "ALTER TABLE unavailable_plugin_plans RENAME TO plugin_execution_plans", + )?; + Ok(()) + }) + .unwrap(); + } + if action == "permission" { + set_policy( + &f.api, + &f.subject, + &f.installed.installation.installation_id, + true, + ); + } + // Another actual owner with an independent SQLite connection has no + // local attempt state. Reopen after ordinary cancellation/revocation; + // the unavailable-owner fault deliberately changes SQLite DDL quotes. + if action != "unavailable" { + recovered_api.config_store = Arc::new( + SqliteWorkspaceStore::open_for_test(&f.api.config.database_path).unwrap(), + ); + } + let recovered = plugin_host::EmbeddedPluginHostOwner::new( + Arc::new(recovered_api), + f.verified_context().await, + TEST_WORKSPACE_ID.into(), + ) + .with_transport(peer.transport_for_requests(4)); + assert!(recovered.call(request).await.is_err()); + assert_eq!(f.count("plugin_effect_receipts"), 1); + assert_eq!( + peer.connection_count(), + 1, + "recovery resent {remote}/{action}" + ); + drop(server); + f.shutdown().await; + } + } +} + +#[tokio::test] +async fn binary_ceiling_matches_generated_remote_router_and_embedded_owner_limits() { + let ceiling = crate::plugin_binding::PluginConnectionPolicyCeiling::default().limits; + let body = vec![255; ceiling.max_request_bytes as usize]; + let mut peer = Peer::new( + HOST, + vec![response(200, &[], &body), response(200, &[], &body)], + ) + .await; + let mut binding = peer_binding(&peer); + binding.limits.max_request_bytes = ceiling.max_request_bytes; + binding.limits.max_response_bytes = ceiling.max_response_bytes; + let mut f = Fixture::new(Some(binding)).await; + f.api.plugin_https_transport = Some(Arc::new(peer.transport_for_requests(4))); + let embedded = f.owner(Some(peer.transport_for_requests(4))).await; + let (remote, server) = remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await; + let host = f.host(remote.clone(), "binary-boundary").await; + let id = &host.context().invocation_id; + for (index, owner) in [&embedded, &remote].into_iter().enumerate() { + let mut input = send(id, PluginHttpMethod::Post, &format!("binary-{index}")); + if let PluginHostRequest::Send { request, .. } = &mut input { + request.body = body.clone(); + } + assert!(serde_json::to_vec(&input).unwrap().len() > 4 * 1024 * 1024); + let PluginHostResponse::Response { response } = owner.call(input.clone()).await.unwrap() + else { + panic!("expected response") + }; + assert_eq!(response.status, 200); + assert_eq!(response.body.len(), body.len()); + assert!(response.body.iter().all(|byte| *byte == 255)); + let observed = peer.observed().await; + assert!(observed.ends_with(&body)); + assert_eq!(peer.connection_count(), index + 1); + // Replays cross adapters without a new connection. + embedded.call(input).await.unwrap(); + let mut oversized = send(id, PluginHttpMethod::Post, &format!("oversized-{index}")); + if let PluginHostRequest::Send { request, .. } = &mut oversized { + request.body = vec![255; body.len() + 1]; + } + let error = owner.call(oversized).await.unwrap_err(); + assert_eq!(error.code, PluginHostErrorCode::QuotaExceeded); + assert_eq!(peer.connection_count(), index + 1); + } + assert_eq!(f.count("plugin_effect_receipts"), 2); + let mut enormous = send(id, PluginHttpMethod::Post, "wire-overflow"); + if let PluginHostRequest::Send { request, .. } = &mut enormous { + request.body = vec![255; 2 * body.len()]; + } + let path = format!("/api/w/{TEST_WORKSPACE_ID}/plugin-host"); + let response = build_router(f.api.clone()) + .oneshot(runtime_source_request( + &f.identity, + Some(&f.source.worker_id), + "POST", + &path, + serde_json::to_vec(&enormous).unwrap(), + )) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE); + assert_eq!(peer.connection_count(), 2); + drop(server); + f.shutdown().await; +} + +#[tokio::test] +async fn other_workspace_control_plane_hosts_ips_and_dns_aliases_never_connect() { + for dynamic in [false, true] { + for host in [HOST, "8.8.8.8", "runtime-alias.example.com"] { + let peer = Peer::new(HOST, vec![]).await; + let mut binding = peer_binding(&peer); + binding.origin = format!("https://{host}:{}", peer.public_address().port()); + let mut f = Fixture::new(Some(binding)).await; + let endpoint = format!("https://{HOST}:{}", peer.public_address().port()); + if dynamic { + let mut workspace = test_workspace(); + workspace.workspace_id = "workspace-b".into(); + f.api.store.upsert_workspace(&workspace).await.unwrap(); + f.api + .store + .upsert_workspace_runtime_binding_record( + WorkspaceRuntimeBinding { + workspace_id: "workspace-b".into(), + runtime_id: "only-b".into(), + display_name: "Only B".into(), + base_url: endpoint, + public_key: f.identity.public_key.clone(), + public_key_fingerprint: String::new(), + binding_id: String::new(), + authentication_mode: + StoredRuntimeAuthenticationMode::LegacyServerIssuer, + created_at: "1".into(), + updated_at: "1".into(), + revoked_at: None, + }, + true, + ) + .await + .unwrap(); + } else { + let mut operational = f.api.config.clone(); + operational.remote_runtime_sources.push( + RemoteRuntimeConfig::new("only-b", "Only B", endpoint, None) + .with_workspace_id("workspace-b"), + ); + f.api.config = operational + .for_catalog_workspace(&test_workspace(), vec![]) + .unwrap(); + assert!( + !f.api + .config + .remote_runtime_sources + .iter() + .any(|source| source.runtime_id == "only-b") + ); + } + // Only external DNS/socket/trust are substituted. The operational + // inventory and concrete shared owner/adapters remain production code. + let policy = plugin_host::transport_policy(&f.api, |name, port| async move { + let ip = if name == HOST { "8.8.8.8" } else { "9.9.9.9" }; + Ok(vec![format!("{ip}:{port}").parse().unwrap()]) + }) + .await + .unwrap(); + assert!(policy.denied_hosts.contains(HOST)); + assert!( + policy + .denied_addresses + .contains(&"8.8.8.8".parse().unwrap()) + ); + f.api.plugin_https_transport = + Some(Arc::new(peer.transport_with_policy(policy.clone()))); + let embedded = f.owner(Some(peer.transport_with_policy(policy))).await; + let (remote, server) = + remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await; + let invocation = f.host(embedded.clone(), "forbidden-control-plane").await; + for owner in [&embedded, &remote] { + assert_denied( + owner + .call(send( + &invocation.context().invocation_id, + PluginHttpMethod::Get, + "blocked", + )) + .await, + ); + } + assert_eq!(peer.connection_count(), 0, "{dynamic}/{host}"); + drop(server); + f.shutdown().await; + } + } +} + +#[tokio::test] +async fn stalled_authority_monitor_closes_active_transport_and_consumes_effect() { + use crate::plugin_https::test_support::Reply; + for remote in [false, true] { + let mut peer = Peer::new(HOST, vec![Reply::Stall]).await; + let mut f = Fixture::new(Some(peer_binding(&peer))).await; + f.api.plugin_https_transport = Some(Arc::new(peer.transport_for_requests(4))); + let (remote_owner, server) = + remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await; + let owner = if remote { + remote_owner + } else { + f.owner(Some(peer.transport_for_requests(4))).await + }; + let host = f.host(owner.clone(), "monitor-unavailable").await; + let request = send( + &host.context().invocation_id, + PluginHttpMethod::Post, + "once", + ); + let attempt = owner.call(request.clone()); + tokio::pin!(attempt); + tokio::select! { + result = &mut attempt => panic!("request must reach peer: {result:?}"), + _ = peer.observed() => {}, + } + // Hold the actual authority DB connection. The monitor cannot complete + // its read, and may not treat a past permission as continued authority. + let store = f.api.config_store.clone(); + let (started, waiting) = tokio::sync::oneshot::channel(); + let (release, gate) = tokio::sync::oneshot::channel(); + let held = tokio::task::spawn_blocking(move || { + store.with_conn(|_| { + let _ = started.send(()); + let _ = gate.blocking_recv(); + Ok(()) + }) + }); + waiting.await.unwrap(); + tokio::select! { + result = &mut attempt => panic!("DB is held but owner finished: {result:?}"), + _ = peer.closed() => {}, + } + release.send(()).unwrap(); + held.await.unwrap().unwrap(); + let error = tokio::time::timeout(std::time::Duration::from_secs(3), &mut attempt) + .await + .unwrap() + .unwrap_err(); + assert_eq!( + error.effect, + server_api::PluginEffectObservation::Indeterminate + ); + assert_eq!(error.code, PluginHostErrorCode::Unavailable); + assert!(owner.call(request).await.is_err()); + assert_eq!(f.count("plugin_effect_receipts"), 1); + assert_eq!(peer.connection_count(), 1); + drop(server); + f.shutdown().await; + } +} diff --git a/crates/workspace-server/src/worker_source/runtime_request.rs b/crates/workspace-server/src/worker_source/runtime_request.rs index ab1a99bef..7f6b441d4 100644 --- a/crates/workspace-server/src/worker_source/runtime_request.rs +++ b/crates/workspace-server/src/worker_source/runtime_request.rs @@ -454,6 +454,7 @@ mod tests { runtime_event_sources: vec![], remote_runtime_sources: vec![], backend_base_url: Some("server-auth".into()), + control_plane_endpoints: Default::default(), }; let identity = RuntimeIdentityMaterial::generate("runtime-auth").unwrap(); store