From 071371e7329107d16d02f7825a9311bfb693ccc5 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:04:49 +0000 Subject: [PATCH 1/2] test(daemon): park a broker dispatch and expect its admission slot back --- .../src/daemon/connection_serving.rs | 144 ++++++++++++++++++ 1 file changed, 144 insertions(+) diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index cff8d7bf5e..7f9cca00e5 100644 --- a/crates/tracedecay/src/daemon/connection_serving.rs +++ b/crates/tracedecay/src/daemon/connection_serving.rs @@ -2139,6 +2139,150 @@ pub(super) async fn serve_windows_broker_client_with_class_and_invocation( Ok(()) } +/// A request parked on a barrier gives its connection's admission slot back +/// through `park_admission`, which reads the connection's task-local lease. +/// Dispatch must run inside the connection task's scope for that read to +/// succeed; polled on another thread, the park keeps the slot. +#[cfg(all(test, unix))] +mod broker_dispatch_admission_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}; + use crate::daemon::{ + DaemonClientAdmission, DaemonClientAdmissionOutcome, park_admission, + with_connection_admission, + }; + + const REQUEST_ID: &str = "request.parked-dispatch"; + + fn handshake() -> DaemonHandshake { + let profile_root = PathBuf::from("/profiles/parked-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.parked-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.parked-dispatch") + .expect("cancellation"), + ) + } + + #[test] + fn a_dispatch_parked_on_a_barrier_releases_its_admission_slot() { + let admission = DaemonClientAdmission::with_reserved_capacity(2, 1); + let DaemonClientAdmissionOutcome::Admitted(permit) = admission.try_admit() else { + panic!("the first client is admitted"); + }; + assert_eq!(admission.available_general_permits(), 0); + + 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 (barrier, lifted) = tokio::sync::watch::channel(false); + let invocation = DaemonInvocationState::default(); + let connection = NativeTask::spawn(with_connection_admission(permit, 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| { + let mut lifted = lifted.clone(); + Box::pin(async move { + park_admission(async { + while !*lifted.borrow_and_update() { + lifted.changed().await.expect("barrier sender retained"); + } + }) + .await; + DaemonInvocationResponse::problem( + request.request_id, + DaemonInvocationProblem::Unavailable, + ) + }) + }, + ) + .await + .map_err(|error| error.to_string()) + })) + .expect("native connection task"); + + let parked_by = Instant::now() + Duration::from_secs(10); + while admission.available_general_permits() == 0 && Instant::now() < parked_by { + std::thread::sleep(Duration::from_millis(10)); + } + let released = admission.available_general_permits(); + barrier.send(true).expect("lift the barrier"); + + 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 parked 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(())) + ); + assert_eq!( + released, 1, + "the parked dispatch kept its connection's admission slot" + ); + assert_eq!( + admission.available_general_permits(), + 1, + "the connection returns its slot when it ends" + ); + } +} + #[cfg(test)] mod delivery_ack_tests { use super::{ From 76431040ae33e45f39a8551536afaea8c8a44313 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 03:04:54 +0000 Subject: [PATCH 2/2] revert(daemon): dispatch broker requests on the connection task again This reverts merge ab27bb6d5f (#4065). It polled each broker request on a blocking-pool thread through native_runtime::off_native_thread. That thread runs outside the connection task's task-local admission scopes, so park_admission could not find the lease and a request parked on a project open, the writer gate, or a generation decode kept its admission slot. The hop also moved whole handler futures onto pool threads with smaller stacks, lost disconnect cancellation, and measured no latency gain. The long-poll failure it worked around is gone: long native polls are now counted, not failed. --- .../src/daemon/connection_serving.rs | 176 ++---------------- 1 file changed, 15 insertions(+), 161 deletions(-) diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index 7f9cca00e5..58b1d2e768 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( +async fn drive_retained_invocation_responses<'a>( mut invocation: std::result::Result, transport: &mut BrokerStreamTransport, service: &DaemonInvocationService, @@ -865,7 +865,7 @@ async fn drive_retained_invocation_responses( mut execute: impl FnMut( DaemonInvocationRequest, ) -> std::pin::Pin< - Box + Send + 'static>, + Box + Send + 'a>, >, ) -> Result<()> { let mut pending_line = None; @@ -888,19 +888,7 @@ async fn drive_retained_invocation_responses( .ok() .and_then(invocation_lsp_session_transition); let response = match invocation { - // 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, - ) - } - }, + Ok(request) => execute(request).await, Err(response) => response, }; update_connection_lsp_sessions(owned_lsp_sessions, session_transition.as_ref(), &response); @@ -1084,11 +1072,7 @@ async fn serve_retained_invocation_connection( &handshake, &lifecycle, &mut owned_lsp_sessions, - |request| { - let engine = engine.clone(); - let handshake = handshake.clone(); - Box::pin(async move { execute_daemon_invocation(&engine, &handshake, request).await }) - }, + |request| Box::pin(execute_daemon_invocation(&engine, &handshake, request)), )) .await; cleanup_connection_lsp_sessions(&engine.invocation, owned_lsp_sessions).await; @@ -1930,28 +1914,17 @@ pub(super) async fn serve_windows_broker_client_with_class_and_invocation( &lifecycle, &mut owned_lsp_sessions, |request| { - 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 - }) + 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(), + )) }, )) .await; @@ -2325,122 +2298,3 @@ 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(); - } - // 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; - 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" - ); - } -}