From 5cc2d120e7c689e7236b3d7a1ca34a366171efc5 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Wed, 12 Aug 2026 17:20:20 +0700 Subject: [PATCH] fix(core): serialize ABI cancellation state --- core/include/versevdi_core.h | 14 +-- core/src/abi.rs | 178 +++++++++++++++++++++++++++++++---- 2 files changed, 170 insertions(+), 22 deletions(-) diff --git a/core/include/versevdi_core.h b/core/include/versevdi_core.h index 9affa83..c97a88b 100644 --- a/core/include/versevdi_core.h +++ b/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; - * signature_out is exactly 64 writable bytes. Admission may return OK, - * AUTHORITY_REJECTED, CANCELLED, or INTERNAL. TLS may return OK, TLS, CANCELLED, - * or INTERNAL. Any other value is normalized to INTERNAL. + * Signers run synchronously inline on the thread invoking the API call. + * transcript/tls_message is borrowed only for the callback; signature_out is + * exactly 64 writable bytes. Admission may return OK, AUTHORITY_REJECTED, + * 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 - * is borrowed only for its callback and must not be retained. + * Event callbacks run asynchronously on one core-owned worker thread and are + * 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, diff --git a/core/src/abi.rs b/core/src/abi.rs index a9f71dc..4160f00 100644 --- a/core/src/abi.rs +++ b/core/src/abi.rs @@ -50,6 +50,9 @@ static HANDLES: OnceLock>>> = OnceLock::new( #[cfg(test)] static CANCEL_ADMISSION_HOOK: OnceLock>> = OnceLock::new(); +#[cfg(test)] +static FINAL_CONNECT_HOOK: OnceLock>> = 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 = const { Cell::new(CallbackMode::None) }; } @@ -219,6 +229,7 @@ struct CoreInner { worker: Mutex>>, 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,21 +442,23 @@ 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), - Err(TryLockError::Poisoned(error)) => apply_cancellation(&mut error.into_inner()), + Ok(mut state) => apply_cancellation(inner, &mut state), + Err(TryLockError::Poisoned(error)) => { + apply_cancellation(inner, &mut error.into_inner()); + } Err(TryLockError::WouldBlock) => {} } - if first { - enqueue_callback_under_state_lock(inner, CallbackEvent::State(STATE_CANCELLED)); - } } #[derive(Clone, Copy)] @@ -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, - BUSY, CANCELLED, CANCEL_ADMISSION_HOOK, INTERNAL, OK, + ffi_boundary, handles, lock, BytesView, CancelAdmissionHook, Config, ConnectRequest, + 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>, + 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::() }; + // 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, + ) -> *mut CoreHandle { let config = Config { struct_size: u32::try_from(size_of::()).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 { 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::(); + 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::()).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); + } }