refactor(daemon): Part of #3736: T2-S6 open and serving spawns native

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),
     }
 }
 

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<RemoteCredentialAuthorityRecordV1, RemoteCredentialLookupErrorV1> {
-        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<OneCredentialAuthority> {
+    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<OneCredentialAuthority> {
     let scope = repository_scope("snapshot.remote");
     let enrollment = EnrollmentCredentialRecordV1 {
         enrollment_id: id::<EntityId>("enrollment.remote"),
@@ -221,7 +232,7 @@ fn credential_authority() -> RemoteCredentialAdmissionServiceV1<OneCredentialAut
         grant_id: id::<EntityId>("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<OneCredentialAut
         Deadline::new(UtcMicros(100)).unwrap(),
     )
     .unwrap();
+    if class == RemoteCredentialClassV1::EnrollmentGrant {
+        return RemoteCredentialAdmissionServiceV1::new(OneCredentialAuthority {
+            class,
+            fingerprint: grant.fingerprint.clone(),
+            record: RemoteCredentialAuthorityRecordV1::Grant {
+                grant: Box::new(grant),
+                admission: Box::new(admission),
+            },
+        });
+    }
     let receipt = RemoteEnrollmentCommitReceiptV1 {
         admission,
         prior_grant_digest: grant_digest,
@@ -281,6 +302,7 @@ fn credential_authority() -> RemoteCredentialAdmissionServiceV1<OneCredentialAut
         receipt: Box::new(receipt),
     };
     RemoteCredentialAdmissionServiceV1::new(OneCredentialAuthority {
+        class,
         fingerprint: RemoteCredentialFingerprintV1::from_secret(ACTIVE_CREDENTIAL).unwrap(),
         record,
     })
@@ -696,3 +718,102 @@ fn block_on<F: Future>(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<std::sync::Mutex<Option<bool>>>,
+}
+
+impl RemoteEnrollmentProtocolPortV1 for EnrollmentThreadProbe {
+    fn execute_enrollment(
+        &self,
+        request: RemoteProtocolRequestV1<EnrollmentRequestV1>,
+        _grant_credential: OpaqueRemoteCredential,
+        _enrollment_credential: OpaqueRemoteCredential,
+    ) -> Result<
+        RemoteProtocolResponseV1<EnrollmentCredentialRecordV1>,
+        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::<BrainId>("brain.remote"),
+            id::<BrainNodeId>("node.remote"),
+            fixed_remote_clock(),
+            EnrollmentRequestV1 {
+                grant_id: id::<EntityId>("grant.remote"),
+                grant_revision: 1,
+                enrollment_id: id::<EntityId>("enrollment.remote"),
+                brain_id: id::<BrainId>("brain.remote"),
+                node_id: id::<BrainNodeId>("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"
+    );
+}

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<native_runtime::NativeTask<()>>);
+
+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,

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<StoreAdministration> {
     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 {