fix(daemon): dispatch broker requests off native workers

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<DaemonInvocationRequest, DaemonInvocationResponse>,
     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<dyn std::future::Future<Output = DaemonInvocationResponse> + Send + 'a>,
+        Box<dyn std::future::Future<Output = DaemonInvocationResponse> + 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;
@@ -2154,3 +2181,122 @@ 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"
+        );
+    }
+}