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-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 248e15b6eee..32b17be16f6 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 -# Synchronous storage ports wait on the writer from Tokio daemon 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 c3bd756dd55..f3cf38e375d 100644 --- a/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/exact_sql/mod.rs @@ -100,36 +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, 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 -/// 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 0cf5ce17ed2..165743fbf90 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 7b8e82733d0..84224f4cc98 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; @@ -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() + } } } @@ -901,9 +919,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/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/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; 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`