diff --git a/crates/tracedecay-daemon-service/src/invocation/lsp.rs b/crates/tracedecay-daemon-service/src/invocation/lsp.rs index a4328c5daa..7c19fb938f 100644 --- a/crates/tracedecay-daemon-service/src/invocation/lsp.rs +++ b/crates/tracedecay-daemon-service/src/invocation/lsp.rs @@ -189,7 +189,12 @@ impl DaemonInvocationService { ) -> Option { let owner = self.lsp_owner(Some(project_root)).await?; let grant = owner.scope_grant?; - let scope_set = owner.scope_set_storage?.read(scope_set_id).ok().flatten()?; + let scope_set = owner + .scope_set_storage? + .read_async(scope_set_id) + .await + .ok() + .flatten()?; // Persisted scope routes are actor-sealed: an admission on the active // root authorizes reading only the scope sets its own actor persisted, // never another actor's mounted routes. @@ -221,7 +226,10 @@ impl DaemonInvocationService { let active_owner = self.lsp_owner(Some(active_project_root)).await?; let active_scope = active_owner.scope_grant.as_ref()?.scope.clone(); let active_storage = active_owner.scope_set_storage?; - let current = active_storage.read(&request.scope_set_id).ok()?; + let current = active_storage + .read_async(&request.scope_set_id) + .await + .ok()?; let next_revision = match (request.expected_revision, current.as_ref()) { (None, None) => ScopeSetRevision::new(1).ok()?, (Some(expected), Some(current)) if current.revision() == expected => { diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/scope_set.rs b/crates/tracedecay-rusqlite-runtime/src/repository/scope_set.rs index a9b2276f2b..58b0f242c3 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/scope_set.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/scope_set.rs @@ -168,6 +168,21 @@ impl AuthorizedScopeSetSqliteStorage { decode_registered_rows(rows.rows) } + /// Await reader admission and the query without parking a runtime worker. + pub async fn read_async( + &self, + scope_set_id: &ScopeSetId, + ) -> Result, AuthorizedScopeSetStoreError> { + let rows = self + .handle() + .query_async( + registered_read_statement(scope_set_id)?, + std::time::Duration::from_secs(5), + ) + .await?; + decode_registered_rows(rows.rows) + } + pub fn compare_and_swap( &self, expected_revision: Option, diff --git a/crates/tracedecay/src/daemon.rs b/crates/tracedecay/src/daemon.rs index 18ac7a6d22..cb9168216e 100644 --- a/crates/tracedecay/src/daemon.rs +++ b/crates/tracedecay/src/daemon.rs @@ -450,8 +450,6 @@ use project_open_handshake::{ }; mod project_open_orchestration; mod project_routing; -// ambient-tokio: its Tokio timers run on Tokio tasks or inside the -// native_runtime::on_tokio hops in project_server_lifecycle_hops. mod project_server_lifecycle; #[cfg(unix)] use project_open_orchestration::{ @@ -482,11 +480,10 @@ use project_routing::{ }; #[cfg(test)] use project_server_lifecycle::replay_user_profile_host_admission_for_identity; -use project_server_lifecycle::{schedule_project_server_retirement, shutdown_project_servers}; -mod project_server_lifecycle_hops; -use project_server_lifecycle_hops::{ +use project_server_lifecycle::{ await_user_profile_host_admission_replay_for_identity, cancel_retained_session_history, - schedule_user_profile_host_admission_replay_for_identity, + schedule_project_server_retirement, schedule_user_profile_host_admission_replay_for_identity, + shutdown_project_servers, }; #[cfg(unix)] mod scheduler; diff --git a/crates/tracedecay/src/daemon/bootstrap.rs b/crates/tracedecay/src/daemon/bootstrap.rs index 8f4e7f9039..05b225452b 100644 --- a/crates/tracedecay/src/daemon/bootstrap.rs +++ b/crates/tracedecay/src/daemon/bootstrap.rs @@ -427,7 +427,7 @@ async fn run_foreground_loopback( owner_phases, move |project_server_deadline| async move { shutdown_project_servers( - tokio::time::Instant::from_std(project_server_deadline), + project_server_deadline, &server_store_administration, &http_application_registry, ) diff --git a/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs b/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs index f2164363cf..e5f8374ff4 100644 --- a/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs +++ b/crates/tracedecay/src/daemon/branch_admin/project_retirement.rs @@ -96,7 +96,7 @@ impl ProjectServerRetirementAdmission<'_> { track_project_server_retirement_after_admission( &mut self.retirements, owner, - spawn_tokio_retirement(retirement), + spawn_native_retirement(retirement), ); } @@ -144,31 +144,22 @@ where }; finalizer.complete(status); }; - // Without a Tokio runtime the retirement is dropped unstarted, and its - // finalizer publishes the failed receipt. - if let Some(runtime) = native_runtime::tokio_handle(RETIREMENT_TOKIO_SITE) { - runtime.spawn(retirement); + if let Err(error) = NativeTask::spawn(retirement) { + tracing::error!(%error, "project server capacity retirement was refused"); } completion } -/// Retirements reach `project_server_lifecycle`, whose timers are Tokio's, so -/// they run on the daemon's Tokio runtime. -const RETIREMENT_TOKIO_SITE: &str = "daemon.project_retirement"; - -/// Spawns `retirement` on the Tokio runtime and returns its receipt. Without -/// a Tokio runtime the retirement never runs and its receipt is a failure. -fn spawn_tokio_retirement( +fn spawn_native_retirement( retirement: impl std::future::Future + Send + 'static, ) -> impl std::future::Future + Send + 'static { - let task = native_runtime::tokio_handle(RETIREMENT_TOKIO_SITE) - .map(|runtime| runtime.spawn(retirement)); + let task = NativeTask::spawn(retirement); async move { match task { - Some(task) => tokio_retirement_status(task).await, - None => ProjectServerRetirementStatus::Failed( - "no Tokio runtime ran the project server retirement".to_owned(), - ), + Ok(task) => native_retirement_status(task).await, + Err(error) => ProjectServerRetirementStatus::Failed(format!( + "native runtime refused project server retirement: {error}" + )), } } } @@ -226,6 +217,7 @@ async fn wait_for_project_server_retirement( } } +#[cfg(test)] async fn tokio_retirement_status( task: tokio::task::JoinHandle<()>, ) -> ProjectServerRetirementStatus { @@ -248,6 +240,18 @@ async fn aborted_native_retirement_status(task: NativeTask<()>) -> ProjectServer } } +async fn native_retirement_status(task: NativeTask<()>) -> ProjectServerRetirementStatus { + match task.wait().await { + NativeTaskOutcome::Completed(()) => ProjectServerRetirementStatus::Clean, + NativeTaskOutcome::Cancelled => { + ProjectServerRetirementStatus::Failed("retirement task was cancelled".to_owned()) + } + NativeTaskOutcome::Panicked(panic) => { + ProjectServerRetirementStatus::Failed(format!("task panicked: {panic}")) + } + } +} + /// Records one retirement and publishes `joined`'s status from a native /// task. When the native runtime refuses that task, its finalizer publishes /// a failed receipt. @@ -293,13 +297,12 @@ pub(super) async fn track_retirement_task( track_project_server_retirement(retirements, owner, tokio_retirement_status(task)).await; } -/// Spawns `retirement` on the Tokio runtime and tracks its receipt. -pub(super) async fn track_tokio_retirement( +pub(super) async fn track_native_retirement( retirements: &tokio::sync::Mutex>, owner: StoreOwnerKey, retirement: impl std::future::Future + Send + 'static, ) { - track_project_server_retirement(retirements, owner, spawn_tokio_retirement(retirement)).await; + track_project_server_retirement(retirements, owner, spawn_native_retirement(retirement)).await; } pub(super) async fn track_aborted_native_retirement_task( @@ -808,13 +811,19 @@ mod tests { let capacity_admission = Arc::clone(&capacity_gate).lock_owned().await; let release_capacity = Arc::new(tokio::sync::Notify::new()); let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let ran_without_tokio = Arc::new(AtomicBool::new(false)); let mut admission = administration .acquire_project_server_retirement_admission() .await; let completion = admission.spawn_and_track_fallible(owner("project-capacity"), { let release_capacity = Arc::clone(&release_capacity); + let ran_without_tokio = Arc::clone(&ran_without_tokio); async move { let _capacity_admission = capacity_admission; + ran_without_tokio.store( + tokio::runtime::Handle::try_current().is_err(), + Ordering::Release, + ); let _ = started_tx.send(()); release_capacity.notified().await; Ok(()) @@ -823,15 +832,30 @@ mod tests { drop(admission); drop(completion); started_rx.await.expect("capacity retirement started"); + assert!( + ran_without_tokio.load(Ordering::Acquire), + "capacity retirement must run on the native runtime without entering Tokio" + ); assert!( Arc::clone(&capacity_gate).try_lock_owned().is_err(), "a cancelled caller cannot release capacity while detached cleanup is pending" ); + let tracked_completion = administration + .acquire_project_server_retirement_admission() + .await + .prior_completions_for_owner(&owner("project-capacity")) + .pop() + .expect("the cancelled caller must leave a tracked capacity receipt"); release_capacity.notify_one(); - let recovered_capacity = tokio::time::timeout( - std::time::Duration::from_secs(1), - Arc::clone(&capacity_gate).lock_owned(), - ) + let recovered_capacity = tokio::time::timeout(std::time::Duration::from_secs(1), async { + let recovered_capacity = Arc::clone(&capacity_gate).lock_owned().await; + // The guard drops inside the retirement, before its terminal receipt is published. + tracked_completion + .wait() + .await + .expect("detached capacity retirement completes cleanly"); + recovered_capacity + }) .await .expect("detached cleanup must release capacity after completion"); @@ -1059,7 +1083,7 @@ mod tests { let http_application_registry = crate::daemon::http_application::DaemonHttpApplicationRegistry::default(); let mut shutdown = Box::pin(project_server_lifecycle::shutdown_project_servers( - tokio::time::Instant::now() + std::time::Duration::from_secs(5), + Instant::now() + std::time::Duration::from_secs(5), &administration, &http_application_registry, )); diff --git a/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs b/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs index 61335b292b..b296d74352 100644 --- a/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs +++ b/crates/tracedecay/src/daemon/branch_admin/remote_recovery_lifecycle.rs @@ -278,7 +278,7 @@ pub(super) async fn retire_runtime_work( } super::super::project_server_lifecycle::retire_project_servers_now(servers).await; }; - super::project_retirement::track_tokio_retirement(tracked_retirements, owner, retirement) + super::project_retirement::track_native_retirement(tracked_retirements, owner, retirement) .await; } let deadline = std::time::Instant::now() + DAEMON_TASK_ABORT_DEADLINE; diff --git a/crates/tracedecay/src/daemon/engine/shutdown.rs b/crates/tracedecay/src/daemon/engine/shutdown.rs index 5c91f6d0fe..3324d89f98 100644 --- a/crates/tracedecay/src/daemon/engine/shutdown.rs +++ b/crates/tracedecay/src/daemon/engine/shutdown.rs @@ -288,7 +288,7 @@ impl DaemonEngine { deadline: std::time::Instant, ) -> ShutdownTaskReceipt { shutdown_project_servers( - tokio::time::Instant::from_std(deadline), + deadline, &self.store_administration, &self.http_application_registry, ) diff --git a/crates/tracedecay/src/daemon/production_harness.rs b/crates/tracedecay/src/daemon/production_harness.rs index ea03100063..7709963a9a 100644 --- a/crates/tracedecay/src/daemon/production_harness.rs +++ b/crates/tracedecay/src/daemon/production_harness.rs @@ -1393,7 +1393,7 @@ async fn shutdown_production_project_harness(mut resources: ProductionProjectHar .await; tracing::Instrument::instrument( shutdown_detached_project_servers( - tokio::time::Instant::now() + tracedecay_runtime_core::DAEMON_SHUTDOWN_DEADLINE, + std::time::Instant::now() + tracedecay_runtime_core::DAEMON_SHUTDOWN_DEADLINE, servers, ), tracing::trace_span!("daemon.harness.shutdown_detached"), diff --git a/crates/tracedecay/src/daemon/project_server_lifecycle.rs b/crates/tracedecay/src/daemon/project_server_lifecycle.rs index 2612c78e10..320d2a4b82 100644 --- a/crates/tracedecay/src/daemon/project_server_lifecycle.rs +++ b/crates/tracedecay/src/daemon/project_server_lifecycle.rs @@ -6,12 +6,14 @@ use super::*; use std::collections::HashSet; +use std::time::Instant; use tracedecay_agent_hosts::agents::context_scout::owner::unregister_registered_context_scout_owner; use tracedecay_agent_hosts::hooks::hook_project_id_for_layout; use tracedecay_daemon_identity::authority; use tracedecay_daemon_service::ProfileHostAdmissionBootstrapStatus; use tracedecay_daemon_service::shutdown::ShutdownStatus; use tracedecay_runtime_core::logging::log_daemon_event; +use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; use tracedecay_store_runtime::{ ShutdownTaskOutcome, ShutdownTaskReceipt, join_shutdown_tasks_until, }; @@ -32,7 +34,7 @@ pub(super) async fn cancel_retained_session_history(store_administration: &Store /// those owners; typed failures are replayed into every subsequent receipt. #[tracing::instrument(name = "daemon.project.lifecycle.shutdown", level = "trace", skip_all)] pub(super) async fn shutdown_project_servers( - deadline: tokio::time::Instant, + deadline: Instant, store_administration: &StoreAdministration, http_application_registry: &super::http_application::DaemonHttpApplicationRegistry, ) -> ShutdownTaskReceipt { @@ -40,13 +42,23 @@ pub(super) async fn shutdown_project_servers( .drain_project_routes_for_shutdown() .await; let (detached, mut receipt) = - match tokio::time::timeout_at(deadline, detach_project_servers(store_administration)).await + match native_runtime::timeout_at(deadline, detach_project_servers(store_administration)) + .await { Ok(servers) => (servers, ShutdownTaskReceipt::default()), - Err(_) => ( + Err(TimeoutError::Elapsed) => ( Vec::new(), ShutdownTaskReceipt::timed_out("project_server_detach"), ), + Err(error @ TimeoutError::TimerUnavailable(_)) => ( + Vec::new(), + ShutdownTaskReceipt { + outcomes: vec![ShutdownTaskOutcome { + owner: "project_server_detach".to_owned(), + status: ShutdownStatus::Failed(error.to_string()), + }], + }, + ), }; let mut retained = { let mut owners = store_administration @@ -156,7 +168,7 @@ pub(super) async fn detach_project_servers( skip_all )] pub(super) async fn shutdown_detached_project_servers( - deadline: tokio::time::Instant, + deadline: Instant, servers: Vec>, ) -> ShutdownTaskReceipt { join_shutdown_tasks_until( @@ -199,30 +211,30 @@ pub(in crate::daemon) async fn retire_project_servers( servers: Vec>, route_registered: Option>, ) { - if tokio::time::timeout( + if let Err(error) = native_runtime::timeout( PROJECT_SERVER_REQUEST_DRAIN_DEADLINE, wait_for_project_server_request_drains(&servers), ) .await - .is_err() { tracing::warn!( deadline_secs = PROJECT_SERVER_REQUEST_DRAIN_DEADLINE.as_secs(), + %error, server_count = servers.len(), "retired project requests exceeded their drain deadline; cancelling them" ); for server in &servers { server.abort_project_server_requests(); } - if tokio::time::timeout( + if let Err(error) = native_runtime::timeout( PROJECT_SERVER_ABORT_DRAIN_DEADLINE, wait_for_project_server_request_drains(&servers), ) .await - .is_err() { tracing::warn!( deadline_secs = PROJECT_SERVER_ABORT_DRAIN_DEADLINE.as_secs(), + %error, server_count = servers.len(), "cancelled project requests have not yielded; retaining safe shutdown ownership" ); @@ -439,7 +451,7 @@ mod shutdown_owner_tests { for _ in 0..2 { let receipt = shutdown_project_servers( - tokio::time::Instant::now() + Duration::from_secs(5), + Instant::now() + Duration::from_secs(5), &administration, &super::super::http_application::DaemonHttpApplicationRegistry::default(), ) diff --git a/crates/tracedecay/src/daemon/project_server_lifecycle_hops.rs b/crates/tracedecay/src/daemon/project_server_lifecycle_hops.rs deleted file mode 100644 index 23d8c882fc..0000000000 --- a/crates/tracedecay/src/daemon/project_server_lifecycle_hops.rs +++ /dev/null @@ -1,60 +0,0 @@ -//! The `project_server_lifecycle` entry points that native code awaits: -//! bootstrap, the connection handshake and retained profile reads. That -//! module still polls Tokio timers, so each call here runs with the daemon's -//! Tokio runtime entered, in the caller's own task: the connection admission -//! scope stays in reach, and dropping the caller drops the call at once. Its -//! other entry points are awaited only inside Tokio tasks (project open, -//! retirements and the shutdown runner). - -use tracedecay_daemon_protocol::DaemonClientIdentity; -use tracedecay_daemon_service::ProfileHostAdmissionBootstrapStatus; -use tracedecay_domain::errors::{Result, TraceDecayError}; -use tracedecay_runtime_core::native_runtime; - -use super::{StoreAdministration, project_server_lifecycle}; - -/// A miss is counted and logged by `tokio_handle`; history is left retained. -pub(super) async fn cancel_retained_session_history(administration: &StoreAdministration) { - if let Some(cancel) = native_runtime::on_tokio( - "daemon.lifecycle.cancel_retained_session_history", - project_server_lifecycle::cancel_retained_session_history(administration), - ) { - cancel.await; - } -} - -/// `None` when no Tokio runtime ran the scheduling; that miss is counted. -pub(super) async fn schedule_user_profile_host_admission_replay_for_identity( - administration: &StoreAdministration, - identity: &DaemonClientIdentity, -) -> Option { - native_runtime::on_tokio( - "daemon.lifecycle.schedule_profile_host_admission_replay", - project_server_lifecycle::schedule_user_profile_host_admission_replay_for_identity( - administration, - identity, - ), - )? - .await -} - -pub(super) async fn await_user_profile_host_admission_replay_for_identity( - administration: &StoreAdministration, - identity: &DaemonClientIdentity, -) -> Result<()> { - native_runtime::on_tokio( - "daemon.lifecycle.await_profile_host_admission_replay", - project_server_lifecycle::await_user_profile_host_admission_replay_for_identity( - administration, - identity, - ), - ) - .ok_or_else(|| { - TraceDecayError::project_route( - "tokio_runtime_unavailable", - false, - "the daemon has no Tokio runtime for the profile host admission replay", - ) - })? - .await -} diff --git a/crates/tracedecay/src/daemon/tests/bootstrap.rs b/crates/tracedecay/src/daemon/tests/bootstrap.rs index fd3fe4e4c5..a08b73b9aa 100644 --- a/crates/tracedecay/src/daemon/tests/bootstrap.rs +++ b/crates/tracedecay/src/daemon/tests/bootstrap.rs @@ -3271,7 +3271,7 @@ async fn portable_broker_bootstrap_bypasses_project_writer_gate() { "portable initialize warmup must singleflight one project open" ); let receipt = super::super::shutdown_project_servers( - tokio::time::Instant::now() + tracedecay_runtime_core::DAEMON_SHUTDOWN_DEADLINE, + std::time::Instant::now() + tracedecay_runtime_core::DAEMON_SHUTDOWN_DEADLINE, &store_administration, &super::super::http_application::DaemonHttpApplicationRegistry::default(), ) diff --git a/crates/tracedecay/src/daemon/tests/multi_root_journey.rs b/crates/tracedecay/src/daemon/tests/multi_root_journey.rs index eebf95bffc..d77351d0b9 100644 --- a/crates/tracedecay/src/daemon/tests/multi_root_journey.rs +++ b/crates/tracedecay/src/daemon/tests/multi_root_journey.rs @@ -13,17 +13,20 @@ use tracedecay_contracts::{ MultiRootScopeSetReadRequestV1, RegisteredRootSelectorV1, }; use tracedecay_domain::{ScopeSetId, UtcMicros}; +use tracedecay_runtime_core::native_runtime::{ + NativeTask, NativeTaskOutcome, Runtime, in_tokio_context, +}; use super::{ enter_test_daemon_database_scope, test_client_identity_for, test_daemon_engine_for_profile, test_handshake_defaults, }; use crate::daemon::{ - DaemonHandshake, execute_daemon_invocation, execute_portable_daemon_invocation, + DaemonEngine, DaemonHandshake, execute_daemon_invocation, execute_portable_daemon_invocation, }; use tracedecay_daemon_service::{ DaemonInvocationOutcome, DaemonInvocationPayload, DaemonInvocationProblem, - DaemonInvocationRequest, parse_daemon_invocation_request, + DaemonInvocationRequest, DaemonInvocationResponse, parse_daemon_invocation_request, }; fn repository() -> TempDir { @@ -90,6 +93,36 @@ fn wire_round_trip(request: &DaemonInvocationRequest) -> DaemonInvocationRequest .expect("valid daemon invocation envelope") } +async fn execute_native_daemon_invocation( + engine: &DaemonEngine, + handshake: &DaemonHandshake, + request: DaemonInvocationRequest, +) -> DaemonInvocationResponse { + let engine = engine.clone(); + let handshake = handshake.clone(); + let tokio_runtime = tokio::runtime::Handle::current(); + let task = NativeTask::spawn(in_tokio_context(tokio_runtime, async move { + assert!( + Runtime::current_handle().is_some(), + "daemon invocation must be polled on the native runtime" + ); + Box::pin(execute_daemon_invocation(&engine, &handshake, request)).await + })) + .expect("native daemon invocation"); + match task.wait().await { + NativeTaskOutcome::Completed(response) => response, + outcome => panic!("native daemon invocation did not settle: {outcome:?}"), + } +} + +fn scope_set_native_reader_waits() -> u64 { + tracedecay_rusqlite_runtime::exact_sql::native_thread_storage_wait_sites() + .into_iter() + .filter(|site| site.kind == "reader" && site.call_site.contains("repository/scope_set.rs:")) + .map(|site| site.hits) + .sum() +} + #[cfg(unix)] #[test] fn authenticated_multi_root_journey_reaches_scope_set_storage() { @@ -253,7 +286,8 @@ async fn run_scope_set_cas_with_registered_cold_root() { ]; let observed_at = now(); let (deadline, cancellation) = controls("cold-roots-cas", observed_at); - let cas = execute_daemon_invocation( + let reader_waits = scope_set_native_reader_waits(); + let cas = execute_native_daemon_invocation( &engine, &handshake, DaemonInvocationRequest::multi_root_scope_set_compare_and_swap( @@ -284,7 +318,7 @@ async fn run_scope_set_cas_with_registered_cold_root() { let observed_at = now(); let (deadline, cancellation) = controls("cold-roots-read", observed_at); - let read = execute_daemon_invocation( + let read = execute_native_daemon_invocation( &engine, &handshake, DaemonInvocationRequest::multi_root_scope_set_read( @@ -306,7 +340,7 @@ async fn run_scope_set_cas_with_registered_cold_root() { let observed_at = now(); let (deadline, cancellation) = controls("cold-roots-stale-cas", observed_at); - let stale = execute_daemon_invocation( + let stale = execute_native_daemon_invocation( &engine, &handshake, DaemonInvocationRequest::multi_root_scope_set_compare_and_swap( @@ -329,6 +363,13 @@ async fn run_scope_set_cas_with_registered_cold_root() { let conflict = packet.payload.expect("stale CAS evidence payload"); assert_eq!(conflict.status, MultiRootScopeSetCasStatusV1::Conflict); assert_eq!(conflict.scope_set.as_ref(), Some(&applied)); + assert_eq!( + scope_set_native_reader_waits(), + reader_waits, + "scope-set admission and replay reads must not park native workers" + ); + drop(registry); + engine.shutdown_all().await; } #[test]