fix(core): linearize session publication

core/src/abi.rs

@@ -235,6 +235,13 @@ struct SessionState {
     features: Vec<String>,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum SessionPublication {
+    Starting,
+    Published,
+    Failed(CoreError),
+}
+
 #[derive(Clone, Copy, Eq, PartialEq)]
 enum Lifecycle {
     Created,
@@ -895,7 +902,10 @@ unsafe extern "C" fn verse_core_connect_v1(
                 bytes,
             )
         });
-        let runtime = match tokio::runtime::Runtime::new() {
+        let runtime = match tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+        {
             Ok(runtime) => runtime,
             Err(_) => return INTERNAL,
         };
@@ -905,6 +915,8 @@ unsafe extern "C" fn verse_core_connect_v1(
         };
         #[cfg(test)]
         let test_session = credential.certificate_chain_pem().contains("AQID");
+        #[cfg(test)]
+        let test_session_failure = credential.certificate_chain_pem().contains("AQIDFAIL");
         #[cfg(not(test))]
         let test_session = false;
         let session = if test_session {
@@ -951,7 +963,10 @@ unsafe extern "C" fn verse_core_connect_v1(
         let features = manifest.features().to_vec();
         let (command_tx, command_rx) = tokio::sync::mpsc::channel(CONTROL_QUEUE_CAPACITY);
         let (session_done_tx, session_done_rx) = mpsc::channel();
+        let (session_started_tx, session_started_rx) = mpsc::channel();
         let (event_tx, event_rx) = bounded_session_events();
+        let publication = Arc::new(Mutex::new(SessionPublication::Starting));
+        let worker_publication = Arc::clone(&publication);
         let callbacks = handle.callbacks;
         let id = handle.id;
         let callbacks_closed = Arc::clone(&handle.callbacks_closed);
@@ -1000,9 +1015,25 @@ unsafe extern "C" fn verse_core_connect_v1(
                     }
                 });
                 let session_result = if let Some(session) = session {
-                    runtime.block_on(session.run(command_rx, event_tx, cancellation))
+                    runtime.block_on(session.run_with_startup(
+                        command_rx,
+                        event_tx,
+                        cancellation,
+                        Some(session_started_tx),
+                    ))
                 } else {
                     event_tx.close();
+                    #[cfg(test)]
+                    if test_session_failure {
+                        let mut publication = lock(&worker_publication);
+                        *publication = SessionPublication::Failed(CoreError::Transport);
+                        let _ = session_started_tx.send(());
+                        drop(publication);
+                        let _ = bridge.join();
+                        let _ = session_done_tx.send(());
+                        return;
+                    }
+                    let _ = session_started_tx.send(());
                     runtime.block_on(async move {
                         let mut command_rx = command_rx;
                         while !cancellation.is_cancelled() {
@@ -1019,17 +1050,30 @@ unsafe extern "C" fn verse_core_connect_v1(
                 };
                 let _ = bridge.join();
                 if let Err(error) = session_result {
-                    let mut state = lock(&session_handle.state);
+                    let published = {
-                    if state.lifecycle == Lifecycle::Connected
+                        let mut publication = lock(&worker_publication);
-                        && !session_handle.callbacks_closed.load(Ordering::Acquire)
+                        match *publication {
-                    {
+                            SessionPublication::Starting => {
-                        if error != CoreError::Cancelled {
+                                *publication = SessionPublication::Failed(error);
-                            enqueue_callback_under_state_lock(
+                                false
-                                &session_handle,
+                            }
-                                CallbackEvent::Error(status(error)),
+                            SessionPublication::Published => true,
-                            );
+                            SessionPublication::Failed(_) => false,
+                        }
+                    };
+                    if published {
+                        let mut state = lock(&session_handle.state);
+                        if state.lifecycle == Lifecycle::Connected
+                            && !session_handle.callbacks_closed.load(Ordering::Acquire)
+                        {
+                            if error != CoreError::Cancelled {
+                                enqueue_callback_under_state_lock(
+                                    &session_handle,
+                                    CallbackEvent::Error(status(error)),
+                                );
+                            }
+                            apply_cancellation(&session_handle, &mut state);
                         }
-                        apply_cancellation(&session_handle, &mut state);
                     }
                 }
                 let _ = session_done_tx.send(());
@@ -1040,8 +1084,26 @@ unsafe extern "C" fn verse_core_connect_v1(
         };
         *lock(&handle.session_done) = Some(session_done_rx);
         *lock(&handle.session_worker) = Some(session_worker);
+        if session_started_rx
+            .recv_timeout(Duration::from_secs(10))
+            .is_err()
         {
+            cancel_inner(&handle);
+            return INTERNAL;
+        }
+        {
+            let mut publication = lock(&publication);
             let mut state = lock(&handle.state);
+            if let SessionPublication::Failed(error) = *publication {
+                handle
+                    .transition
+                    .store(TRANSITION_CANCELLED, Ordering::Release);
+                if error != CoreError::Cancelled {
+                    enqueue_callback_under_state_lock(&handle, CallbackEvent::Error(status(error)));
+                }
+                apply_cancellation(&handle, &mut state);
+                return status(error);
+            }
             if cancellation_admitted(&handle) || state.lifecycle == Lifecycle::Destroying {
                 apply_cancellation(&handle, &mut state);
                 return CANCELLED;
@@ -1061,6 +1123,7 @@ unsafe extern "C" fn verse_core_connect_v1(
             state.command_tx = Some(command_tx);
             state.features = features;
             state.lifecycle = Lifecycle::Connected;
+            *publication = SessionPublication::Published;
             enqueue_callback_under_state_lock(&handle, CallbackEvent::State(STATE_CONNECTED));
             if cancellation_admitted(&handle) {
                 apply_cancellation(&handle, &mut state);
@@ -1364,6 +1427,7 @@ mod tests {
         ConnectTransitionHook, CoreHandle, FinalConnectHook, Lifecycle, SignFn, StateEvent,
         StateFn, BUSY, CANCELLED, CANCEL_ADMISSION_HOOK, FINAL_CONNECT_HOOK, FINAL_PUBLISH_HOOK,
         INITIAL_CONNECT_HOOK, INTERNAL, OK, STATE_CANCELLED, STATE_CONNECTED, STATE_CONNECTING,
+        TRANSPORT,
     };
 
     const MANIFEST: &[u8] = br#"{
@@ -1684,4 +1748,42 @@ mod tests {
             [STATE_CONNECTING, STATE_CANCELLED]
         );
     }
+
+    #[test]
+    fn terminal_session_failure_before_publication_never_connects() {
+        let context = Box::new(CallbackContext::default());
+        let context_pointer = ptr::from_ref(&*context).cast_mut().cast::<c_void>();
+        let core = create_core_with_callback(context_pointer, Some(record_state));
+        let failing_credential = String::from_utf8(CREDENTIAL.to_vec())
+            .expect("credential UTF-8")
+            .replace("AQID", "AQIDFAIL");
+        let request = ConnectRequest {
+            struct_size: u32::try_from(size_of::<ConnectRequest>()).expect("request size"),
+            abi_version: 1,
+            manifest_json: BytesView {
+                data: MANIFEST.as_ptr(),
+                length: MANIFEST.len(),
+            },
+            tunnel_credential_json: BytesView {
+                data: failing_credential.as_ptr(),
+                length: failing_credential.len(),
+            },
+        };
+
+        assert_eq!(
+            unsafe { super::verse_core_connect_v1(core, &raw const request) },
+            TRANSPORT
+        );
+        wait_for_state(&context, STATE_CANCELLED);
+        assert_eq!(
+            unsafe { super::verse_core_request_idr_v1(core) },
+            CANCELLED,
+            "terminal connect failure must commit the cancelled transition"
+        );
+        assert_eq!(unsafe { super::verse_core_destroy_v1(core, 2_000) }, OK);
+        assert_eq!(
+            context.states.lock().expect("states lock").as_slice(),
+            [STATE_CONNECTING, STATE_CANCELLED]
+        );
+    }
 }

core/src/transport.rs

@@ -331,6 +331,17 @@ impl TransportSession {
         commands: tokio::sync::mpsc::Receiver<SessionCommand>,
         events: SessionEventSender,
         cancellation: Cancellation,
+    ) -> Result<()> {
+        self.run_with_startup(commands, events, cancellation, None)
+            .await
+    }
+
+    pub(crate) async fn run_with_startup(
+        self,
+        commands: tokio::sync::mpsc::Receiver<SessionCommand>,
+        events: SessionEventSender,
+        cancellation: Cancellation,
+        started: Option<std::sync::mpsc::Sender<()>>,
     ) -> Result<()> {
         let Self {
             endpoint,
@@ -358,6 +369,9 @@ impl TransportSession {
             result_tx.clone(),
         ));
         let datagrams = tokio::spawn(datagram_loop(connection.clone(), events.clone(), result_tx));
+        if let Some(started) = started {
+            let _ = started.send(());
+        }
         loop {
             if cancellation.check().is_err() {
                 connection.close(APPLICATION_ERROR, b"cancelled");

core/tests/abi_contract.rs

@@ -742,6 +742,7 @@ fn signer_callback_rejects_every_stateful_api_across_handles() {
     let other = create(&mut other_ctx);
 
     let mut ctx = Context::default();
+    ctx.reentry_cancel.store(OK, Ordering::SeqCst);
     ctx.reentry_target.store(other as usize, Ordering::SeqCst);
     let mut cfg = config(&mut ctx);
     cfg.sign_admission = Some(sign_admission_probes_global_reentry);

core/tests/gateway_oracle.rs

@@ -204,25 +204,52 @@ pub(crate) struct Ready {
 }
 
 pub(crate) struct Oracle {
-    _serial: std::sync::MutexGuard<'static, ()>,
     child: Child,
     directory: std::path::PathBuf,
     pub(crate) ready: Ready,
 }
 
+#[test]
+fn production_oracles_can_run_concurrently_with_isolated_state() {
+    let (first_ready_tx, first_ready_rx) = std::sync::mpsc::channel();
+    let (release_tx, release_rx) = std::sync::mpsc::channel();
+    let first = thread::spawn(move || {
+        let oracle = Oracle::start("");
+        first_ready_tx.send(()).expect("first ready");
+        release_rx.recv().expect("release first oracle");
+        drop(oracle);
+    });
+    first_ready_rx
+        .recv_timeout(Duration::from_secs(20))
+        .expect("first oracle ready");
+    let (second_ready_tx, second_ready_rx) = std::sync::mpsc::channel();
+    let second = thread::spawn(move || {
+        let oracle = Oracle::start("");
+        second_ready_tx.send(()).expect("second ready");
+        drop(oracle);
+    });
+    let concurrent = second_ready_rx.recv_timeout(Duration::from_secs(5)).is_ok();
+    release_tx.send(()).expect("release first");
+    first.join().expect("first oracle");
+    second.join().expect("second oracle");
+    assert!(concurrent, "second isolated oracle was globally serialized");
+}
+
 impl Oracle {
     pub(crate) fn start(mode: &str) -> Self {
-        static SERIAL: Mutex<()> = Mutex::new(());
         static NEXT: AtomicUsize = AtomicUsize::new(1);
-        let serial = SERIAL
+        let directory = loop {
-            .lock()
+            let candidate = std::env::temp_dir().join(format!(
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
+                "versevdi-rust-gateway-oracle-{}-{}",
-        let directory = std::env::temp_dir().join(format!(
+                std::process::id(),
-            "versevdi-rust-gateway-oracle-{}-{}",
+                NEXT.fetch_add(1, Ordering::Relaxed)
-            std::process::id(),
+            ));
-            NEXT.fetch_add(1, Ordering::Relaxed)
+            match fs::create_dir(&candidate) {
-        ));
+                Ok(()) => break candidate,
-        fs::create_dir(&directory).expect("create oracle directory");
+                Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {}
+                Err(error) => panic!("create oracle directory: {error}"),
+            }
+        };
         let source = directory.join("main.go");
         let ready_path = directory.join("ready.json");
         let stop_path = directory.join("stop");
@@ -242,7 +269,6 @@ impl Oracle {
             if let Ok(bytes) = fs::read(&ready_path) {
                 if let Ok(ready) = serde_json::from_slice(&bytes) {
                     return Self {
-                        _serial: serial,
                         child,
                         directory,
                         ready,
@@ -522,6 +548,8 @@ unsafe extern "C" {
 struct AbiSigners {
     admission: Arc<dyn SigningKey>,
     tls: Arc<dyn SigningKey>,
+    admission_threads: Mutex<Vec<thread::ThreadId>>,
+    tls_threads: Mutex<Vec<thread::ThreadId>>,
 }
 unsafe extern "C" fn abi_admission(raw: *mut c_void, input: AbiBytes, output: *mut u8) -> u32 {
     unsafe { abi_sign(raw, input, output, true) }
@@ -531,6 +559,15 @@ unsafe extern "C" fn abi_tls(raw: *mut c_void, input: AbiBytes, output: *mut u8)
 }
 unsafe fn abi_sign(raw: *mut c_void, input: AbiBytes, output: *mut u8, admission: bool) -> u32 {
     let context = unsafe { &*raw.cast::<AbiSigners>() };
+    let threads = if admission {
+        &context.admission_threads
+    } else {
+        &context.tls_threads
+    };
+    threads
+        .lock()
+        .expect("signer threads")
+        .push(thread::current().id());
     let bytes = unsafe { std::slice::from_raw_parts(input.data, input.length) };
     let key = if admission {
         &context.admission
@@ -552,6 +589,8 @@ fn real_abi_exports_connect_send_cancel_and_destroy_the_go_gateway_session() {
     let mut signers = Box::new(AbiSigners {
         admission: test_key(&oracle.ready.admission_key),
         tls: test_key(&oracle.ready.client_key),
+        admission_threads: Mutex::new(Vec::new()),
+        tls_threads: Mutex::new(Vec::new()),
     });
     let config = AbiConfig {
         struct_size: size_of::<AbiConfig>() as u32,
@@ -579,7 +618,20 @@ fn real_abi_exports_connect_send_cancel_and_destroy_the_go_gateway_session() {
             length: oracle.ready.credential.len(),
         },
     };
+    let calling_thread = thread::current().id();
     assert_eq!(unsafe { verse_core_connect_v1(core, &request) }, 0);
+    assert_eq!(
+        signers
+            .admission_threads
+            .lock()
+            .expect("admission threads")
+            .as_slice(),
+        [calling_thread]
+    );
+    assert_eq!(
+        signers.tls_threads.lock().expect("TLS threads").as_slice(),
+        [calling_thread]
+    );
     let mut values = [0; 12];
     values[0] = 1;
     values[2] = 30;