diff --git a/core/src/abi.rs b/core/src/abi.rs index 274e1bc..ba97684 100644 --- a/core/src/abi.rs +++ b/core/src/abi.rs @@ -235,6 +235,13 @@ struct SessionState { features: Vec, } +#[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); - 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)), - ); + let published = { + let mut publication = lock(&worker_publication); + match *publication { + SessionPublication::Starting => { + *publication = SessionPublication::Failed(error); + false + } + 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::(); + 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::()).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] + ); + } } diff --git a/core/src/transport.rs b/core/src/transport.rs index 2b4fec3..f36561d 100644 --- a/core/src/transport.rs +++ b/core/src/transport.rs @@ -331,6 +331,17 @@ impl TransportSession { commands: tokio::sync::mpsc::Receiver, 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, + events: SessionEventSender, + cancellation: Cancellation, + started: Option>, ) -> 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"); diff --git a/core/tests/abi_contract.rs b/core/tests/abi_contract.rs index 6ce6939..860ca2c 100644 --- a/core/tests/abi_contract.rs +++ b/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); diff --git a/core/tests/gateway_oracle.rs b/core/tests/gateway_oracle.rs index d21fc50..5ff3020 100644 --- a/core/tests/gateway_oracle.rs +++ b/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 - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let directory = std::env::temp_dir().join(format!( - "versevdi-rust-gateway-oracle-{}-{}", - std::process::id(), - NEXT.fetch_add(1, Ordering::Relaxed) - )); - fs::create_dir(&directory).expect("create oracle directory"); + let directory = loop { + let candidate = std::env::temp_dir().join(format!( + "versevdi-rust-gateway-oracle-{}-{}", + std::process::id(), + NEXT.fetch_add(1, Ordering::Relaxed) + )); + match fs::create_dir(&candidate) { + Ok(()) => break candidate, + 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, tls: Arc, + admission_threads: Mutex>, + tls_threads: Mutex>, } 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::() }; + 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::() 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;