Repository navigation
revert(daemon): dispatch broker requests on the connection task again #4078
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -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>( | ||
| mut invocation: std::result::Result<DaemonInvocationRequest, DaemonInvocationResponse>, | ||
| transport: &mut BrokerStreamTransport, | ||
| service: &DaemonInvocationService, | ||
|
|
@@ -865,7 +865,7 @@ async fn drive_retained_invocation_responses( | |
| mut execute: impl FnMut( | ||
| DaemonInvocationRequest, | ||
| ) -> std::pin::Pin< | ||
| Box<dyn std::future::Future<Output = DaemonInvocationResponse> + 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 { | ||
| Ok(request) => execute(request).await, | ||
|
Contributor
There was a problem hiding this comment. 🔴 Writer waits stall broker workers When Learn moreThe foreground broker runs each connection on a native worker. Synchronous storage ports reach recv_writer_reply, which calls Example: If all native workers dispatch requests whose synchronous ports wait on a busy writer, none can poll another accepted socket or timer until a writer reply arrives. Recommended fix: Keep Was this helpful? React with 👍 or 👎 to provide feedback. |
||
| 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| 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| { | ||
| 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(()) | ||
| } | ||
|
|
||
| /// 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.parked-dispatch"; | ||
|
|
||
| fn handshake() -> DaemonHandshake { | ||
| 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.parked-dispatch".to_owned(), | ||
| tool_list_changed_capable: false, | ||
| catalog_version: String::new(), | ||
| moved_store_adoption: MovedStoreAdoption::Never, | ||
|
|
@@ -2233,22 +2168,29 @@ mod long_dispatch_tests { | |
| 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()), | ||
|
|
@@ -2258,16 +2200,14 @@ mod long_dispatch_tests { | |
| &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, | ||
|
|
@@ -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 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::{ | ||
| 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) | ||
| ); | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🔴 Long dispatch drops broker responses
When
execute(request)uses over 250 ms of CPU in one poll,LongPollGuardstill panics in debug builds. The connection closes without answering the request.Learn more
Broker connections run as native tasks. LongPollGuard still asserts when a native task poll exceeds
LONG_POLL_CPU(250 ms). Awaiting dispatch directly makes its synchronous CPU work part of the connection task's poll again. The removed long-dispatch test exercised exactly this case; the assertion prevents the response from being written.Example: A broker request spends 300 ms of thread CPU inside dispatch before its next await. The native task panics at 250 ms and the waiting client receives a closed socket instead of its response.
Recommended fix: Preserve connection-task admission scope without subjecting synchronous dispatch to the fatal native-poll guard. Change the guard to nonfatal counting as planned, or isolate CPU-heavy work while retaining the admission lease, and retain a regression test for long dispatch.
Was this helpful? React with 👍 or 👎 to provide feedback.