From 7a03644a9d510ebbb73e258aec87eead281a4a01 Mon Sep 17 00:00:00 2001 From: sechmachine <97589681+sechmachine727@users.noreply.github.com> Date: Wed, 12 Aug 2026 23:14:23 +0700 Subject: [PATCH] fix(core): arbitrate session startup --- core/src/abi.rs | 43 +++++---- core/src/transport.rs | 125 +++++++++++++++++++++++++-- core/tests/gateway_oracle.rs | 163 ++++++++++++++++++++++++++++++----- 3 files changed, 289 insertions(+), 42 deletions(-) diff --git a/core/src/abi.rs b/core/src/abi.rs index ba97684..abe02f4 100644 --- a/core/src/abi.rs +++ b/core/src/abi.rs @@ -963,7 +963,8 @@ 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 (session_startup, session_prepared_rx, session_commit_tx, session_outcome_rx) = + transport::session_startup(); let (event_tx, event_rx) = bounded_session_events(); let publication = Arc::new(Mutex::new(SessionPublication::Starting)); let worker_publication = Arc::clone(&publication); @@ -1019,22 +1020,18 @@ unsafe extern "C" fn verse_core_connect_v1( command_rx, event_tx, cancellation, - Some(session_started_tx), + Some(session_startup), )) } 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 (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1); + #[cfg(test)] + if test_session_failure { + let _ = result_tx.try_send(Err(CoreError::Transport)); + } + let _result_tx = result_tx; + transport::await_session_startup(&mut result_rx, session_startup).await?; let mut command_rx = command_rx; while !cancellation.is_cancelled() { if matches!( @@ -1045,8 +1042,8 @@ unsafe extern "C" fn verse_core_connect_v1( break; } } - }); - Err(CoreError::Cancelled) + Err(CoreError::Cancelled) + }) }; let _ = bridge.join(); if let Err(error) = session_result { @@ -1084,13 +1081,27 @@ 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 + if session_prepared_rx .recv_timeout(Duration::from_secs(10)) .is_err() { cancel_inner(&handle); return INTERNAL; } + let _ = session_commit_tx.send(()); + let startup_result = match session_outcome_rx.recv_timeout(Duration::from_secs(10)) { + Ok(result) => result, + Err(_) => { + cancel_inner(&handle); + return INTERNAL; + } + }; + if let Err(error) = startup_result { + let mut publication = lock(&publication); + if *publication == SessionPublication::Starting { + *publication = SessionPublication::Failed(error); + } + } { let mut publication = lock(&publication); let mut state = lock(&handle.state); diff --git a/core/src/transport.rs b/core/src/transport.rs index f36561d..a730975 100644 --- a/core/src/transport.rs +++ b/core/src/transport.rs @@ -167,6 +167,57 @@ pub struct TransportSession { features: Vec, } +pub(crate) struct SessionStartup { + prepared: std::sync::mpsc::Sender<()>, + commit: tokio::sync::oneshot::Receiver<()>, + outcome: std::sync::mpsc::Sender>, +} + +pub(crate) fn session_startup() -> ( + SessionStartup, + std::sync::mpsc::Receiver<()>, + tokio::sync::oneshot::Sender<()>, + std::sync::mpsc::Receiver>, +) { + let (prepared_tx, prepared_rx) = std::sync::mpsc::channel(); + let (commit_tx, commit_rx) = tokio::sync::oneshot::channel(); + let (outcome_tx, outcome_rx) = std::sync::mpsc::channel(); + ( + SessionStartup { + prepared: prepared_tx, + commit: commit_rx, + outcome: outcome_tx, + }, + prepared_rx, + commit_tx, + outcome_rx, + ) +} + +pub(crate) async fn await_session_startup( + result_rx: &mut tokio::sync::mpsc::Receiver>, + startup: SessionStartup, +) -> Result<()> { + startup + .prepared + .send(()) + .map_err(|_| CoreError::Cancelled)?; + let mut loop_result = std::pin::pin!(result_rx.recv()); + let mut commit = std::pin::pin!(startup.commit); + let result = std::future::poll_fn(|context| { + if let std::task::Poll::Ready(result) = + std::future::Future::poll(loop_result.as_mut(), context) + { + return std::task::Poll::Ready(result.unwrap_or(Err(CoreError::Transport))); + } + std::future::Future::poll(commit.as_mut(), context) + .map(|result| result.map_err(|_| CoreError::Cancelled)) + }) + .await; + let _ = startup.outcome.send(result); + result +} + #[derive(Debug, Eq, PartialEq)] pub enum SessionCommand { Input(Vec), @@ -341,7 +392,7 @@ impl TransportSession { commands: tokio::sync::mpsc::Receiver, events: SessionEventSender, cancellation: Cancellation, - started: Option>, + startup: Option, ) -> Result<()> { let Self { endpoint, @@ -369,8 +420,16 @@ 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(()); + if let Some(startup) = startup { + if let Err(error) = await_session_startup(&mut result_rx, startup).await { + connection.close(APPLICATION_ERROR, b"session stopped"); + writer.abort(); + reliable.abort(); + datagrams.abort(); + endpoint.wait_idle().await; + events.close(); + return Err(error); + } } loop { if cancellation.check().is_err() { @@ -968,8 +1027,9 @@ mod tests { use std::time::{Duration, Instant}; use super::{ - bounded_lookup, bounded_session_events, decode_reliable_frame, encode_reliable_frame, - hello_length, Cancellation, SessionEvent, HELLO_LIMIT, + await_session_startup, bounded_lookup, bounded_session_events, decode_reliable_frame, + encode_reliable_frame, hello_length, session_startup, Cancellation, SessionEvent, + HELLO_LIMIT, }; use crate::error::CoreError; use crate::media::{EncodedUnit, MediaChannel}; @@ -1096,4 +1156,59 @@ mod tests { Err(tokio::sync::mpsc::error::TrySendError::Full(_)) )); } + + #[test] + fn queued_loop_failure_wins_before_session_ready() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1); + result_tx + .try_send(Err(crate::error::CoreError::Protocol)) + .expect("queue terminal loop result"); + let (startup, prepared, commit, outcome) = session_startup(); + let running = + runtime.spawn(async move { await_session_startup(&mut result_rx, startup).await }); + runtime.block_on(async { tokio::task::yield_now().await }); + prepared.recv().expect("session prepared"); + let _ = commit.send(()); + + assert_eq!( + outcome.recv().expect("startup outcome"), + Err(crate::error::CoreError::Protocol) + ); + assert_eq!( + runtime.block_on(running).expect("startup task"), + Err(crate::error::CoreError::Protocol) + ); + } + + #[test] + fn loop_failure_after_session_ready_remains_for_publication_race() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("runtime"); + let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1); + let (startup, prepared, commit, outcome) = session_startup(); + let running = runtime.spawn(async move { + let result = await_session_startup(&mut result_rx, startup).await; + (result, result_rx) + }); + runtime.block_on(async { tokio::task::yield_now().await }); + prepared.recv().expect("session prepared"); + commit.send(()).expect("commit startup"); + + let (startup_result, mut result_rx) = runtime.block_on(running).expect("startup task"); + assert_eq!(startup_result, Ok(())); + assert_eq!(outcome.recv().expect("startup outcome"), Ok(())); + result_tx + .try_send(Err(CoreError::Protocol)) + .expect("queue post-ready terminal result"); + assert_eq!( + runtime.block_on(result_rx.recv()), + Some(Err(CoreError::Protocol)) + ); + } } diff --git a/core/tests/gateway_oracle.rs b/core/tests/gateway_oracle.rs index 5ff3020..e6ab696 100644 --- a/core/tests/gateway_oracle.rs +++ b/core/tests/gateway_oracle.rs @@ -209,6 +209,39 @@ pub(crate) struct Oracle { pub(crate) ready: Ready, } +struct OracleStartup { + child: Option, + directory: Option, +} + +impl OracleStartup { + fn new(directory: std::path::PathBuf) -> Self { + Self { + child: None, + directory: Some(directory), + } + } + + fn finish(mut self, ready: Ready) -> Oracle { + Oracle { + child: self.child.take().expect("started oracle child"), + directory: self.directory.take().expect("oracle directory"), + ready, + } + } +} + +impl Drop for OracleStartup { + fn drop(&mut self) { + if let Some(child) = self.child.as_mut() { + stop_child(child, self.directory.as_deref()); + } + if let Some(directory) = self.directory.as_ref() { + let _ = fs::remove_dir_all(directory); + } + } +} + #[test] fn production_oracles_can_run_concurrently_with_isolated_state() { let (first_ready_tx, first_ready_rx) = std::sync::mpsc::channel(); @@ -235,10 +268,61 @@ fn production_oracles_can_run_concurrently_with_isolated_state() { assert!(concurrent, "second isolated oracle was globally serialized"); } +#[test] +fn oracle_startup_timeout_kills_child_and_removes_directory() { + let directory = Oracle::unique_directory(); + let mut command = Command::new("sh"); + command.args(["-c", "while :; do sleep 1; done"]); + let started = Instant::now(); + + let result = Oracle::start_in("", directory.clone(), command, Duration::from_millis(20)); + + assert!(result.is_err(), "non-ready child unexpectedly became ready"); + assert!(started.elapsed() < Duration::from_secs(2)); + assert!(!directory.exists(), "timed-out oracle directory leaked"); +} + +#[test] +fn oracle_source_failure_removes_directory_before_spawn() { + let directory = Oracle::unique_directory(); + fs::create_dir(directory.join("main.go")).expect("block source file creation"); + + let result = Oracle::start_in( + "", + directory.clone(), + Command::new("go"), + Duration::from_millis(20), + ); + + assert!(result.is_err(), "invalid source path unexpectedly started"); + assert!(!directory.exists(), "failed oracle directory leaked"); +} + +#[test] +fn oracle_spawn_failure_removes_directory() { + let directory = Oracle::unique_directory(); + + let result = Oracle::start_in( + "", + directory.clone(), + Command::new("/definitely/missing/versevdi-go"), + Duration::from_millis(20), + ); + + assert!(result.is_err(), "missing executable unexpectedly started"); + assert!(!directory.exists(), "spawn-failed oracle directory leaked"); +} + impl Oracle { pub(crate) fn start(mode: &str) -> Self { + let directory = Self::unique_directory(); + Self::start_in(mode, directory, Command::new("go"), Duration::from_secs(20)) + .unwrap_or_else(|error| panic!("{error}")) + } + + fn unique_directory() -> std::path::PathBuf { static NEXT: AtomicUsize = AtomicUsize::new(1); - let directory = loop { + loop { let candidate = std::env::temp_dir().join(format!( "versevdi-rust-gateway-oracle-{}-{}", std::process::id(), @@ -249,13 +333,26 @@ impl Oracle { Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => {} Err(error) => panic!("create oracle directory: {error}"), } - }; + } + } + + fn start_in( + mode: &str, + directory: std::path::PathBuf, + mut command: Command, + timeout: Duration, + ) -> Result { + let mut startup = OracleStartup::new(directory.clone()); let source = directory.join("main.go"); let ready_path = directory.join("ready.json"); let stop_path = directory.join("stop"); - fs::write(&source, ORACLE_SOURCE).expect("write oracle source"); - let child = Command::new("go") - .args(["run", source.to_str().expect("source path is UTF-8")]) + fs::write(&source, ORACLE_SOURCE) + .map_err(|error| format!("write Go oracle source: {error}"))?; + let source = source + .to_str() + .ok_or_else(|| "Go oracle source path is not UTF-8".to_owned())?; + let child = command + .args(["run", source]) .arg(&ready_path) .arg(&stop_path) .env("VERSEVDI_RUST_ORACLE_MODE", mode) @@ -263,27 +360,29 @@ impl Oracle { .stdout(Stdio::piped()) .stderr(Stdio::piped()) .spawn() - .expect("start Go gateway oracle"); - let deadline = Instant::now() + Duration::from_secs(20); + .map_err(|error| format!("start Go gateway oracle: {error}"))?; + startup.child = Some(child); + let deadline = Instant::now() + timeout; while Instant::now() < deadline { if let Ok(bytes) = fs::read(&ready_path) { if let Ok(ready) = serde_json::from_slice(&bytes) { - return Self { - child, - directory, - ready, - }; + return Ok(startup.finish(ready)); } } + if let Some(status) = startup + .child + .as_mut() + .expect("started oracle child") + .try_wait() + .map_err(|error| format!("poll Go gateway oracle: {error}"))? + { + return Err(format!( + "Go gateway oracle exited before becoming ready: status={status}" + )); + } thread::sleep(Duration::from_millis(10)); } - let output = child.wait_with_output().expect("collect Go oracle output"); - panic!( - "Go gateway oracle did not become ready: status={} stdout={} stderr={}", - output.status, - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); + Err("Go gateway oracle did not become ready before the startup deadline".to_owned()) } fn provider_starts(&self) -> usize { @@ -317,12 +416,34 @@ impl Oracle { impl Drop for Oracle { fn drop(&mut self) { - let _ = fs::write(self.directory.join("stop"), []); - let _ = self.child.wait(); + stop_child(&mut self.child, Some(&self.directory)); let _ = fs::remove_dir_all(&self.directory); } } +fn stop_child(child: &mut Child, directory: Option<&std::path::Path>) { + if let Some(directory) = directory { + let _ = fs::write(directory.join("stop"), []); + } + if wait_for_child(child, Duration::from_millis(250)) { + return; + } + if child.kill().is_ok() { + let _ = wait_for_child(child, Duration::from_millis(500)); + } +} + +fn wait_for_child(child: &mut Child, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + loop { + match child.try_wait() { + Ok(Some(_)) => return true, + Ok(None) if Instant::now() < deadline => thread::sleep(Duration::from_millis(5)), + Ok(None) | Err(_) => return false, + } + } +} + pub(crate) fn test_key(encoded: &str) -> Arc { let der = STANDARD.decode(encoded).expect("decode test key"); rustls::crypto::ring::default_provider()