From a3ac18a5fbd93ffc06a8776042e8921109452f99 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 01:36:09 +0000 Subject: [PATCH 1/3] refactor(api): run remote enrollment on the native blocking pool --- crates/tracedecay-api/src/remote.rs | 23 ++-- crates/tracedecay-api/src/remote_tests.rs | 125 +++++++++++++++++++++- 2 files changed, 137 insertions(+), 11 deletions(-) diff --git a/crates/tracedecay-api/src/remote.rs b/crates/tracedecay-api/src/remote.rs index 8d7b321daa..f5907a5963 100644 --- a/crates/tracedecay-api/src/remote.rs +++ b/crates/tracedecay-api/src/remote.rs @@ -450,19 +450,24 @@ where request } }; - let match_result = { - let _span = tracing::trace_span!("api.http.handler").entered(); - { - state.service.execute_enrollment( + // The enrollment commit waits on the store writer; a native worker must + // not park on it, so it runs on the blocking pool as `protocol_route` does. + let service = Arc::clone(&state.service); + let execution = tracing::Instrument::instrument( + tracedecay_rusqlite_runtime::native_runtime::spawn_blocking(move || { + service.execute_enrollment( request.request, admission.grant_credential, admission.enrollment_credential, ) - } - }; - match match_result { - Ok(response) => Ok(remote_protocol_response(response.into())), - Err(error) => Err(RemoteHttpRejection::Contract(error)), + }), + tracing::trace_span!("api.http.handler"), + ) + .await; + match execution { + Ok(Ok(response)) => Ok(remote_protocol_response(response.into())), + Ok(Err(error)) => Err(RemoteHttpRejection::Contract(error)), + Err(_) => invalid_remote_request_response().map_err(RemoteHttpRejection::Contract), } } diff --git a/crates/tracedecay-api/src/remote_tests.rs b/crates/tracedecay-api/src/remote_tests.rs index d000c283e2..cbb0deb374 100644 --- a/crates/tracedecay-api/src/remote_tests.rs +++ b/crates/tracedecay-api/src/remote_tests.rs @@ -59,6 +59,7 @@ struct RejectingCredentialAdmission { } struct OneCredentialAuthority { + class: RemoteCredentialClassV1, fingerprint: RemoteCredentialFingerprintV1, record: RemoteCredentialAuthorityRecordV1, } @@ -69,7 +70,7 @@ impl RemoteCredentialLookupPortV1 for OneCredentialAuthority { class: RemoteCredentialClassV1, fingerprint: &RemoteCredentialFingerprintV1, ) -> Result { - if class == RemoteCredentialClassV1::Enrollment && fingerprint == &self.fingerprint { + if class == self.class && fingerprint == &self.fingerprint { return Ok(self.record.clone()); } Err(RemoteCredentialLookupErrorV1::NotFound) @@ -200,6 +201,16 @@ fn repository_scope(snapshot_id: &str) -> RemoteRepositoryScopeV1 { } fn credential_authority() -> RemoteCredentialAdmissionServiceV1 { + credential_authority_for(RemoteCredentialClassV1::Enrollment) +} + +const GRANT_CREDENTIAL: &[u8; 32] = b"fedcba9876543210fedcba9876543210"; + +/// Admits [`ACTIVE_CREDENTIAL`] as the enrollment, or [`GRANT_CREDENTIAL`] as +/// the grant it was enrolled from. +fn credential_authority_for( + class: RemoteCredentialClassV1, +) -> RemoteCredentialAdmissionServiceV1 { let scope = repository_scope("snapshot.remote"); let enrollment = EnrollmentCredentialRecordV1 { enrollment_id: id::("enrollment.remote"), @@ -221,7 +232,7 @@ fn credential_authority() -> RemoteCredentialAdmissionServiceV1("grant.remote"), brain_id: enrollment.brain_id.clone(), node_id: enrollment.node_id.clone(), - fingerprint: RemoteCredentialFingerprintV1::from_secret(&[3_u8; 32]).unwrap(), + fingerprint: RemoteCredentialFingerprintV1::from_secret(GRANT_CREDENTIAL).unwrap(), revision: 1, issued_at: UtcMicros(1), expires_at: UtcMicros(100), @@ -262,6 +273,16 @@ fn credential_authority() -> RemoteCredentialAdmissionServiceV1 RemoteCredentialAdmissionServiceV1(future: F) -> F::Output { } } } + +/// Records whether its enrollment ran on a native runtime thread, where the +/// enrollment commit's store-writer wait would park that thread. +struct EnrollmentThreadProbe { + ran_on_native_thread: Arc>>, +} + +impl RemoteEnrollmentProtocolPortV1 for EnrollmentThreadProbe { + fn execute_enrollment( + &self, + request: RemoteProtocolRequestV1, + _grant_credential: OpaqueRemoteCredential, + _enrollment_credential: OpaqueRemoteCredential, + ) -> Result< + RemoteProtocolResponseV1, + tracedecay_contracts::ApplicationContractError, + > { + // `block_on` refuses exactly on native runtime threads. + let native_thread = + tracedecay_rusqlite_runtime::native_runtime::block_on(async {}).is_err(); + *self.ran_on_native_thread.lock().unwrap() = Some(native_thread); + unavailable_response(request) + } +} + +/// The enrollment route runs its port on the native blocking pool, as every +/// operation route does, so a native worker never waits on the store writer. +#[test] +fn enrollment_route_runs_its_port_off_native_runtime_threads() { + let ran_on_native_thread = Arc::new(std::sync::Mutex::new(None)); + let operations = Arc::new(UnreachedProtocolPort { + calls: Arc::new(AtomicUsize::new(0)), + controlled_deadline: None, + }); + let router = remote_protocol_router( + Arc::new(EnrollmentThreadProbe { + ran_on_native_thread: Arc::clone(&ran_on_native_thread), + }), + RemoteOperationProtocolPortsV1 { + capture: operations.clone(), + replay: operations.clone(), + replay_pending: operations.clone(), + frame_transfer: operations.clone(), + query: operations.clone(), + promotion: operations.clone(), + authority: operations, + }, + Arc::new(credential_authority_for( + RemoteCredentialClassV1::EnrollmentGrant, + )), + fixed_remote_clock, + ); + let scope = repository_scope("snapshot.remote"); + let body = serde_json::to_vec(&RemoteHttpRequestV1 { + request: RemoteProtocolRequestV1::new_initial_enrollment( + RequestId::new("request.remote.enrollment").unwrap(), + id::("brain.remote"), + id::("node.remote"), + fixed_remote_clock(), + EnrollmentRequestV1 { + grant_id: id::("grant.remote"), + grant_revision: 1, + enrollment_id: id::("enrollment.remote"), + brain_id: id::("brain.remote"), + node_id: id::("node.remote"), + expires_at: UtcMicros(100), + capabilities: std::collections::BTreeSet::from([RemoteCapabilityV1::Query]), + scope, + }, + ) + .unwrap(), + }) + .unwrap(); + let request = Request::builder() + .method("POST") + .uri("/enrollment") + .header( + AUTHORIZATION, + [b"Bearer ".as_slice(), GRANT_CREDENTIAL].concat(), + ) + .header( + REMOTE_ENROLLMENT_CREDENTIAL_HEADER, + ACTIVE_CREDENTIAL.as_slice(), + ) + .header("content-type", "application/json") + .body(Body::from(body)) + .unwrap(); + + let response = tracedecay_rusqlite_runtime::native_runtime::block_on(router.oneshot(request)) + .expect("native runtime") + .unwrap(); + + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!( + *ran_on_native_thread.lock().unwrap(), + Some(false), + "the enrollment port must run off native runtime threads" + ); +} From 359c6e6f67e0afbe66ef64e3d643d4b60afda65e Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 01:36:16 +0000 Subject: [PATCH 2/3] refactor(daemon): serve Remote Brain TLS connections natively --- .../src/remote_http_transport.rs | 82 ++++++++---- .../tracedecay/src/daemon/http_application.rs | 30 +++-- .../http_application_tests/remote_tls.rs | 123 ++++++++++++++++++ 3 files changed, 200 insertions(+), 35 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/remote_http_transport.rs b/crates/tracedecay-daemon-service/src/remote_http_transport.rs index f985904b9f..44ff9145d4 100644 --- a/crates/tracedecay-daemon-service/src/remote_http_transport.rs +++ b/crates/tracedecay-daemon-service/src/remote_http_transport.rs @@ -28,7 +28,6 @@ use rustls::pki_types::pem::PemObject; use rustls::pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime}; use rustls::server::ParsedCertificate; use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; -use tokio::task::JoinSet; use tokio_util::sync::CancellationToken; use tracedecay_daemon_control::RemoteBrainTlsConfig; use tracedecay_dashboard_api::serve_http_connection; @@ -162,44 +161,38 @@ impl RemoteBrainTlsListener { ) -> Result<()> { let router = router.layer(middleware::from_fn(force_remote_connection_close)); let graceful = CancellationToken::new(); - let mut connections = JoinSet::new(); + let mut connections = RemoteBrainTlsConnections::default(); loop { tokio::select! { biased; _ = &mut shutdown_requested => break, - joined = connections.join_next(), if !connections.is_empty() => { - if let Some(joined) = joined { - observe_remote_brain_connection_join(joined); - } - } accepted = self.accept() => { if let Some((io, address)) = accepted { + connections.reap_finished(); let router = router.clone(); let graceful = graceful.clone(); - connections.spawn(tracing::Instrument::instrument(async move { + let connection = native_runtime::NativeTask::spawn(tracing::Instrument::instrument(async move { if let Err(error) = serve_http_connection(io, router, false, graceful).await { tracing::debug!(%error, %address, "Remote Brain TLS connection stopped"); } }, tracing::trace_span!("daemon.http.application.remote_tls_connection"))); + match connection { + Ok(connection) => connections.0.push(connection), + // The refused connection's IO drops here, closing it. + Err(error) => tracing::warn!(%error, %address, "Remote Brain TLS connection could not start"), + } } } } } graceful.cancel(); - let drain = async { - while let Some(joined) = connections.join_next().await { - observe_remote_brain_connection_join(joined); - } - }; - if native_runtime::timeout(self.deadlines.shutdown_drain, drain) + if native_runtime::timeout(self.deadlines.shutdown_drain, connections.join_all()) .await .is_err() { connections.abort_all(); - while let Some(joined) = connections.join_next().await { - observe_remote_brain_connection_join(joined); - } + connections.join_all().await; } Ok(()) } @@ -358,21 +351,56 @@ pub fn validate_remote_brain_tls_identity_at( ) } -fn observe_remote_brain_connection_join(joined: std::result::Result<(), tokio::task::JoinError>) { - if let Err(error) = joined - && !error.is_cancelled() - { - tracing::warn!(%error, "Remote Brain TLS connection task failed"); +/// The listener's open connections. Dropping the set aborts them. +#[derive(Default)] +struct RemoteBrainTlsConnections(Vec>); + +impl RemoteBrainTlsConnections { + fn reap_finished(&mut self) { + self.0.retain(|connection| match connection.try_finished() { + Some(outcome) => { + observe_remote_brain_connection_end(outcome); + false + } + None => true, + }); + } + + /// Keeps every connection in the set until all have ended, so a cancelled + /// join leaves them for [`Self::abort_all`]. + async fn join_all(&mut self) { + for connection in &self.0 { + observe_remote_brain_connection_end(connection.wait().await); + } + self.0.clear(); + } + + fn abort_all(&self) { + for connection in &self.0 { + connection.abort(); + } + } +} + +impl Drop for RemoteBrainTlsConnections { + fn drop(&mut self) { + self.abort_all(); + } +} + +fn observe_remote_brain_connection_end(outcome: native_runtime::NativeTaskOutcome<()>) { + if let native_runtime::NativeTaskOutcome::Panicked(panic) = outcome { + tracing::warn!(%panic, "Remote Brain TLS connection task failed"); } } /// A resettable connection deadline that reports expiry on every poll. /// -/// The connection task runs on Tokio, so this polls an Asupersync `Sleep` -/// through Asupersync's process-global fallback timer. It is a documented -/// exception to `native_runtime::sleep_until`: the read and write deadlines -/// reset on every transferred chunk, and a native task per reset would cost -/// tens of microseconds each. `Sleep` panics when polled after it completes, +/// The connection task runs on the native runtime and polls an Asupersync +/// `Sleep` directly. It is a documented exception to +/// `native_runtime::sleep_until`: the read and write deadlines reset on every +/// transferred chunk, and a native task per reset would cost tens of +/// microseconds each. `Sleep` panics when polled after it completes, /// so expiry is latched here until the next reset. struct IoDeadline { sleep: Sleep, diff --git a/crates/tracedecay/src/daemon/http_application.rs b/crates/tracedecay/src/daemon/http_application.rs index 59b133eaf3..2643590851 100644 --- a/crates/tracedecay/src/daemon/http_application.rs +++ b/crates/tracedecay/src/daemon/http_application.rs @@ -782,7 +782,7 @@ pub(super) struct DaemonHttpApplicationService { shutdown: Option>, task: Option>, remote_tls_shutdown: Option>, - remote_tls_task: Option>>, + remote_tls_task: Option>>, #[cfg(test)] remote_tls_admission: Option>, #[cfg(test)] @@ -919,7 +919,8 @@ impl DaemonHttpApplicationService { remote_tls_egress = Some(server.egress); } let listener_state = server.credentials; - remote_tls_task = Some(tokio::spawn(tracing::Instrument::instrument( + // A `NativeTask` output must be `Clone`, so the error travels as text. + let task = native_runtime::NativeTask::spawn(tracing::Instrument::instrument( async move { let result = server .listener @@ -929,10 +930,14 @@ impl DaemonHttpApplicationService { Ok(()) => listener_state.publish_listener_stopped(), Err(_) => listener_state.publish_listener_degraded(), } - result + result.map_err(|error| error.to_string()) }, tracing::trace_span!("daemon.http.application.remote_tls_listener"), - ))); + )) + .map_err(|error| TraceDecayError::Config { + message: format!("Remote Brain TLS listener task could not start: {error}"), + })?; + remote_tls_task = Some(task); } Ok(Self { endpoint, @@ -1008,12 +1013,21 @@ impl DaemonHttpApplicationService { task.await.map_err(|error| TraceDecayError::Config { message: format!("daemon HTTP application service task failed: {error}"), })?; - let Some(task) = self.remote_tls_task.as_mut() else { + let Some(task) = self.remote_tls_task.as_ref() else { return Ok(()); }; - task.await.map_err(|error| TraceDecayError::Config { - message: format!("Remote Brain TLS application service task failed: {error}"), - })? + let failure = match task.wait().await { + native_runtime::NativeTaskOutcome::Completed(result) => result.err(), + native_runtime::NativeTaskOutcome::Cancelled => Some("task cancelled".to_owned()), + native_runtime::NativeTaskOutcome::Panicked(panic) => { + Some(format!("task panicked: {panic}")) + } + }; + failure.map_or(Ok(()), |failure| { + Err(TraceDecayError::Config { + message: format!("Remote Brain TLS application service task failed: {failure}"), + }) + }) } } diff --git a/crates/tracedecay/src/daemon/http_application_tests/remote_tls.rs b/crates/tracedecay/src/daemon/http_application_tests/remote_tls.rs index 60163e8087..e0c2d7f33a 100644 --- a/crates/tracedecay/src/daemon/http_application_tests/remote_tls.rs +++ b/crates/tracedecay/src/daemon/http_application_tests/remote_tls.rs @@ -1540,3 +1540,126 @@ async fn remote_tls_listener_expires_saturated_non_reading_responses() { .expect("egress saturation must not delay shutdown") .expect("shutdown TLS egress service"); } + +/// The listener serves each connection on the native runtime, so a remote +/// handler runs with no Tokio runtime entered. +#[tokio::test] +async fn remote_tls_connections_are_served_off_tokio() { + let temporary = tempfile::tempdir().expect("remote TLS runtime fixture"); + let (tls, certificate) = remote_tls_fixture(&temporary); + let credentials = Arc::new( + tracedecay_store_runtime::DaemonRemoteCredentialAuthorityV1::new( + BrainId::new("brain.remote-tls-runtime").expect("remote brain identity"), + UserProfileId::new("profile.remote-tls-runtime").expect("remote profile identity"), + ), + ); + let router = Router::new().route( + "/runtime", + axum::routing::post(|| async { + if tokio::runtime::Handle::try_current().is_err() { + "native" + } else { + "tokio" + } + }), + ); + let registry = DaemonHttpApplicationRegistry::default(); + registry + .install_remote(router, credentials, None) + .expect("install runtime-probe remote router"); + let hazards = crate::test_support::native_runtime_hazards(); + let service = + DaemonHttpApplicationService::bind_with_remote_tls(registry, AUTH_TOKEN, Some(&tls)) + .await + .expect("bind Remote Brain TLS runtime-probe service"); + + let response = remote_tls_request( + service.remote_tls_endpoint().expect("TLS endpoint"), + &certificate, + "/remote/runtime", + None, + None, + b"", + ) + .await; + + assert!(response.starts_with("HTTP/1.1 200"), "{response}"); + assert!( + response.ends_with("native"), + "the remote handler must run off Tokio: {response}" + ); + service + .shutdown() + .await + .expect("shutdown TLS runtime-probe service"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); +} + +struct HandlerDropReceipt(Option>); + +impl Drop for HandlerDropReceipt { + fn drop(&mut self) { + if let Some(receipt) = self.0.take() { + let _ = receipt.send(()); + } + } +} + +/// Dropping the service aborts its listener, and the listener's connections +/// with it, without waiting out the shutdown drain. +#[tokio::test] +async fn dropping_the_remote_tls_service_aborts_an_open_connection() { + let temporary = tempfile::tempdir().expect("remote TLS dropped-service fixture"); + let (tls, certificate) = remote_tls_fixture(&temporary); + let credentials = Arc::new( + tracedecay_store_runtime::DaemonRemoteCredentialAuthorityV1::new( + BrainId::new("brain.remote-tls-dropped").expect("remote brain identity"), + UserProfileId::new("profile.remote-tls-dropped").expect("remote profile identity"), + ), + ); + let (dropped, handler_dropped) = tokio::sync::oneshot::channel(); + let dropped = Arc::new(std::sync::Mutex::new(Some(dropped))); + let started = Arc::new(tokio::sync::Notify::new()); + let handler_started = Arc::clone(&started); + let router = Router::new().route( + "/held", + get(move || { + let receipt = HandlerDropReceipt(dropped.lock().expect("drop receipt").take()); + let handler_started = Arc::clone(&handler_started); + async move { + let _receipt = receipt; + handler_started.notify_one(); + std::future::pending::<&'static str>().await + } + }), + ); + let registry = DaemonHttpApplicationRegistry::default(); + registry + .install_remote(router, credentials, None) + .expect("install held-response remote router"); + let service = + DaemonHttpApplicationService::bind_with_remote_tls(registry, AUTH_TOKEN, Some(&tls)) + .await + .expect("bind Remote Brain TLS dropped-service service"); + let mut peer = remote_tls_connect( + service.remote_tls_endpoint().expect("TLS endpoint"), + &certificate, + ) + .await; + peer.write_all(b"GET /remote/held HTTP/1.1\r\nHost: localhost\r\n\r\n") + .await + .expect("write held request"); + tokio::time::timeout(Duration::from_secs(5), started.notified()) + .await + .expect("held handler must start"); + + drop(service); + + tokio::time::timeout( + RemoteBrainTlsDeadlines::PRODUCTION.shutdown_drain / 2, + handler_dropped, + ) + .await + .expect("dropping the service must abort its open connection") + .expect("handler drop receipt"); +} From 4bf25ac3c14c44248221311c2ae01fc158f5ca4b Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Thu, 15 Oct 2026 01:36:16 +0000 Subject: [PATCH 3/3] refactor(daemon): run open-time mount, activation and binding natively --- .../src/daemon/connection_serving.rs | 29 +++--- .../project_composition/full_owner_mount.rs | 53 ++++++++--- .../src/daemon/project_open_orchestration.rs | 12 ++- .../tracedecay/src/daemon/tests/bootstrap.rs | 25 +++-- .../src/daemon/tests/runtime_identity.rs | 34 +++++++ .../src/daemon/tests/scheduler_config.rs | 94 +++++++++++++++++++ 6 files changed, 216 insertions(+), 31 deletions(-) diff --git a/crates/tracedecay/src/daemon/connection_serving.rs b/crates/tracedecay/src/daemon/connection_serving.rs index 144ed54afb..6342d83a6c 100644 --- a/crates/tracedecay/src/daemon/connection_serving.rs +++ b/crates/tracedecay/src/daemon/connection_serving.rs @@ -270,28 +270,35 @@ fn profile_identity_warming_error() -> TraceDecayError { /// dropping the initializer future would abandon the partially finished open /// and make the next client start over. Detaching it instead lets this client's /// retry, or the next one, find the registry already warm. -async fn bind_authenticated_profile_identity_within_deadline( +pub(super) async fn bind_authenticated_profile_identity_within_deadline( handshake: &mut DaemonHandshake, store_administration: &StoreAdministration, ) -> Result { let mut binding_handshake = handshake.clone(); let binding_administration = store_administration.clone(); - let binding = tokio::spawn(async move { - Box::pin(bind_authenticated_profile_identity( - &mut binding_handshake, - &binding_administration, - )) - .await - .map(|administration| (binding_handshake, administration)) - }); + // A `NativeTask` output must be `Clone`; the binding's error is not. + let (bound, binding) = tokio::sync::oneshot::channel(); + native_runtime::NativeTask::spawn(async move { + let _ = bound.send( + Box::pin(bind_authenticated_profile_identity( + &mut binding_handshake, + &binding_administration, + )) + .await + .map(|administration| (binding_handshake, administration)), + ); + }) + .map_err(|error| TraceDecayError::Config { + message: format!("daemon profile identity binding could not start: {error}"), + })?; match native_runtime::timeout(PROFILE_IDENTITY_BIND_DEADLINE, binding).await { Ok(Ok(Ok((bound_handshake, administration)))) => { *handshake = bound_handshake; Ok(administration) } Ok(Ok(Err(error))) => Err(error), - Ok(Err(join)) => Err(TraceDecayError::Config { - message: format!("daemon profile identity binding failed to join: {join}"), + Ok(Err(_)) => Err(TraceDecayError::Config { + message: "daemon profile identity binding ended without a result".to_owned(), }), Err(TimeoutError::Elapsed) => Err(profile_identity_warming_error()), Err(error @ TimeoutError::TimerUnavailable(_)) => Err(TraceDecayError::Config { diff --git a/crates/tracedecay/src/daemon/project_composition/full_owner_mount.rs b/crates/tracedecay/src/daemon/project_composition/full_owner_mount.rs index 9daae3014f..2e97ce592a 100644 --- a/crates/tracedecay/src/daemon/project_composition/full_owner_mount.rs +++ b/crates/tracedecay/src/daemon/project_composition/full_owner_mount.rs @@ -1,6 +1,16 @@ use std::future::Future; use tracedecay_domain::errors::{Result, TraceDecayError}; +use tracedecay_runtime_core::native_runtime::NativeTask; + +/// Aborts the HTTP mount when the joined pair is dropped or fails. +struct HttpMountTask(NativeTask<()>); + +impl Drop for HttpMountTask { + fn drop(&mut self) { + self.0.abort(); + } +} /// Spawn HTTP so its sync router build cannot starve owner polls. /// Box both arms so the caller's phase future stays one leaf wide. @@ -23,18 +33,19 @@ where )); tracing::Instrument::instrument( async move { - let mut http_tasks = tokio::task::JoinSet::new(); - http_tasks.spawn(http_mount); + // A `NativeTask` output must be `Clone`; the mount's error is not. + let (mounted, http_mounted) = tokio::sync::oneshot::channel(); + let _http_task = NativeTask::spawn(async move { + let _ = mounted.send(http_mount.await); + }) + .map(HttpMountTask) + .map_err(|error| TraceDecayError::Config { + message: format!("http application mount task could not start: {error}"), + })?; tokio::try_join!(owners_mount, async { - http_tasks - .join_next() - .await - .ok_or_else(|| TraceDecayError::Config { - message: "http application mount task missing".to_owned(), - })? - .map_err(|error| TraceDecayError::Config { - message: format!("http application mount task failed: {error}"), - })? + http_mounted.await.map_err(|_| TraceDecayError::Config { + message: "http application mount task ended without a result".to_owned(), + })? }) }, tracing::trace_span!("daemon.project.compose.join_full_mounts"), @@ -198,4 +209,24 @@ mod tests { .expect("full mount cancellation must cancel the spawned HTTP mount") .expect("HTTP cancellation receipt"); } + + #[test] + fn independent_full_mounts_join_without_a_tokio_runtime() { + let hazards = crate::test_support::native_runtime_hazards(); + let joined = tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + join_independent_full_owner_mounts(async { Ok("owners-ready") }, async { + Ok(tokio::runtime::Handle::try_current().is_err()) + }) + .await + }) + .expect("native runtime runs the full mount") + .expect("admit both full mounts"); + assert_eq!( + joined, + ("owners-ready", true), + "the HTTP mount runs off Tokio" + ); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + } } diff --git a/crates/tracedecay/src/daemon/project_open_orchestration.rs b/crates/tracedecay/src/daemon/project_open_orchestration.rs index 7da8fb21ea..6e3c380587 100644 --- a/crates/tracedecay/src/daemon/project_open_orchestration.rs +++ b/crates/tracedecay/src/daemon/project_open_orchestration.rs @@ -195,7 +195,8 @@ pub(super) fn spawn_lifecycle_automation_scheduler_activation( let Some(activity) = lifecycle.try_enter() else { return; }; - tokio::spawn(tracing::Instrument::instrument( + // Detached: draining ends it, and its activity permit holds idle until then. + let started = native_runtime::NativeTask::spawn(tracing::Instrument::instrument( async move { let _activity = activity; tokio::select! { @@ -206,6 +207,15 @@ pub(super) fn spawn_lifecycle_automation_scheduler_activation( }, tracing::trace_span!("daemon.project.activate.automation"), )); + if let Err(error) = started { + log_daemon_event( + "automation_scheduler_activation", + &[ + ("outcome", "runtime_unavailable".to_string()), + ("error", error), + ], + ); + } } #[tracing::instrument(name = "daemon.project.enroll.route", level = "trace", skip_all)] diff --git a/crates/tracedecay/src/daemon/tests/bootstrap.rs b/crates/tracedecay/src/daemon/tests/bootstrap.rs index 44515e9706..a50b54260b 100644 --- a/crates/tracedecay/src/daemon/tests/bootstrap.rs +++ b/crates/tracedecay/src/daemon/tests/bootstrap.rs @@ -3309,20 +3309,29 @@ async fn project_server_warmup_drops_lifecycle_activity_on_draining() { async fn scheduler_activation_drain_wins_when_discovery_is_simultaneously_ready() { for _ in 0..32 { let lifecycle = DaemonLifecycle::default(); - let discovery_polled = Arc::new(tokio::sync::Notify::new()); - let discovery_polled_by_future = Arc::clone(&discovery_polled); + let discovery_parked = Arc::new(tokio::sync::Notify::new()); + let discovery_parked_by_future = Arc::clone(&discovery_parked); let discovery_lifecycle = lifecycle.clone(); let discovery_won = Arc::new(std::sync::atomic::AtomicBool::new(false)); let discovery_won_by_future = Arc::clone(&discovery_won); + let mut discovery = Box::pin(async move { + discovery_lifecycle.wait_for_draining().await; + discovery_won_by_future.store(true, std::sync::atomic::Ordering::Release); + }); + // The activation runs on a native worker beside this thread. Draining + // only after discovery has returned `Pending` makes both arms ready + // on the activation's next poll instead of racing the current one. super::super::project_open_orchestration::spawn_lifecycle_automation_scheduler_activation( lifecycle.clone(), - async move { - discovery_polled_by_future.notify_one(); - discovery_lifecycle.wait_for_draining().await; - discovery_won_by_future.store(true, std::sync::atomic::Ordering::Release); - }, + std::future::poll_fn(move |context| { + let polled = discovery.as_mut().poll(context); + if polled.is_pending() { + discovery_parked_by_future.notify_one(); + } + polled + }), ); - discovery_polled.notified().await; + discovery_parked.notified().await; lifecycle.begin_draining(); tokio::time::timeout( diff --git a/crates/tracedecay/src/daemon/tests/runtime_identity.rs b/crates/tracedecay/src/daemon/tests/runtime_identity.rs index 1d30f646fc..9ce1fed24f 100644 --- a/crates/tracedecay/src/daemon/tests/runtime_identity.rs +++ b/crates/tracedecay/src/daemon/tests/runtime_identity.rs @@ -985,3 +985,37 @@ async fn retiring_observability_owner_reopens_same_route_without_cached_degradat drop(db); engine.shutdown_all().await; } + +/// A handshake's profile-identity binding runs on the native runtime: with no +/// Tokio runtime it opens the cold profile runtime and binds the client to it. +#[test] +fn profile_identity_binding_opens_a_cold_profile_runtime_without_a_tokio_runtime() { + let temp = TempDir::new().expect("temp dir"); + let profile_root = temp.path().join("profile"); + let store_administration = test_store_administration_for_profile(&profile_root); + let _database_scope = + enter_test_daemon_database_scope(&profile_root, "profile-binding-off-tokio"); + let mut handshake = DaemonHandshake { + client_identity: test_client_identity_for(profile_root.clone()), + ..test_handshake_defaults() + }; + + let hazards = crate::test_support::native_runtime_hazards(); + tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + super::super::connection_serving::bind_authenticated_profile_identity_within_deadline( + &mut handshake, + &store_administration, + ) + .await + }) + .expect("native runtime runs the binding") + .expect("bind the client's profile identity"); + + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + assert_eq!( + handshake.client_identity.profile_root, + profile_root.canonicalize().expect("canonical profile root"), + "the binding returns the canonical bound identity" + ); +} diff --git a/crates/tracedecay/src/daemon/tests/scheduler_config.rs b/crates/tracedecay/src/daemon/tests/scheduler_config.rs index 3bb2775304..fab9edc6f8 100644 --- a/crates/tracedecay/src/daemon/tests/scheduler_config.rs +++ b/crates/tracedecay/src/daemon/tests/scheduler_config.rs @@ -1096,3 +1096,97 @@ fn the_automation_loop_starts_and_exits_without_a_tokio_runtime() { assert_eq!(decision, AutomationSchedulerExitBarrier::EXIT); tokio.block_on(engine.shutdown_all()); } + +/// Project-open activation runs on the native runtime: with no Tokio runtime +/// it reconciles reserved automation effects, probes the project's automation +/// configuration and starts its loop. +#[cfg(unix)] +#[test] +fn project_open_activation_starts_the_automation_loop_without_a_tokio_runtime() { + let tokio = tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build() + .expect("tokio runtime"); + let dir = TempDir::new().expect("temp dir"); + let project = dir.path().join("project"); + let client_identity = test_client_identity_for(dir.path().join("profile")); + std::fs::create_dir_all(project.join("src")).expect("src dir"); + std::fs::write(project.join("src/main.rs"), "fn main() {}\n").expect("source file"); + tokio.block_on(initialize_test_project(&project, &client_identity)); + let handshake = DaemonHandshake { + project_path: Some(project.clone()), + client_identity, + ..test_handshake_defaults() + }; + let engine = test_daemon_engine_for_profile(&handshake.client_identity.profile_root); + let _database_scope = enter_test_daemon_database_scope( + &handshake.client_identity.profile_root, + "activation-off-tokio-test", + ); + let server = tokio + .block_on(engine.project_server(&handshake)) + .expect("open project"); + let cg = tokio.block_on(server.cg()); + let key = ProjectServerKey::from_open_project(&cg, &handshake).expect("owner key"); + // Project open activated its own loop; retire it so the activation below + // starts the one under test. + tokio.block_on(async { + wait_for_automation_scheduler_state(&engine, |schedulers| schedulers.contains_key(&key)) + .await; + if let Some(retirement) = engine.retire_automation_scheduler_locked(&key).await { + retirement.wait().await; + } + engine + .store_administration + .wait_for_retirement_reaper_count_for_test(0) + .await; + }); + let probes = engine + .automation_config_probe_attempts + .load(std::sync::atomic::Ordering::Relaxed); + + let hazards = crate::test_support::native_runtime_hazards(); + tracedecay_runtime_core::native_runtime::block_on(async { + assert!(tokio::runtime::Handle::try_current().is_err()); + super::super::project_open_orchestration::spawn_lifecycle_automation_scheduler_activation( + engine.lifecycle.clone(), + crate::daemon::project_open_owners::reconcile_project_open_automation_effects( + Arc::clone(&cg), + ), + ); + let activation_engine = engine.clone(); + let activation_key = key.clone(); + let activation_cg = Arc::clone(&cg); + super::super::project_open_orchestration::spawn_lifecycle_automation_scheduler_activation( + engine.lifecycle.clone(), + async move { + activation_engine + .activate_automation_scheduler_for_open_project( + activation_key, + project, + handshake, + activation_cg, + ) + .await; + }, + ); + tracedecay_runtime_core::native_runtime::timeout( + std::time::Duration::from_secs(20), + wait_for_automation_scheduler_state(&engine, |schedulers| { + schedulers.contains_key(&key) + }), + ) + .await + .expect("the activation starts the project's automation loop"); + }) + .expect("native runtime"); + assert_eq!(crate::test_support::native_runtime_hazards(), hazards); + assert!( + engine + .automation_config_probe_attempts + .load(std::sync::atomic::Ordering::Relaxed) + > probes, + "the activation probed the automation configuration" + ); + tokio.block_on(engine.shutdown_all()); +}