fix(core): serialize ABI cancellation state

core/include/versevdi_core.h

@@ -89,10 +89,11 @@ typedef struct verse_control_event_v1 {
 } verse_control_event_v1_t;
 
 /*
- * Signers run synchronously. transcript/tls_message is borrowed only for the call;
+ * Signers run synchronously inline on the thread invoking the API call.
- * signature_out is exactly 64 writable bytes. Admission may return OK,
+ * transcript/tls_message is borrowed only for the callback; signature_out is
- * AUTHORITY_REJECTED, CANCELLED, or INTERNAL. TLS may return OK, TLS, CANCELLED,
+ * exactly 64 writable bytes. Admission may return OK, AUTHORITY_REJECTED,
- * or INTERNAL. Any other value is normalized to INTERNAL.
+ * CANCELLED, or INTERNAL. TLS may return OK, TLS, CANCELLED, or INTERNAL. Any
+ * other value is normalized to INTERNAL.
  */
 typedef verse_status_t (*verse_sign_admission_v1_fn)(
     void *signer_context,
@@ -103,8 +104,9 @@ typedef verse_status_t (*verse_sign_tls_ed25519_v1_fn)(
     verse_bytes_view_t tls_message,
     uint8_t signature_out[64]);
 /*
- * Event callbacks are serialized with one another. Each event and nested byte view
+ * Event callbacks run asynchronously on one core-owned worker thread and are
- * is borrowed only for its callback and must not be retained.
+ * serialized with one another. Each event and nested byte view is borrowed only
+ * for its callback and must not be retained.
  */
 typedef void (*verse_state_event_v1_fn)(
     void *context,

core/src/abi.rs

@@ -50,6 +50,9 @@ static HANDLES: OnceLock<Mutex<HashMap<usize, Arc<CoreInner>>>> = OnceLock::new(
 #[cfg(test)]
 static CANCEL_ADMISSION_HOOK: OnceLock<Mutex<Option<CancelAdmissionHook>>> = OnceLock::new();
 
+#[cfg(test)]
+static FINAL_CONNECT_HOOK: OnceLock<Mutex<Option<FinalConnectHook>>> = OnceLock::new();
+
 #[cfg(test)]
 struct CancelAdmissionHook {
     core: usize,
@@ -57,6 +60,13 @@ struct CancelAdmissionHook {
     resume: Receiver<()>,
 }
 
+#[cfg(test)]
+struct FinalConnectHook {
+    core: usize,
+    reached: Sender<()>,
+    resume: Receiver<()>,
+}
+
 thread_local! {
     static CALLBACK_MODE: Cell<CallbackMode> = const { Cell::new(CallbackMode::None) };
 }
@@ -219,6 +229,7 @@ struct CoreInner {
     worker: Mutex<Option<JoinHandle<()>>>,
     in_flight: AtomicUsize,
     cancelled: AtomicBool,
+    cancellation_emitted: AtomicBool,
     freeing: AtomicBool,
     destroying: AtomicBool,
     dropped_callbacks: AtomicU64,
@@ -281,6 +292,23 @@ fn pause_after_cancel_admission(core: *mut CoreHandle) {
     }
 }
 
+#[cfg(test)]
+fn pause_before_connected(core: *mut CoreHandle) {
+    let hook = FINAL_CONNECT_HOOK.get_or_init(|| Mutex::new(None));
+    let selected = {
+        let mut hook = lock(hook);
+        if hook.as_ref().is_some_and(|hook| hook.core == core as usize) {
+            hook.take()
+        } else {
+            None
+        }
+    };
+    if let Some(hook) = selected {
+        hook.reached.send(()).expect("announce final connect");
+        hook.resume.recv().expect("resume final connect");
+    }
+}
+
 fn status(error: CoreError) -> u32 {
     match error {
         CoreError::QueueFull => QUEUE_FULL,
@@ -414,20 +442,22 @@ fn callback_worker(
     let _ = done.send(());
 }
 
-fn apply_cancellation(state: &mut SessionState) {
+fn apply_cancellation(inner: &CoreInner, state: &mut SessionState) {
     state.session.cancel();
     state.lifecycle = Lifecycle::Cancelled;
+    if !inner.cancellation_emitted.swap(true, Ordering::AcqRel) {
+        enqueue_callback_under_state_lock(inner, CallbackEvent::State(STATE_CANCELLED));
+    }
 }
 
 fn cancel_inner(inner: &CoreInner) {
-    let first = !inner.cancelled.swap(true, Ordering::AcqRel);
+    inner.cancelled.store(true, Ordering::Release);
     match inner.state.try_lock() {
-        Ok(mut state) => apply_cancellation(&mut state),
+        Ok(mut state) => apply_cancellation(inner, &mut state),
-        Err(TryLockError::Poisoned(error)) => apply_cancellation(&mut error.into_inner()),
+        Err(TryLockError::Poisoned(error)) => {
-        Err(TryLockError::WouldBlock) => {}
+            apply_cancellation(inner, &mut error.into_inner());
         }
-    if first {
+        Err(TryLockError::WouldBlock) => {}
-        enqueue_callback_under_state_lock(inner, CallbackEvent::State(STATE_CANCELLED));
     }
 }
 
@@ -593,6 +623,7 @@ unsafe extern "C" fn verse_core_create_v1(
             worker: Mutex::new(Some(worker)),
             in_flight: AtomicUsize::new(0),
             cancelled: AtomicBool::new(false),
+            cancellation_emitted: AtomicBool::new(false),
             freeing: AtomicBool::new(false),
             destroying: AtomicBool::new(false),
             dropped_callbacks: AtomicU64::new(0),
@@ -646,7 +677,7 @@ unsafe extern "C" fn verse_core_connect_v1(
         {
             let mut state = lock(&handle.state);
             if handle.cancelled.load(Ordering::Acquire) {
-                apply_cancellation(&mut state);
+                apply_cancellation(&handle, &mut state);
                 return CANCELLED;
             }
             if state.lifecycle != Lifecycle::Created {
@@ -682,11 +713,28 @@ unsafe extern "C" fn verse_core_connect_v1(
             let mut state = lock(&handle.state);
             if handle.cancelled.load(Ordering::Acquire) || state.lifecycle == Lifecycle::Destroying
             {
-                apply_cancellation(&mut state);
+                apply_cancellation(&handle, &mut state);
+                return CANCELLED;
+            }
+            #[cfg(test)]
+            pause_before_connected(core);
+            if handle.cancelled.load(Ordering::Acquire) {
+                apply_cancellation(&handle, &mut state);
                 return CANCELLED;
             }
             state.lifecycle = Lifecycle::Connected;
             enqueue_callback_under_state_lock(&handle, CallbackEvent::State(STATE_CONNECTED));
+            if handle.cancelled.load(Ordering::Acquire) {
+                apply_cancellation(&handle, &mut state);
+                return CANCELLED;
+            }
+        }
+        if handle.cancelled.load(Ordering::Acquire)
+            && !handle.cancellation_emitted.load(Ordering::Acquire)
+        {
+            let mut state = lock(&handle.state);
+            apply_cancellation(&handle, &mut state);
+            return CANCELLED;
         }
         OK
     })
@@ -713,7 +761,7 @@ unsafe extern "C" fn verse_core_send_input_v1(
         };
         let mut state = lock(&handle.state);
         if handle.cancelled.load(Ordering::Acquire) {
-            apply_cancellation(&mut state);
+            apply_cancellation(&handle, &mut state);
             return CANCELLED;
         }
         if state.lifecycle != Lifecycle::Connected {
@@ -748,7 +796,7 @@ unsafe extern "C" fn verse_core_request_idr_v1(core: *mut CoreHandle) -> u32 {
         };
         let mut state = lock(&handle.state);
         if handle.cancelled.load(Ordering::Acquire) {
-            apply_cancellation(&mut state);
+            apply_cancellation(&handle, &mut state);
             return CANCELLED;
         }
         if state.lifecycle != Lifecycle::Connected {
@@ -936,29 +984,63 @@ mod tests {
     use std::mem::size_of;
     use std::ptr;
     use std::sync::mpsc;
-    use std::sync::Arc;
+    use std::sync::{Arc, Condvar, Mutex};
     use std::thread;
     use std::time::Duration;
 
     use super::{
-        ffi_boundary, handles, lock, BytesView, CancelAdmissionHook, Config, CoreHandle, SignFn,
+        ffi_boundary, handles, lock, BytesView, CancelAdmissionHook, Config, ConnectRequest,
-        BUSY, CANCELLED, CANCEL_ADMISSION_HOOK, INTERNAL, OK,
+        CoreHandle, FinalConnectHook, SignFn, StateEvent, StateFn, BUSY, CANCELLED,
+        CANCEL_ADMISSION_HOOK, FINAL_CONNECT_HOOK, INTERNAL, OK, STATE_CANCELLED, STATE_CONNECTED,
+        STATE_CONNECTING,
     };
 
+    const MANIFEST: &[u8] = br#"{
+      "version":"1","purpose":"launch","session_id":"session","reconnect_sequence":0,
+      "gateway":{"id":"gateway","addresses":["gateway.test:443"],"public_identity":"gateway.test"},
+      "tunnel":{"versions":["verse-gateway-v1/1"],"features":["control.v1","input.absolute.v1","input.scroll.v1"]},
+      "profile":{"id":"standard","bounds":{"minimum_kbps":1000,"target_kbps":5000,"maximum_kbps":10000},"display_mode":{"resolution_width":1920,"resolution_height":1080,"fps":60}},
+      "grant":{"opaque_value":"ggggggggggggggggggggggggggggggggggggggggggg","expires_at":"2099-01-01T00:00:00Z","audience":"audience"},
+      "correlation_id":"correlation"
+    }"#;
+    const CREDENTIAL: &[u8] = br#"{"client_device_id":"device","device_key_id":"key","certificate_chain_pem":"-----BEGIN CERTIFICATE-----\nAQID\n-----END CERTIFICATE-----","trust_bundle_pem":"-----BEGIN CERTIFICATE-----\nAQID\n-----END CERTIFICATE-----","expires_at":"2099-01-01T00:00:00Z"}"#;
+
+    #[derive(Default)]
+    struct CallbackContext {
+        states: Mutex<Vec<u32>>,
+        wake: Condvar,
+    }
+
     unsafe extern "C" fn sign(_context: *mut c_void, _input: BytesView, signature: *mut u8) -> u32 {
         // Test invariant: create supplies this callback only to the ABI, which provides 64 bytes.
         unsafe { ptr::write_bytes(signature, 0, 64) };
         OK
     }
 
+    unsafe extern "C" fn record_state(context: *mut c_void, event: *const StateEvent) {
+        // Test invariant: the context is the live CallbackContext supplied to create.
+        let context = unsafe { &*context.cast::<CallbackContext>() };
+        // Test invariant: the event is readable for the synchronous callback duration.
+        let state = unsafe { (*event).state };
+        context.states.lock().expect("states lock").push(state);
+        context.wake.notify_all();
+    }
+
     fn create_core() -> *mut CoreHandle {
+        create_core_with_callback(ptr::null_mut(), None)
+    }
+
+    fn create_core_with_callback(
+        context: *mut c_void,
+        on_state: Option<StateFn>,
+    ) -> *mut CoreHandle {
         let config = Config {
             struct_size: u32::try_from(size_of::<Config>()).expect("config size"),
             abi_version: 1,
-            context: ptr::null_mut(),
+            context,
             sign_admission: Some(sign as SignFn),
             sign_tls_ed25519: Some(sign as SignFn),
-            on_state: None,
+            on_state,
             on_error: None,
             on_stats: None,
             on_media: None,
@@ -973,6 +1055,22 @@ mod tests {
         core
     }
 
+    fn wait_for_state(context: &CallbackContext, wanted: u32) {
+        let deadline = std::time::Instant::now() + Duration::from_secs(1);
+        let mut states = context.states.lock().expect("states lock");
+        while !states.contains(&wanted) {
+            let remaining = deadline.saturating_duration_since(std::time::Instant::now());
+            assert!(
+                !remaining.is_zero(),
+                "state {wanted} not observed: {states:?}"
+            );
+            (states, _) = context
+                .wake
+                .wait_timeout(states, remaining)
+                .expect("state wait");
+        }
+    }
+
     fn inner(core: *mut CoreHandle) -> Arc<super::CoreInner> {
         lock(handles())
             .get(&(core as usize))
@@ -1092,4 +1190,52 @@ mod tests {
         assert_eq!(cancel.join().expect("cancel caller"), OK);
         assert_eq!(unsafe { super::verse_core_destroy_v1(core, 2_000) }, OK);
     }
+
+    #[test]
+    fn cancellation_during_final_connect_never_emits_connected_after_cancelled() {
+        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 (reached_tx, reached_rx) = mpsc::channel();
+        let (resume_tx, resume_rx) = mpsc::channel();
+        *lock(FINAL_CONNECT_HOOK.get_or_init(|| Mutex::new(None))) = Some(FinalConnectHook {
+            core: core as usize,
+            reached: reached_tx,
+            resume: resume_rx,
+        });
+
+        let address = core as usize;
+        let connect = thread::spawn(move || {
+            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: CREDENTIAL.as_ptr(),
+                    length: CREDENTIAL.len(),
+                },
+            };
+            // Test invariant: the request and static byte views remain valid for the call.
+            unsafe { super::verse_core_connect_v1(address as *mut CoreHandle, &raw const request) }
+        });
+        reached_rx.recv().expect("final connect reached");
+
+        assert_eq!(unsafe { super::verse_core_cancel_v1(core) }, OK);
+        resume_tx.send(()).expect("resume final connect");
+        let connect_status = connect.join().expect("connect caller");
+        wait_for_state(&context, STATE_CANCELLED);
+        if connect_status == OK {
+            wait_for_state(&context, STATE_CONNECTED);
+        }
+        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]
+        );
+        assert_eq!(connect_status, CANCELLED);
+    }
 }