From 618f9cb4a439f25d82af4d6dfc45a102b9501379 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 22:33:33 +0000 Subject: [PATCH 1/3] test(daemon): answer a broker request whose dispatch outlasts a poll A broker connection whose request dispatch spends longer than LONG_POLL_CPU in one native poll trips the debug long-poll guard. The guard fails the connection task, so the socket closes after the request was sent and the client sees no response. The test fails until dispatch leaves the native poll (#4056). --- .../src/daemon/connection_serving.rs | 117 ++++++++++++++++++ 1 file changed, 117 insertions(+) diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index 6342d83a6c..944cfa1c94 100644 --- a/crates/tracedecay/src/daemon/connection_serving.rs +++ b/crates/tracedecay/src/daemon/connection_serving.rs @@ -2154,3 +2154,120 @@ mod delivery_ack_tests { ); } } + +/// Debug builds fail a native task whose poll runs longer than +/// `LONG_POLL_CPU`; a broker connection's request dispatch must not be that +/// poll, or the daemon closes the socket after the request was sent (#4056). +#[cfg(all(test, unix, debug_assertions))] +mod long_dispatch_tests { + use std::collections::HashMap; + use std::io::{BufRead, BufReader}; + use std::path::PathBuf; + use std::time::{Duration, Instant}; + + use tracedecay_daemon_protocol::{ + BrokerStream, DaemonClientIdentity, DaemonHandshake, DaemonInvocationProblem, + DaemonInvocationRequest, DaemonInvocationResponse, MovedStoreAdoption, + }; + use tracedecay_daemon_service::shutdown::DaemonLifecycle; + use tracedecay_mcp::BrokerStreamTransport; + use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome}; + + use super::{DaemonInvocationState, drive_retained_invocation_responses}; + + const REQUEST_ID: &str = "request.long-dispatch"; + + fn handshake() -> DaemonHandshake { + let profile_root = PathBuf::from("/profiles/long-dispatch"); + DaemonHandshake { + project_path: None, + scope_prefix: None, + timings: false, + allow_init: false, + allow_initialize_root_routing: false, + client_identity: DaemonClientIdentity { + global_db_path: profile_root.join("global.db"), + profile_root, + }, + client_version: String::new(), + client_instance_id: "client.long-dispatch".to_owned(), + tool_list_changed_capable: false, + catalog_version: String::new(), + moved_store_adoption: MovedStoreAdoption::Never, + } + } + + fn request() -> DaemonInvocationRequest { + let now = tracedecay_contracts::now_micros(); + DaemonInvocationRequest::profile_graph_tool( + REQUEST_ID, + tracedecay_tool_catalog::ApplicationSurfaceOperation::AffectedTests, + serde_json::Map::new(), + now, + tracedecay_contracts::Deadline::new(tracedecay_domain::UtcMicros(now.0 + 30_000_000)) + .expect("deadline"), + tracedecay_contracts::CancellationContext::active("cancel.long-dispatch") + .expect("cancellation"), + ) + } + + #[test] + fn a_dispatch_longer_than_one_native_poll_still_answers_its_request() { + let (daemon_end, client) = std::os::unix::net::UnixStream::pair().expect("socket pair"); + daemon_end + .set_nonblocking(true) + .expect("nonblocking daemon end"); + let daemon_end = BrokerStream::Unix( + asupersync::net::unix::UnixStream::from_std(daemon_end).expect("daemon end"), + ); + let invocation = DaemonInvocationState::default(); + let connection = NativeTask::spawn(async move { + let mut transport = BrokerStreamTransport::new(daemon_end); + drive_retained_invocation_responses( + Ok(request()), + &mut transport, + &invocation.service, + &handshake(), + &DaemonLifecycle::default(), + &mut HashMap::new(), + |request| { + Box::pin(async move { + let spun = Instant::now(); + while spun.elapsed() < native_runtime::LONG_POLL_CPU * 2 { + std::hint::spin_loop(); + } + let _ = + native_runtime::sleep_until(Instant::now() + Duration::from_millis(1)) + .await; + DaemonInvocationResponse::problem( + request.request_id, + DaemonInvocationProblem::Unavailable, + ) + }) + }, + ) + .await + .map_err(|error| error.to_string()) + }) + .expect("native connection task"); + + client + .set_read_timeout(Some(Duration::from_secs(30))) + .expect("client read timeout"); + let mut line = String::new(); + BufReader::new(&client) + .read_line(&mut line) + .expect("read the response"); + let response: DaemonInvocationResponse = + serde_json::from_str(&line).expect("the connection answers the sent request"); + assert_eq!(response.request_id, REQUEST_ID); + client + .shutdown(std::net::Shutdown::Both) + .expect("close the client"); + assert_eq!( + native_runtime::block_on(connection.wait()).expect("process runtime"), + NativeTaskOutcome::Completed(Ok(())), + "the connection ends cleanly when its client closes" + ); + } +} From 3b0fe06fa733732d7e25ccdfafe2334ccc382a1a Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 22:40:34 +0000 Subject: [PATCH 2/3] fix(daemon): dispatch broker requests off native workers A broker connection ran each request's dispatch inside its own native task poll. Under host load that poll spent hundreds of milliseconds of thread CPU on routing and storage work, and in debug builds the long-poll guard failed the connection task after the request was sent. The daemon's HTTP surface reaches the daemon through this socket, so a Workflow call answered workflow.transport_unavailable (#4056). The retained invocation loop now runs every request's dispatch through native_runtime::off_native_thread, the helper #4052 applied to Work and Workflow, so the Unix and portable brokers keep their native polls to framing and socket I/O. Fixes #4056 --- .../src/daemon/connection_serving.rs | 57 ++++++++++++++----- 1 file changed, 42 insertions(+), 15 deletions(-) diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index 944cfa1c94..2f8c208ae8 100644 --- a/crates/tracedecay/src/daemon/connection_serving.rs +++ b/crates/tracedecay/src/daemon/connection_serving.rs @@ -855,7 +855,7 @@ async fn serve_broker_socket_client( clippy::too_many_lines, reason = "Delivery ACK settlement is one state machine shared by the Unix and portable brokers." )] -async fn drive_retained_invocation_responses<'a>( +async fn drive_retained_invocation_responses( mut invocation: std::result::Result, transport: &mut BrokerStreamTransport, service: &DaemonInvocationService, @@ -865,7 +865,7 @@ async fn drive_retained_invocation_responses<'a>( mut execute: impl FnMut( DaemonInvocationRequest, ) -> std::pin::Pin< - Box + Send + 'a>, + Box + Send + 'static>, >, ) -> Result<()> { let mut pending_line = None; @@ -888,7 +888,19 @@ async fn drive_retained_invocation_responses<'a>( .ok() .and_then(invocation_lsp_session_transition); let response = match invocation { - Ok(request) => execute(request).await, + // Dispatch runs synchronous routing and storage work. On a native + // worker that work holds the reactor (#3994), and in debug builds + // a long poll fails this connection after the request was sent. + Ok(request) => match native_runtime::off_native_thread(execute(request)).await { + Ok(response) => response, + Err(error) => { + tracing::error!(%error, "daemon request did not run on the blocking pool"); + DaemonInvocationResponse::problem( + request_id.clone().unwrap_or_default(), + tracedecay_daemon_protocol::DaemonInvocationProblem::Unavailable, + ) + } + }, Err(response) => response, }; update_connection_lsp_sessions(owned_lsp_sessions, session_transition.as_ref(), &response); @@ -1072,7 +1084,11 @@ async fn serve_retained_invocation_connection( &handshake, &lifecycle, &mut owned_lsp_sessions, - |request| Box::pin(execute_daemon_invocation(&engine, &handshake, request)), + |request| { + let engine = engine.clone(); + let handshake = handshake.clone(); + Box::pin(async move { execute_daemon_invocation(&engine, &handshake, request).await }) + }, )) .await; cleanup_connection_lsp_sessions(&engine.invocation, owned_lsp_sessions).await; @@ -1914,17 +1930,28 @@ pub(super) async fn serve_windows_broker_client_with_class_and_invocation( &lifecycle, &mut owned_lsp_sessions, |request| { - Box::pin(execute_portable_daemon_invocation( - lifecycle.clone(), - store_administration.clone(), - Arc::clone(&project_open_gates), - &handshake, - &invocation, - http_application_registry.clone(), - request, - #[cfg(test)] - project_open_attempts.clone(), - )) + let lifecycle = lifecycle.clone(); + let store_administration = store_administration.clone(); + let project_open_gates = Arc::clone(&project_open_gates); + let handshake = handshake.clone(); + let invocation = invocation.clone(); + let http_application_registry = http_application_registry.clone(); + #[cfg(test)] + let project_open_attempts = project_open_attempts.clone(); + Box::pin(async move { + execute_portable_daemon_invocation( + lifecycle, + store_administration, + project_open_gates, + &handshake, + &invocation, + http_application_registry, + request, + #[cfg(test)] + project_open_attempts, + ) + .await + }) }, )) .await; From 23213559c95bfed16858ffdc4da45b4de7d4a50c Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 23:24:44 +0000 Subject: [PATCH 3/3] test(daemon): say why the long dispatch awaits after spinning --- crates/tracedecay/src/daemon/connection_serving.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index 2f8c208ae8..cff8d7bf5e 100644 --- a/crates/tracedecay/src/daemon/connection_serving.rs +++ b/crates/tracedecay/src/daemon/connection_serving.rs @@ -2263,6 +2263,8 @@ mod long_dispatch_tests { while spun.elapsed() < native_runtime::LONG_POLL_CPU * 2 { std::hint::spin_loop(); } + // Await again after the synchronous work, so the + // response is not written inside the long poll. let _ = native_runtime::sleep_until(Instant::now() + Duration::from_millis(1)) .await;