diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index cff8d7bf5e..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; @@ -2139,54 +2112,12 @@ pub(super) async fn serve_windows_broker_client_with_class_and_invocation( Ok(()) } -#[cfg(test)] -mod delivery_ack_tests { - use super::{ - DaemonDeliveryAckWait, await_daemon_delivery_ack, classify_daemon_delivery_ack_wait, - }; - use std::time::Duration; - use tracedecay_mcp::transport::ChannelTransport; - use tracedecay_runtime_core::native_runtime; - - /// Runs outside any Tokio runtime: the deadline must be a native timer. - #[test] - fn delivery_ack_wait_uses_the_exact_deadline_budget() { - const BUDGET: Duration = Duration::from_millis(100); - let (mut transport, _input, _output) = ChannelTransport::new(); - let started = std::time::Instant::now(); - let wait = native_runtime::block_on(native_runtime::timeout( - Duration::from_secs(10), - await_daemon_delivery_ack(&mut transport, BUDGET, None, std::future::pending::<()>()), - )) - .expect("native runtime runs the ack wait") - .expect("the ack wait ends at its deadline"); - let waited = started.elapsed(); - - assert!(matches!(wait, Ok(DaemonDeliveryAckWait::Deadline))); - assert!( - waited >= BUDGET, - "the ack wait must spend its whole budget, waited {waited:?}" - ); - } - - #[test] - fn withheld_ack_terminalizes_as_deadline_drop() { - assert_eq!( - classify_daemon_delivery_ack_wait(DaemonDeliveryAckWait::Deadline), - Err(tracedecay_domain::DeliveryDropReasonV1::Deadline) - ); - assert_eq!( - classify_daemon_delivery_ack_wait(DaemonDeliveryAckWait::Cancelled), - Err(tracedecay_domain::DeliveryDropReasonV1::Cancelled) - ); - } -} - -/// 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 { +/// 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; @@ -2201,11 +2132,15 @@ mod long_dispatch_tests { 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.long-dispatch"; + const REQUEST_ID: &str = "request.parked-dispatch"; fn handshake() -> DaemonHandshake { - let profile_root = PathBuf::from("/profiles/long-dispatch"); + let profile_root = PathBuf::from("/profiles/parked-dispatch"); DaemonHandshake { project_path: None, scope_prefix: None, @@ -2217,7 +2152,7 @@ mod long_dispatch_tests { profile_root, }, client_version: String::new(), - client_instance_id: "client.long-dispatch".to_owned(), + client_instance_id: "client.parked-dispatch".to_owned(), tool_list_changed_capable: false, catalog_version: String::new(), moved_store_adoption: MovedStoreAdoption::Never, @@ -2233,13 +2168,19 @@ mod long_dispatch_tests { now, tracedecay_contracts::Deadline::new(tracedecay_domain::UtcMicros(now.0 + 30_000_000)) .expect("deadline"), - tracedecay_contracts::CancellationContext::active("cancel.long-dispatch") + tracedecay_contracts::CancellationContext::active("cancel.parked-dispatch") .expect("cancellation"), ) } #[test] - fn a_dispatch_longer_than_one_native_poll_still_answers_its_request() { + 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) @@ -2247,8 +2188,9 @@ mod long_dispatch_tests { 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(async move { + let connection = NativeTask::spawn(with_connection_admission(permit, async move { let mut transport = BrokerStreamTransport::new(daemon_end); drive_retained_invocation_responses( Ok(request()), @@ -2258,16 +2200,14 @@ mod long_dispatch_tests { &DaemonLifecycle::default(), &mut HashMap::new(), |request| { + let mut lifted = lifted.clone(); 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; + park_admission(async { + while !*lifted.borrow_and_update() { + lifted.changed().await.expect("barrier sender retained"); + } + }) + .await; DaemonInvocationResponse::problem( request.request_id, DaemonInvocationProblem::Unavailable, @@ -2277,9 +2217,16 @@ mod long_dispatch_tests { ) .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"); @@ -2288,15 +2235,66 @@ mod long_dispatch_tests { .read_line(&mut line) .expect("read the response"); let response: DaemonInvocationResponse = - serde_json::from_str(&line).expect("the connection answers the sent request"); + 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(())), - "the connection ends cleanly when its client closes" + 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::{ + DaemonDeliveryAckWait, await_daemon_delivery_ack, classify_daemon_delivery_ack_wait, + }; + use std::time::Duration; + use tracedecay_mcp::transport::ChannelTransport; + use tracedecay_runtime_core::native_runtime; + + /// Runs outside any Tokio runtime: the deadline must be a native timer. + #[test] + fn delivery_ack_wait_uses_the_exact_deadline_budget() { + const BUDGET: Duration = Duration::from_millis(100); + let (mut transport, _input, _output) = ChannelTransport::new(); + let started = std::time::Instant::now(); + let wait = native_runtime::block_on(native_runtime::timeout( + Duration::from_secs(10), + await_daemon_delivery_ack(&mut transport, BUDGET, None, std::future::pending::<()>()), + )) + .expect("native runtime runs the ack wait") + .expect("the ack wait ends at its deadline"); + let waited = started.elapsed(); + + assert!(matches!(wait, Ok(DaemonDeliveryAckWait::Deadline))); + assert!( + waited >= BUDGET, + "the ack wait must spend its whole budget, waited {waited:?}" + ); + } + + #[test] + fn withheld_ack_terminalizes_as_deadline_drop() { + assert_eq!( + classify_daemon_delivery_ack_wait(DaemonDeliveryAckWait::Deadline), + Err(tracedecay_domain::DeliveryDropReasonV1::Deadline) + ); + assert_eq!( + classify_daemon_delivery_ack_wait(DaemonDeliveryAckWait::Cancelled), + Err(tracedecay_domain::DeliveryDropReasonV1::Cancelled) ); } }