From 6f11bdc87c53eb2a829a649735c0b8052688262e Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 23:10:38 +0000 Subject: [PATCH 1/3] refactor(dashboard): serve the hosted dashboard natively serve_loopback takes a ConnectionExecutor. The hosted dashboard runs its serve loop and every connection on the native runtime; the daemon's HTTP application listener keeps Tokio connections, because /remote-deletions reaches project_server_lifecycle inside its connection task. The hosted dashboard task becomes a NativeTask with the same stop, abort at deadline, and abort-on-drop handling. --- crates/tracedecay-dashboard-api/src/lib.rs | 5 +- crates/tracedecay-dashboard-api/src/server.rs | 88 +++++++++++++++---- .../src/server_tests.rs | 33 ++++--- .../src/test_transport.rs | 9 +- .../tracedecay/src/daemon/http_application.rs | 13 ++- .../src/mcp/tools/handlers/dashboard.rs | 57 +++++++----- .../src/mcp/tools/handlers/dashboard/tests.rs | 21 +++-- 7 files changed, 156 insertions(+), 70 deletions(-) diff --git a/crates/tracedecay-dashboard-api/src/lib.rs b/crates/tracedecay-dashboard-api/src/lib.rs index 89f83c9648b..425ac4ab49b 100644 --- a/crates/tracedecay-dashboard-api/src/lib.rs +++ b/crates/tracedecay-dashboard-api/src/lib.rs @@ -145,7 +145,10 @@ pub use http_admission::{ with_dashboard_http_admission, }; pub use routes::router; -pub use server::{bind_dashboard, serve_http_connection, serve_loopback, validate_dashboard_host}; +pub use server::{ + ConnectionExecutor, bind_dashboard, serve_http_connection, serve_loopback, + validate_dashboard_host, +}; pub use state::{ AdmittedDoctorReportV1, AutomationSchedulerReconcileFuture, AutomationSchedulerReconciler, DashboardAutomationObservationFuture, DashboardAutomationObservationRecorderV1, DashboardState, diff --git a/crates/tracedecay-dashboard-api/src/server.rs b/crates/tracedecay-dashboard-api/src/server.rs index 3a1d2f7c8e7..1b2438f0093 100644 --- a/crates/tracedecay-dashboard-api/src/server.rs +++ b/crates/tracedecay-dashboard-api/src/server.rs @@ -16,6 +16,7 @@ use hyper_util::service::TowerToHyperService; use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use tower::ServiceExt; +use tracedecay_runtime_core::native_runtime::{NativeTaskOutcome, NativeTaskSet}; use crate::state::config_error; use tracedecay_domain::errors::Result; @@ -50,33 +51,29 @@ pub fn validate_dashboard_host(host: &str) -> Result<&str> { /// Matches `axum::serve(..).with_graceful_shutdown(..)` on an Asupersync /// socket. The drain has no deadline because callers own one. Accept errors /// never end the loop. Dropping the future aborts every connection. -/// Connections run on `runtime`, because daemon routes served here still -/// reach Tokio. pub async fn serve_loopback( - runtime: tokio::runtime::Handle, + executor: ConnectionExecutor, listener: TcpListener, router: Router, shutdown: impl Future, ) { let graceful = CancellationToken::new(); - let mut connections = JoinSet::new(); + let mut connections = Connections::new(executor); let mut shutdown = std::pin::pin!(shutdown); loop { tokio::select! { biased; () = &mut shutdown => break, - Some(joined) = connections.join_next(), if !connections.is_empty() => { - observe_connection_join(joined); - } + Some(()) = connections.join_next(), if !connections.is_empty() => {} accepted = listener.accept() => match accepted { Ok((stream, address)) => { let router = router.clone(); let graceful = graceful.clone(); - connections.spawn_on(async move { + connections.spawn(async move { if let Err(error) = serve_http_connection(stream, router, true, graceful).await { tracing::trace!(%error, %address, "loopback HTTP connection stopped"); } - }, &runtime); + }); } Err(error) => wait_after_accept_error(&error).await, }, @@ -84,8 +81,69 @@ pub async fn serve_loopback( } drop(listener); graceful.cancel(); - while let Some(joined) = connections.join_next().await { - observe_connection_join(joined); + while connections.join_next().await.is_some() {} +} + +/// Where [`serve_loopback`] runs each connection. +pub enum ConnectionExecutor { + Native, + /// For daemon routes that still reach `project_server_lifecycle` inside + /// their connection task. + Tokio(tokio::runtime::Handle), +} + +enum Connections { + Native(NativeTaskSet<()>), + Tokio(tokio::runtime::Handle, JoinSet<()>), +} + +impl Connections { + fn new(executor: ConnectionExecutor) -> Self { + match executor { + ConnectionExecutor::Native => Self::Native(NativeTaskSet::default()), + ConnectionExecutor::Tokio(runtime) => Self::Tokio(runtime, JoinSet::new()), + } + } + + /// A refused connection's IO drops with it, closing the socket. + fn spawn(&mut self, connection: impl Future + Send + 'static) { + match self { + Self::Native(set) => { + if let Err(error) = set.spawn(connection) { + tracing::warn!(%error, "loopback HTTP connection not admitted"); + } + } + Self::Tokio(runtime, set) => { + set.spawn_on(connection, runtime); + } + } + } + + fn is_empty(&self) -> bool { + match self { + Self::Native(set) => set.is_empty(), + Self::Tokio(_, set) => set.is_empty(), + } + } + + /// Joins the next connection to end, logging a failed one; `None` once + /// none remain. + async fn join_next(&mut self) -> Option<()> { + match self { + Self::Native(set) => { + if let NativeTaskOutcome::Panicked(payload) = set.join_next().await? { + tracing::warn!(?payload, "loopback HTTP connection task failed"); + } + } + Self::Tokio(_, set) => { + if let Err(error) = set.join_next().await? + && !error.is_cancelled() + { + tracing::warn!(%error, "loopback HTTP connection task failed"); + } + } + } + Some(()) } } @@ -104,14 +162,6 @@ async fn wait_after_accept_error(error: &io::Error) { Sleep::after(wall_now(), Duration::from_secs(1)).await; } -fn observe_connection_join(joined: std::result::Result<(), tokio::task::JoinError>) { - if let Err(error) = joined - && !error.is_cancelled() - { - tracing::warn!(%error, "loopback HTTP connection task failed"); - } -} - /// Serves one HTTP/1.1 connection with `router` and starts a graceful /// shutdown once `graceful` is cancelled. hyper runs without an executor or a /// timer, so the IO is the only runtime seam. diff --git a/crates/tracedecay-dashboard-api/src/server_tests.rs b/crates/tracedecay-dashboard-api/src/server_tests.rs index 97be68dafba..5fdc64dcf10 100644 --- a/crates/tracedecay-dashboard-api/src/server_tests.rs +++ b/crates/tracedecay-dashboard-api/src/server_tests.rs @@ -8,7 +8,7 @@ use tokio::sync::{Notify, oneshot}; use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome}; -use crate::server::{bind_dashboard, serve_loopback}; +use crate::server::{ConnectionExecutor, bind_dashboard, serve_loopback}; fn get_over_std_socket(addr: SocketAddr, path: &str) -> String { let mut stream = TcpStream::connect(addr).expect("connect loopback server"); @@ -51,9 +51,14 @@ fn loopback_server_drains_in_flight_request_after_shutdown_without_tokio_drivers .await .expect("bind loopback listener"); let (stop, stopped) = oneshot::channel::<()>(); - let server = tokio::spawn(serve_loopback(runtime.clone(), listener, router, async { - let _ = stopped.await; - })); + let server = tokio::spawn(serve_loopback( + ConnectionExecutor::Tokio(runtime.clone()), + listener, + router, + async { + let _ = stopped.await; + }, + )); let client = tokio::task::spawn_blocking(move || get_over_std_socket(addr, "/slow")); entered.notified().await; @@ -80,21 +85,23 @@ fn loopback_server_drains_in_flight_request_after_shutdown_without_tokio_drivers }); } -/// A native caller hands the serve loop its Tokio handle; the loop needs no -/// ambient Tokio runtime to serve. +/// The hosted dashboard serves every connection natively, with no Tokio +/// runtime in the process path. #[test] -fn loopback_server_serves_from_a_native_task_with_an_explicit_handle() { - let tokio = tokio::runtime::Builder::new_multi_thread() - .worker_threads(1) - .build() - .expect("Tokio runtime"); - let router = Router::new().route("/ping", get(|| async { "pong" })); +fn loopback_server_serves_native_connections_without_tokio() { + let router = Router::new().route( + "/ping", + get(|| async { + assert!(tokio::runtime::Handle::try_current().is_err()); + "pong" + }), + ); let (listener, addr) = native_runtime::block_on(bind_dashboard("127.0.0.1", 0)) .expect("native root") .expect("bind loopback listener"); let (stop, stopped) = oneshot::channel::<()>(); let server = NativeTask::spawn(serve_loopback( - tokio.handle().clone(), + ConnectionExecutor::Native, listener, router, async { diff --git a/crates/tracedecay-dashboard-api/src/test_transport.rs b/crates/tracedecay-dashboard-api/src/test_transport.rs index a563de19913..28471f76c92 100644 --- a/crates/tracedecay-dashboard-api/src/test_transport.rs +++ b/crates/tracedecay-dashboard-api/src/test_transport.rs @@ -9,9 +9,10 @@ use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use crate::state::build_state_inner; use crate::{ - DashboardAccessToken, DashboardApplicationRuntime, DashboardAutomationAuthorityV1, - DashboardAutomationWriter, DashboardDeliveryReadPortV1, DashboardGitCorrelationReadPortV1, - DashboardLcmReadPortV1, DashboardProfileCodeIndexWorkerSettingsPort, DashboardProjectContext, + ConnectionExecutor, DashboardAccessToken, DashboardApplicationRuntime, + DashboardAutomationAuthorityV1, DashboardAutomationWriter, DashboardDeliveryReadPortV1, + DashboardGitCorrelationReadPortV1, DashboardLcmReadPortV1, + DashboardProfileCodeIndexWorkerSettingsPort, DashboardProjectContext, DashboardSessionAuthoritiesV1, DashboardSessionMountV1, DashboardSessionResolutionV1, DashboardStateCompositionV1, PrAutoTrackManagedSummaryReader, bind_dashboard, code_read_api, router, serve_loopback, standalone_dashboard_automation_writer, with_dashboard_http_admission, @@ -318,6 +319,6 @@ where .ok_or_else(|| { crate::state::config_error("the test dashboard found no Tokio runtime") })?; - serve_loopback(runtime, listener, app, shutdown).await; + serve_loopback(ConnectionExecutor::Tokio(runtime), listener, app, shutdown).await; Ok(()) } diff --git a/crates/tracedecay/src/daemon/http_application.rs b/crates/tracedecay/src/daemon/http_application.rs index 7b8e82733d0..12ea6b8106d 100644 --- a/crates/tracedecay/src/daemon/http_application.rs +++ b/crates/tracedecay/src/daemon/http_application.rs @@ -43,7 +43,7 @@ use tracedecay_daemon_service::remote_http_transport::{ RemoteBrainTlsDeadlines, RemoteBrainTlsEgressObserver, RemoteBrainTlsEgressSnapshot, RemoteBrainTlsIngressSnapshot, }; -use tracedecay_dashboard_api::serve_loopback; +use tracedecay_dashboard_api::{ConnectionExecutor, serve_loopback}; use tracedecay_domain::{BrainNodeId, EnrollmentGrantV1, ProjectId}; use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; use tracedecay_rusqlite_runtime::remote::RemoteWriterPublicationErrorV1; @@ -901,9 +901,14 @@ impl DaemonHttpApplicationService { let serve_runtime = runtime.clone(); let task = runtime.spawn(tracing::Instrument::instrument( async move { - serve_loopback(serve_runtime, listener, app, async { - let _ = shutdown_requested.await; - }) + serve_loopback( + ConnectionExecutor::Tokio(serve_runtime), + listener, + app, + async { + let _ = shutdown_requested.await; + }, + ) .await; task_active.store(false, Ordering::Release); }, diff --git a/crates/tracedecay/src/mcp/tools/handlers/dashboard.rs b/crates/tracedecay/src/mcp/tools/handlers/dashboard.rs index 8ac1e9d6b3e..fcd28322c78 100644 --- a/crates/tracedecay/src/mcp/tools/handlers/dashboard.rs +++ b/crates/tracedecay/src/mcp/tools/handlers/dashboard.rs @@ -33,12 +33,13 @@ use tracedecay_daemon_service::{ProjectRuntimePublicationStateV1, ProjectRuntime use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use tracedecay_project::project::TraceDecay; +use tracedecay_runtime_core::native_runtime::{NativeTask, NativeTaskOutcome}; use tracedecay_mcp::handlers::dashboard_lcm::DashboardLcmReadAdapter; use code_reads::DashboardCodeReadAdapter; use tracedecay_dashboard_api::{ - AutomationSchedulerReconciler, DEFAULT_PORT, DashboardApplicationRouters, + AutomationSchedulerReconciler, ConnectionExecutor, DEFAULT_PORT, DashboardApplicationRouters, DashboardApplicationRuntime, DashboardAutomationWriter, DashboardCodeIndexWorkerConfigurationV1, DashboardCodeIndexWorkerSettingsCommitFuture, DashboardCodeIndexWorkerSettingsCommitV1, DashboardCodeIndexWorkerSettingsErrorV1, @@ -513,7 +514,7 @@ struct RunningDashboard { url: String, addr: std::net::SocketAddr, shutdown: Option>, - task: tokio::task::JoinHandle<()>, + task: Arc>, completed: Arc, } @@ -530,7 +531,7 @@ impl RunningDashboard { /// /// The signal is level triggered on purpose. A `Notify::notify_waiters` only /// wakes the waiters already registered, and the drop runs strictly before -/// the runtime marks the `JoinHandle` finished, so a stop that registered in +/// the runtime marks the task finished, so a stop that registered in /// between observed neither the wake nor `is_finished` and then slept until /// its deadline. A closed semaphore is observable by every later waiter: /// `is_closed` reports it synchronously and `acquire` resolves at once. @@ -570,16 +571,19 @@ async fn take_finished_dashboard_for(project_root: &Path) -> Option Result<()> { let url = format!("http://{}/", dashboard.addr); - match dashboard.task.await { - Ok(()) if !exceeded_deadline => Ok(()), - Ok(()) => Err(TraceDecayError::Config { + match dashboard.task.wait().await { + NativeTaskOutcome::Completed(()) if !exceeded_deadline => Ok(()), + NativeTaskOutcome::Completed(()) => Err(TraceDecayError::Config { message: format!("dashboard '{url}' exceeded its shutdown deadline"), }), - Err(error) if error.is_cancelled() && exceeded_deadline => Err(TraceDecayError::Config { + NativeTaskOutcome::Cancelled if exceeded_deadline => Err(TraceDecayError::Config { message: format!("dashboard '{url}' was aborted after its shutdown deadline"), }), - Err(error) => Err(TraceDecayError::Config { - message: format!("dashboard '{url}' task failed: {error}"), + NativeTaskOutcome::Cancelled => Err(TraceDecayError::Config { + message: format!("dashboard '{url}' task failed: task was cancelled"), + }), + NativeTaskOutcome::Panicked(payload) => Err(TraceDecayError::Config { + message: format!("dashboard '{url}' task failed: {payload:?}"), }), } } @@ -652,10 +656,10 @@ pub(crate) async fn shutdown_dashboard_until(deadline: std::time::Instant) -> Re let manager = get_manager().lock().await; ( manager.keys().cloned().collect::>(), - crate::daemon::AbortTasksOnDrop( + AbortDashboardsOnDrop( manager .values() - .map(|dashboard| dashboard.task.abort_handle()) + .map(|dashboard| Arc::clone(&dashboard.task)) .collect(), ), ) @@ -674,6 +678,18 @@ pub(crate) async fn shutdown_dashboard_until(deadline: std::time::Instant) -> Re } } +/// Aborts every dashboard task it holds when a whole-daemon stop is dropped +/// before joining them. +struct AbortDashboardsOnDrop(Vec>>); + +impl Drop for AbortDashboardsOnDrop { + fn drop(&mut self) { + for task in &self.0 { + task.abort(); + } + } +} + pub(crate) async fn shutdown_dashboard() -> Result<()> { shutdown_dashboard_until( std::time::Instant::now() + tracedecay_runtime_core::DAEMON_SHUTDOWN_DEADLINE, @@ -1096,13 +1112,6 @@ pub(super) async fn compute_dashboard( tracedecay_api::static_dashboard_router(product_runtime.dashboard()), ) .await?; - // Dashboard routes still reach Tokio, so the server runs there. - let runtime = tracedecay_runtime_core::native_runtime::tokio_handle( - "mcp.dashboard.hosted_server", - ) - .ok_or_else(|| TraceDecayError::Config { - message: "the hosted dashboard found no Tokio runtime".to_owned(), - })?; let (listener, addr) = bind_dashboard(&host, port).await?; let access = tracedecay_dashboard_api::DashboardAccessToken::mint()?; let url = access.launch_url(addr); @@ -1111,14 +1120,16 @@ pub(super) async fn compute_dashboard( let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); let completed = Arc::new(tokio::sync::Semaphore::new(0)); let task_completion = DashboardTaskCompletion(Arc::clone(&completed)); - let serve_runtime = runtime.clone(); - let task = runtime.spawn(async move { + let task = NativeTask::spawn(async move { let _completion = task_completion; - serve_loopback(serve_runtime, listener, app, async move { + serve_loopback(ConnectionExecutor::Native, listener, app, async move { let _ = shutdown_rx.await; }) .await; - }); + }) + .map_err(|error| TraceDecayError::Config { + message: format!("the hosted dashboard could not start: {error}"), + })?; guard.insert( requested_root, @@ -1126,7 +1137,7 @@ pub(super) async fn compute_dashboard( url: url.clone(), addr, shutdown: Some(shutdown_tx), - task, + task: Arc::new(task), completed, }, ); diff --git a/crates/tracedecay/src/mcp/tools/handlers/dashboard/tests.rs b/crates/tracedecay/src/mcp/tools/handlers/dashboard/tests.rs index 6ca054c40bb..e542bec70d3 100644 --- a/crates/tracedecay/src/mcp/tools/handlers/dashboard/tests.rs +++ b/crates/tracedecay/src/mcp/tools/handlers/dashboard/tests.rs @@ -1,10 +1,16 @@ use std::sync::Arc; +use tracedecay_runtime_core::native_runtime::{self, NativeTask}; + use super::{ DashboardTaskCompletion, RunningDashboard, get_manager, shutdown_dashboard_for_until, shutdown_dashboard_until, }; +fn native_task(work: impl Future + Send + 'static) -> Arc> { + Arc::new(NativeTask::spawn(work).expect("native dashboard task")) +} + /// The dashboard manager is process global, and these tests assert on its /// whole contents, so they must not observe each other's entries. static TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); @@ -21,7 +27,7 @@ async fn shutdown_deadline_aborts_joins_and_clears_dashboard_task() { let (shutdown, _shutdown_requested) = tokio::sync::oneshot::channel(); let completed = Arc::new(tokio::sync::Semaphore::new(0)); let completion = DashboardTaskCompletion(Arc::clone(&completed)); - let task = tokio::spawn(async move { + let task = native_task(async move { let _completion = completion; std::future::pending::<()>().await; }); @@ -47,7 +53,7 @@ async fn shutdown_deadline_aborts_joins_and_clears_dashboard_task() { /// A serving task signals its completion by dropping its /// [`DashboardTaskCompletion`], which happens strictly before the runtime -/// marks the `JoinHandle` finished. A stop that starts inside that window +/// marks the task finished. A stop that starts inside that window /// must still observe the completion: the server is already down, so /// reporting a missed deadline would fail a stop that in fact succeeded. #[tokio::test] @@ -58,13 +64,16 @@ async fn shutdown_observes_a_completion_signalled_before_the_stop_began() { let completed = Arc::new(tokio::sync::Semaphore::new(0)); let completion = DashboardTaskCompletion(Arc::clone(&completed)); let (signalled, observe_signalled) = tokio::sync::oneshot::channel(); - let task = tokio::spawn(async move { + let task = native_task(async move { drop(completion); let _ = signalled.send(()); // Holds the task open well past the stop's deadline so that a stop - // which waits on the `JoinHandle` transition instead of the signal + // which waits on the task's join instead of the signal // reports a deadline miss. - tokio::time::sleep(std::time::Duration::from_millis(300)).await; + let _ = native_runtime::sleep_until( + std::time::Instant::now() + std::time::Duration::from_millis(300), + ) + .await; }); get_manager().lock().await.insert( project_root.clone(), @@ -101,7 +110,7 @@ async fn timed_out_shutdown_owner_aborts_the_dashboard_task() { let completed = Arc::new(tokio::sync::Semaphore::new(0)); let completion = DashboardTaskCompletion(Arc::clone(&completed)); let (alive, task_dropped) = std::sync::mpsc::channel::<()>(); - let task = tokio::spawn(async move { + let task = native_task(async move { let _completion = completion; let _alive = alive; std::future::pending::<()>().await; From 0b49c5cbb5f377361481175821fa516662a6a7e6 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 01:51:44 +0000 Subject: [PATCH 2/3] refactor(daemon): run in-process invocations and MCP dispatch natively The in-process invocation executor and the MCP tool dispatch registry spawned on the daemon's Tokio runtime because the operations they run reached Tokio-only sites: Work attempts, the hosted dashboard, the memory curator, the scheduler reconcile, hook orchestration. Those now run natively, so the executor spawns its invocation and settlement as NativeTasks and the dispatch registry keeps a NativeTaskSet. The controlled RMCP test fixture now polls native timers, since its dispatched executor runs on a native worker. The test hazard list drops the Tokio sites that no longer exist. tracedecay-mcp's production Tokio features drop rt. The recv_writer_reply block_in_place guard stays and now names its remaining Tokio caller, the loopback HTTP application, which project_server_lifecycle.rs keeps on Tokio. --- crates/tracedecay-mcp/Cargo.toml | 2 +- .../tracedecay-mcp/src/server/settlement.rs | 66 ++++++++------- crates/tracedecay-rusqlite-runtime/Cargo.toml | 4 +- .../src/exact_sql/mod.rs | 9 +- .../src/daemon/invocation_executor.rs | 83 +++++++++++-------- .../controlled_invocation_tests.rs | 19 +++-- .../src/daemon/retained_test_support.rs | 30 +++---- .../tracedecay/src/daemon/tests/rmcp_route.rs | 11 ++- crates/tracedecay/src/test_support/mod.rs | 16 +--- 9 files changed, 131 insertions(+), 109 deletions(-) diff --git a/crates/tracedecay-mcp/Cargo.toml b/crates/tracedecay-mcp/Cargo.toml index a8fdf36ec51..7a38e9d7347 100644 --- a/crates/tracedecay-mcp/Cargo.toml +++ b/crates/tracedecay-mcp/Cargo.toml @@ -32,7 +32,7 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" sha2 = "0.11" thiserror = "2" -tokio = { version = "1", features = ["macros", "rt", "sync"] } +tokio = { version = "1", features = ["macros", "sync"] } toml = "1" tracedecay-agent-hosts = { path = "../tracedecay-agent-hosts" } tracedecay-api = { path = "../tracedecay-api" } diff --git a/crates/tracedecay-mcp/src/server/settlement.rs b/crates/tracedecay-mcp/src/server/settlement.rs index 2fcc148b226..2430246f8ac 100644 --- a/crates/tracedecay-mcp/src/server/settlement.rs +++ b/crates/tracedecay-mcp/src/server/settlement.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU8, AtomicUsize, Ordering}; use tracedecay_domain::errors::{Result, TraceDecayError}; -use tracedecay_runtime_core::native_runtime; +use tracedecay_runtime_core::native_runtime::{self, NativeTaskOutcome, NativeTaskSet}; const SETTLEMENT_NOT_STARTED: u8 = 0; const SETTLEMENT_SETTLING: u8 = 1; @@ -155,9 +155,12 @@ impl Drop for DispatchCapacityLease { } } +type DispatchId = u64; + struct RetainedDispatchState { - tasks: tokio::task::JoinSet>, - active: HashMap, + tasks: NativeTaskSet<(DispatchId, Arc)>, + active: HashMap, + next_id: DispatchId, } /// Daemon-owned lifetime authority for admitted MCP handlers. @@ -186,8 +189,9 @@ impl RetainedDispatchRegistry { capacity, active_slots: Arc::new(AtomicUsize::new(0)), state: tokio::sync::Mutex::new(RetainedDispatchState { - tasks: tokio::task::JoinSet::new(), + tasks: NativeTaskSet::default(), active: HashMap::new(), + next_id: 0, }), #[cfg(any(test, feature = "test-transport"))] retained_spawn_count: AtomicUsize::new(0), @@ -239,10 +243,6 @@ impl RetainedDispatchRegistry { T: Send + 'static, F: Future> + Send + 'static, { - // Tool dispatches still reach Tokio, so they settle there. - let runtime = - tracedecay_runtime_core::native_runtime::tokio_handle("mcp.dispatch.settlement") - .ok_or_else(dispatch_runtime_unavailable_error)?; let capacity_lease = self.acquire_capacity()?; let mut state = self.state.lock().await; Self::reap_finished(&mut state); @@ -254,21 +254,26 @@ impl RetainedDispatchRegistry { let worker_settlement = Arc::clone(&settlement); let settlement_lease = DispatchSettlementLease(Arc::clone(&settlement)); let (sender, receiver) = tokio::sync::oneshot::channel(); - #[cfg(any(test, feature = "test-transport"))] - self.retained_spawn_count.fetch_add(1, Ordering::AcqRel); - let task = state.tasks.spawn_on( - async move { + let id = state.next_id; + state.next_id += 1; + // Tool dispatches run on the native runtime. Handlers move synchronous + // storage-port calls off the worker with `off_native_thread` and reach + // Tokio-only work through `native_runtime::tokio_handle`. + state + .tasks + .spawn(async move { let _capacity_lease = capacity_lease; worker_settlement.mark_settling(); let output = future.await; drop(settlement_lease); let _ = sender.send(output); - worker_settlement - }, - &runtime, - ); + (id, worker_settlement) + }) + .map_err(|error| dispatch_runtime_unavailable_error(&error))?; + #[cfg(any(test, feature = "test-transport"))] + self.retained_spawn_count.fetch_add(1, Ordering::AcqRel); state.active.insert( - task.id(), + id, ActiveDispatch { cancellation, live_cancellable, @@ -295,28 +300,27 @@ impl RetainedDispatchRegistry { } fn reap_finished(state: &mut RetainedDispatchState) { - while let Some(joined) = state.tasks.try_join_next_with_id() { + while let Some(joined) = state.tasks.try_join_next() { Self::settle_joined(state, joined); } } fn settle_joined( state: &mut RetainedDispatchState, - joined: std::result::Result< - (tokio::task::Id, Arc), - tokio::task::JoinError, - >, + joined: NativeTaskOutcome<(DispatchId, Arc)>, ) { match joined { - Ok((id, settlement)) => { + NativeTaskOutcome::Completed((id, settlement)) => { settlement.mark_joined(); state.active.remove(&id); } - Err(error) => { - if let Some(active) = state.active.remove(&error.id()) { - active.settlement.mark_joined(); - } - tracing::error!(error = %error, "retained MCP dispatch task failed"); + failed => { + // A failed task returns no id. Its dropped settlement lease + // already marked it joined, which identifies its entry. + state + .active + .retain(|_, active| active.settlement.snapshot() != DispatchSettlement::Joined); + tracing::error!(outcome = ?failed, "retained MCP dispatch task failed"); } } } @@ -330,7 +334,7 @@ impl RetainedDispatchRegistry { let _ = active.cancellation.cancel(requested_at); } } - while let Some(joined) = state.tasks.join_next_with_id().await { + while let Some(joined) = state.tasks.join_next().await { Self::settle_joined(&mut state, joined); } } @@ -783,11 +787,11 @@ fn dispatch_shutdown_error() -> TraceDecayError { ) } -fn dispatch_runtime_unavailable_error() -> TraceDecayError { +fn dispatch_runtime_unavailable_error(error: &str) -> TraceDecayError { TraceDecayError::project_route( "tool_dispatch_runtime_unavailable", true, - "MCP tool dispatch found no Tokio runtime", + format!("MCP tool dispatch was not started: {error}"), ) } diff --git a/crates/tracedecay-rusqlite-runtime/Cargo.toml b/crates/tracedecay-rusqlite-runtime/Cargo.toml index e674286a74c..dda3f67c8d5 100644 --- a/crates/tracedecay-rusqlite-runtime/Cargo.toml +++ b/crates/tracedecay-rusqlite-runtime/Cargo.toml @@ -24,8 +24,8 @@ ring = "0.17.14" sha2 = "0.11.0" thiserror = "2" asupersync.workspace = true -# Synchronous storage ports wait on the writer from Tokio daemon workers; see -# `recv_writer_reply`. +# The loopback HTTP application's handlers wait on the writer from Tokio +# workers; see `recv_writer_reply`. tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread"] } tracedecay-domain = { path = "../tracedecay-domain" } tracedecay-contracts = { path = "../tracedecay-contracts" } diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs index ffb49fea29a..bedc720d3d6 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs @@ -105,9 +105,12 @@ pub struct ExactSqlHandle { /// /// Every synchronous storage port in this workspace (`WorkAttemptStoragePort`, /// `WorkflowRunStoragePort`, and their siblings) reaches the writer through -/// these one-shot replies, and the daemon calls those ports from Tokio tasks. -/// A plain `recv_blocking` therefore parks a whole worker for as long as the -/// writer's command queue takes to reach this command. On a small runtime two +/// these one-shot replies. The daemon's loopback HTTP application still serves +/// on Tokio workers, because its `/remote-deletions` route reaches the project +/// server lifecycle, and its Remote Brain handlers (node provisioning among +/// them) call those ports there. A plain `recv_blocking` would park a whole +/// worker for as long as the writer's command queue takes to reach this +/// command. On a small runtime two /// concurrent waits starve every other task on it, including the HTTP reads /// whose admitted deadline then expires. `block_in_place` hands the worker's /// run queue to another thread for the wait; it panics outside a multi-thread diff --git a/crates/tracedecay/src/daemon/invocation_executor.rs b/crates/tracedecay/src/daemon/invocation_executor.rs index c58cfa230fc..87b343f20a8 100644 --- a/crates/tracedecay/src/daemon/invocation_executor.rs +++ b/crates/tracedecay/src/daemon/invocation_executor.rs @@ -498,7 +498,7 @@ impl tracedecay_contracts::ApplicationInvocationExecutor for InProcessDaemonInvo pub(super) async fn settle_in_process_invocation( request_cancellations: &tracedecay_daemon_service::RequestCancellationRegistryV1, request_id: &str, - invocation: tokio::task::JoinHandle, + invocation: native_runtime::NativeTask, remaining: Duration, cancellation: tracedecay_contracts::CancellationSignal, admitted_cancellation: Option, @@ -516,8 +516,8 @@ pub(super) async fn settle_in_process_invocation( CancellationStage::EffectInFlight } }; - let invocation = invocation; - tokio::pin!(invocation); + let invocation_wait = invocation.wait(); + tokio::pin!(invocation_wait); let cancellation_wait = tracedecay_daemon_protocol::wait_for_cancellation(cancellation); tokio::pin!(cancellation_wait); let has_admitted_cancellation = admitted_cancellation.is_some(); @@ -529,9 +529,7 @@ pub(super) async fn settle_in_process_invocation( }; tokio::pin!(admitted_cancellation_wait); let timed_out = tokio::select! { - response = &mut invocation => { - return response.map_err(|_| tracedecay_daemon_protocol::DaemonInvocationError::Unavailable); - } + outcome = &mut invocation_wait => return invocation_response(outcome), () = &mut cancellation_wait => false, () = &mut admitted_cancellation_wait => false, // A timer that cannot start fails closed, as an elapsed deadline. @@ -542,7 +540,7 @@ pub(super) async fn settle_in_process_invocation( } match policy { tracedecay_daemon_protocol::InvocationCancellationPolicy::ReadOnly => { - if native_runtime::timeout(Duration::from_secs(1), &mut invocation) + if native_runtime::timeout(Duration::from_secs(1), &mut invocation_wait) .await .is_err() { @@ -565,12 +563,12 @@ pub(super) async fn settle_in_process_invocation( // effect merely outlived its budget. Wait for the authoritative // settlement over the same grace the daemon's own clients keep // reading for, so the effect's real terminal is the one reported. - match native_runtime::timeout(response_grace, &mut invocation).await { - Ok(Ok(response)) => Ok(response), + match native_runtime::timeout(response_grace, &mut invocation_wait).await { + Ok(native_runtime::NativeTaskOutcome::Completed(response)) => Ok(response), Err(native_runtime::TimeoutError::TimerUnavailable(_)) => { Err(tracedecay_daemon_protocol::DaemonInvocationError::Unavailable) } - Ok(Err(_)) | Err(native_runtime::TimeoutError::Elapsed) => { + Ok(_) | Err(native_runtime::TimeoutError::Elapsed) => { Ok(DaemonInvocationResponse::problem( request_id, tracedecay_daemon_protocol::DaemonInvocationProblem::ResetRequired, @@ -581,6 +579,19 @@ pub(super) async fn settle_in_process_invocation( } } +fn invocation_response( + outcome: native_runtime::NativeTaskOutcome, +) -> std::result::Result +{ + match outcome { + native_runtime::NativeTaskOutcome::Completed(response) => Ok(response), + native_runtime::NativeTaskOutcome::Cancelled + | native_runtime::NativeTaskOutcome::Panicked(_) => { + Err(tracedecay_daemon_protocol::DaemonInvocationError::Unavailable) + } + } +} + fn map_operation_event_invocation_error( error: tracedecay_application::operation_stream::OperationEventError, ) -> tracedecay_contracts::InvocationError { @@ -651,34 +662,36 @@ impl tracedecay_daemon_protocol::DaemonInvocationExecutor for InProcessDaemonInv let executor = self.clone(); let admitted_cancellation = self.admitted_cancellation.clone(); let request_cancellations = self.invocation.service.request_cancellations().clone(); - // Dispatched operations still reach Tokio-only sites (the hosted - // dashboard server, the memory curator run). - let runtime = tracedecay_runtime_core::native_runtime::tokio_handle( - "daemon.invocation_executor.in_process", - ) - .ok_or(tracedecay_daemon_protocol::DaemonInvocationError::Unavailable)?; - let invocation_runtime = runtime.clone(); - runtime - .spawn(async move { - let request_id = request.request_id.clone(); - let invocation = invocation_runtime.spawn(tracing::Instrument::instrument( + let unavailable = |_| tracedecay_daemon_protocol::DaemonInvocationError::Unavailable; + match native_runtime::NativeTask::spawn(async move { + let request_id = request.request_id.clone(); + let invocation = + native_runtime::NativeTask::spawn(tracing::Instrument::instrument( async move { executor.invoke_once(request).await }, tracing::trace_span!("daemon.invocation.invoke_once"), - )); - settle_in_process_invocation( - &request_cancellations, - &request_id, - invocation, - remaining, - cancellation, - admitted_cancellation, - policy, - DAEMON_TOOL_RESPONSE_GRACE, - ) - .await - }) + )) + .map_err(unavailable)?; + settle_in_process_invocation( + &request_cancellations, + &request_id, + invocation, + remaining, + cancellation, + admitted_cancellation, + policy, + DAEMON_TOOL_RESPONSE_GRACE, + ) .await - .map_err(|_| tracedecay_daemon_protocol::DaemonInvocationError::Unavailable)? + }) + .map_err(unavailable)? + .await + { + native_runtime::NativeTaskOutcome::Completed(result) => result, + native_runtime::NativeTaskOutcome::Cancelled + | native_runtime::NativeTaskOutcome::Panicked(_) => { + Err(tracedecay_daemon_protocol::DaemonInvocationError::Unavailable) + } + } }) } diff --git a/crates/tracedecay/src/daemon/invocation_executor/controlled_invocation_tests.rs b/crates/tracedecay/src/daemon/invocation_executor/controlled_invocation_tests.rs index 92d7c3d857c..d761c1e6a19 100644 --- a/crates/tracedecay/src/daemon/invocation_executor/controlled_invocation_tests.rs +++ b/crates/tracedecay/src/daemon/invocation_executor/controlled_invocation_tests.rs @@ -10,6 +10,7 @@ use tracedecay_daemon_protocol::{ DaemonInvocationResponse, }; use tracedecay_daemon_service::{DaemonInvocationOperation, RequestCancellationRegistryV1}; +use tracedecay_runtime_core::native_runtime::NativeTask; #[test] fn native_worktree_operations_request_the_native_integration_owner() { @@ -47,11 +48,12 @@ async fn in_process_effect_cancellation_requests_daemon_cancel_and_awaits_settle let daemon_cancellation = lease.token(); let completed = Arc::new(AtomicBool::new(false)); let completed_by_invocation = Arc::clone(&completed); - let invocation = tokio::spawn(async move { + let invocation = NativeTask::spawn(async move { daemon_cancellation.cancelled().await; completed_by_invocation.store(true, Ordering::Release); authoritative_response(REQUEST_ID) - }); + }) + .expect("native invocation task"); let cancellation = CancellationSignal::active("cancel.in-process-effect-cancel").expect("cancellation signal"); let caller_cancellation = cancellation.clone(); @@ -92,11 +94,12 @@ async fn in_process_effect_deadline_requests_daemon_cancel_and_awaits_settlement let daemon_cancellation = lease.token(); let completed = Arc::new(AtomicBool::new(false)); let completed_by_invocation = Arc::clone(&completed); - let invocation = tokio::spawn(async move { + let invocation = NativeTask::spawn(async move { daemon_cancellation.cancelled().await; completed_by_invocation.store(true, Ordering::Release); authoritative_response(REQUEST_ID) - }); + }) + .expect("native invocation task"); let response = tokio::time::timeout( Duration::from_secs(1), @@ -129,7 +132,8 @@ async fn in_process_effect_without_settlement_returns_reset_required() { let lease = request_cancellations .register(REQUEST_ID) .expect("register invocation cancellation"); - let invocation = tokio::spawn(std::future::pending::()); + let invocation = NativeTask::spawn(std::future::pending::()) + .expect("native invocation task"); let started = std::time::Instant::now(); let response = tokio::time::timeout( @@ -180,11 +184,12 @@ async fn in_process_read_observes_admitted_outer_cancellation_after_start() { .expect("register invocation cancellation"); let admitted_cancellation = lease.token(); let (started, started_rx) = tokio::sync::oneshot::channel::<()>(); - let invocation = tokio::spawn(async move { + let invocation = NativeTask::spawn(async move { let _admission = admission; started.send(()).expect("report invocation start"); std::future::pending::().await - }); + }) + .expect("native invocation task"); started_rx.await.expect("nested invocation starts"); let quiescing_registry = registry.clone(); let quiescing_project = project.clone(); diff --git a/crates/tracedecay/src/daemon/retained_test_support.rs b/crates/tracedecay/src/daemon/retained_test_support.rs index 67600c8baa1..7ad8a08d5c6 100644 --- a/crates/tracedecay/src/daemon/retained_test_support.rs +++ b/crates/tracedecay/src/daemon/retained_test_support.rs @@ -163,20 +163,22 @@ impl tracedecay_daemon_protocol::DaemonInvocationExecutor for RetainedOwnerTestE } let request_id = request.request_id.clone(); let executor = self.clone(); - let invocation = tokio::spawn(async move { - executor - .service - .invoke_with_cancellation( - &executor.lsp_registry, - Some(&executor.project_root), - None, - None, - None, - request, - None, - ) - .await - }); + let invocation = + tracedecay_runtime_core::native_runtime::NativeTask::spawn(async move { + executor + .service + .invoke_with_cancellation( + &executor.lsp_registry, + Some(&executor.project_root), + None, + None, + None, + request, + None, + ) + .await + }) + .map_err(|_| tracedecay_daemon_protocol::DaemonInvocationError::Unavailable)?; super::invocation_executor::settle_in_process_invocation( self.service.request_cancellations(), &request_id, diff --git a/crates/tracedecay/src/daemon/tests/rmcp_route.rs b/crates/tracedecay/src/daemon/tests/rmcp_route.rs index b67d44effce..fd5b98ecdcd 100644 --- a/crates/tracedecay/src/daemon/tests/rmcp_route.rs +++ b/crates/tracedecay/src/daemon/tests/rmcp_route.rs @@ -839,14 +839,21 @@ impl ControlledCancellationExecutor { } async fn await_cancellation(&self, cancellation: tracedecay_contracts::CancellationSignal) { + // Dispatches run on native workers, so the fixture polls on native + // timers. + let tick = || { + tracedecay_runtime_core::native_runtime::sleep_until( + std::time::Instant::now() + Duration::from_millis(1), + ) + }; let ordinal = self.started.fetch_add(1, Ordering::SeqCst); while !cancellation.is_cancelled() { - tokio::time::sleep(Duration::from_millis(1)).await; + let _ = tick().await; } self.cancellation_observed.fetch_add(1, Ordering::SeqCst); if ordinal == 0 { while !self.release_first.load(Ordering::SeqCst) { - tokio::time::sleep(Duration::from_millis(1)).await; + let _ = tick().await; } } self.completed.fetch_add(1, Ordering::SeqCst); diff --git a/crates/tracedecay/src/test_support/mod.rs b/crates/tracedecay/src/test_support/mod.rs index 0b2601aa07f..4b32912926a 100644 --- a/crates/tracedecay/src/test_support/mod.rs +++ b/crates/tracedecay/src/test_support/mod.rs @@ -17,21 +17,9 @@ pub mod host_admission; /// that parked a native worker. An off-Tokio test asserts this does not move. #[cfg(test)] pub(crate) fn native_runtime_hazards() -> u64 { - [ - "daemon-service.operation_events.sse_disconnect", - "daemon-service.hook_orchestration.admit", - "daemon-service.work_blocked_interval_recovery.start", - "daemon-service.observability.mount_drain", - "daemon-service.observability.settle_drain", - "daemon-service.observability.producer_drop", - "daemon-service.work_attempt.spawn_for_worktree", - "daemon-service.host_admission.bootstrap", + tracedecay_runtime_core::native_runtime::tokio_runtime_missing_count( "daemon-service.shutdown.runner", - ] - .into_iter() - .map(tracedecay_runtime_core::native_runtime::tokio_runtime_missing_count) - .sum::() - + tracedecay_rusqlite_runtime::exact_sql::native_thread_writer_waits() + ) + tracedecay_rusqlite_runtime::exact_sql::native_thread_writer_waits() } /// Park after core publication when `TRACEDECAY_TEST_HOLD_AFTER_CORE_PUBLISH` From be67296772987a1234159b44561bc73715605e32 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 02:36:12 +0000 Subject: [PATCH 3/3] refactor(rusqlite-runtime): drop the Tokio block_in_place writer wait recv_writer_reply handed a Tokio worker's run queue away during a writer wait. No Tokio worker in the daemon's acceptance suites waits on the writer any more except the loopback HTTP listener's remote-node provisioning, which now runs on a blocking-pool thread, so the branch and tokio's rt-multi-thread feature go. The native-thread writer-wait counter and its log stay. The rusqlite-runtime test that pinned the Tokio hand-off is deleted with it. --- crates/tracedecay-rusqlite-runtime/Cargo.toml | 6 +-- .../src/exact_sql/mod.rs | 35 ++++------------ .../src/exact_sql/tests/async_dispatch.rs | 41 ------------------- .../tracedecay/src/daemon/http_application.rs | 34 +++++++++++---- 4 files changed, 38 insertions(+), 78 deletions(-) diff --git a/crates/tracedecay-rusqlite-runtime/Cargo.toml b/crates/tracedecay-rusqlite-runtime/Cargo.toml index dda3f67c8d5..ede7c20a1f3 100644 --- a/crates/tracedecay-rusqlite-runtime/Cargo.toml +++ b/crates/tracedecay-rusqlite-runtime/Cargo.toml @@ -24,9 +24,9 @@ ring = "0.17.14" sha2 = "0.11.0" thiserror = "2" asupersync.workspace = true -# The loopback HTTP application's handlers wait on the writer from Tokio -# workers; see `recv_writer_reply`. -tokio = { version = "1", default-features = false, features = ["rt", "rt-multi-thread"] } +# `native_runtime::tokio_handle` hands the daemon's Tokio runtime to the sites +# that still need one. +tokio = { version = "1", default-features = false, features = ["rt"] } tracedecay-domain = { path = "../tracedecay-domain" } tracedecay-contracts = { path = "../tracedecay-contracts" } tracedecay-store = { path = "../tracedecay-store" } diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs index bedc720d3d6..6212d54db9c 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs @@ -100,39 +100,22 @@ pub struct ExactSqlHandle { write_authority: Option>, } -/// Waits for the serialized writer actor's reply without stalling the async -/// worker that issued the command. +/// Waits for the serialized writer actor's reply by parking the calling +/// thread. /// /// Every synchronous storage port in this workspace (`WorkAttemptStoragePort`, /// `WorkflowRunStoragePort`, and their siblings) reaches the writer through -/// these one-shot replies. The daemon's loopback HTTP application still serves -/// on Tokio workers, because its `/remote-deletions` route reaches the project -/// server lifecycle, and its Remote Brain handlers (node provisioning among -/// them) call those ports there. A plain `recv_blocking` would park a whole -/// worker for as long as the writer's command queue takes to reach this -/// command. On a small runtime two -/// concurrent waits starve every other task on it, including the HTTP reads -/// whose admitted deadline then expires. `block_in_place` hands the worker's -/// run queue to another thread for the wait; it panics outside a multi-thread -/// runtime, so the flavor is checked first and everything else -/// (current-thread runtimes, plain threads) keeps the previous inline -/// behavior. The remaining `block_in_place` panic case is a `LocalSet` on a -/// multi-thread runtime, which this workspace does not use. -/// -/// Asupersync has no `block_in_place`, so a wait on a native runtime thread -/// (a worker or a `block_on` caller; its blocking pool carries no runtime -/// handle) still parks that thread. Each such wait is logged and counted in -/// [`native_thread_writer_waits`] so the caller can be moved onto -/// `native_runtime::off_native_thread` or `native_runtime::spawn_blocking`. +/// these one-shot replies, so its caller must run where a thread may park: a +/// blocking-pool thread (`native_runtime::spawn_blocking`, +/// `native_runtime::off_native_thread`, `NativeTask::spawn_off_native_thread`) +/// or a dedicated thread. A wait on a native runtime thread (a worker or a +/// `block_on` caller; its blocking pool carries no runtime handle) parks that +/// thread and every task queued on it, so each one is logged and counted in +/// [`native_thread_writer_waits`]. #[track_caller] fn recv_writer_reply( response: async_channel::Receiver, ) -> Result { - if let Ok(handle) = tokio::runtime::Handle::try_current() - && handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread - { - return tokio::task::block_in_place(|| response.recv_blocking()); - } if asupersync::runtime::Runtime::current_handle().is_some() { NATIVE_THREAD_WRITER_WAITS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); tracing::error!( diff --git a/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs b/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs index a6658929a6f..4bcf98d47f2 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/tests/async_dispatch.rs @@ -395,47 +395,6 @@ fn async_commit_drop_after_admission_preserves_committed_data() { }); } -/// Two outstanding writer replies must not take the whole runtime down with -/// them. -/// -/// The synchronous storage ports wait for the writer actor from inside Tokio -/// tasks. When that wait parks its worker, two concurrent waits on a -/// two-worker runtime leave nothing to poll anything else, and every other -/// request on that runtime runs out its deadline instead of being served. -#[test] -fn outstanding_writer_replies_leave_the_runtime_able_to_poll_other_tasks() { - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) - .enable_all() - .build() - .unwrap(); - let (reply, waiting) = mpsc::channel(); - let (progress, observed) = mpsc::channel(); - let mut senders = Vec::new(); - - for _ in 0..2 { - let (sender, response) = async_channel::bounded::<()>(1); - senders.push(sender); - let entering = reply.clone(); - runtime.spawn(async move { - entering.send(()).unwrap(); - let _ = recv_writer_reply(response); - }); - } - for _ in 0..2 { - waiting.recv_timeout(Duration::from_secs(5)).unwrap(); - } - - runtime.spawn(async move { progress.send(()).unwrap() }); - - observed - .recv_timeout(Duration::from_secs(5)) - .expect("a task must still be polled while two writer replies are outstanding"); - for sender in senders { - let _ = sender.send_blocking(()); - } -} - #[test] fn admitted_execute_returning_completes_after_reply_waiter_is_abandoned() { current_thread().block_on(async { diff --git a/crates/tracedecay/src/daemon/http_application.rs b/crates/tracedecay/src/daemon/http_application.rs index 12ea6b8106d..84224f4cc98 100644 --- a/crates/tracedecay/src/daemon/http_application.rs +++ b/crates/tracedecay/src/daemon/http_application.rs @@ -429,14 +429,32 @@ async fn provision_remote_node( let Some(runtime) = registry.remote_runtime() else { return StatusCode::SERVICE_UNAVAILABLE.into_response(); }; - match tracing::Instrument::instrument( - runtime.provision_remote_node(request.grant, request.admission), - tracing::trace_span!("daemon.http.application.remote_node_provision"), - ) - .await - { - Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(error) => remote_node_operator_refusal(error), + // This listener serves on Tokio workers, and provisioning writes the node + // store through synchronous storage ports that park their thread on the + // writer, so it runs on a blocking-pool thread. + let provision = + native_runtime::NativeTask::spawn_off_native_thread(tracing::Instrument::instrument( + async move { + runtime + .provision_remote_node(request.grant, request.admission) + .await + }, + tracing::trace_span!("daemon.http.application.remote_node_provision"), + )); + let Ok(provision) = provision else { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; + match provision.wait().await { + native_runtime::NativeTaskOutcome::Completed(Ok(())) => { + StatusCode::NO_CONTENT.into_response() + } + native_runtime::NativeTaskOutcome::Completed(Err(error)) => { + remote_node_operator_refusal(error) + } + native_runtime::NativeTaskOutcome::Cancelled + | native_runtime::NativeTaskOutcome::Panicked(_) => { + StatusCode::SERVICE_UNAVAILABLE.into_response() + } } }