fix(core): arbitrate session startup

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();
+                    runtime.block_on(async move {
+                        let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(1);
                         #[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 _ = result_tx.try_send(Err(CoreError::Transport));
                         }
-                    let _ = session_started_tx.send(());
-                    runtime.block_on(async move {
+                        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)
+                    })
                 };
                 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);

core/src/transport.rs

@@ -167,6 +167,57 @@ pub struct TransportSession {
     features: Vec<String>,
 }
 
+pub(crate) struct SessionStartup {
+    prepared: std::sync::mpsc::Sender<()>,
+    commit: tokio::sync::oneshot::Receiver<()>,
+    outcome: std::sync::mpsc::Sender<Result<()>>,
+}
+
+pub(crate) fn session_startup() -> (
+    SessionStartup,
+    std::sync::mpsc::Receiver<()>,
+    tokio::sync::oneshot::Sender<()>,
+    std::sync::mpsc::Receiver<Result<()>>,
+) {
+    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<Result<()>>,
+    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<u8>),
@@ -341,7 +392,7 @@ impl TransportSession {
         commands: tokio::sync::mpsc::Receiver<SessionCommand>,
         events: SessionEventSender,
         cancellation: Cancellation,
-        started: Option<std::sync::mpsc::Sender<()>>,
+        startup: Option<SessionStartup>,
     ) -> 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))
+        );
+    }
 }

core/tests/gateway_oracle.rs

@@ -209,6 +209,39 @@ pub(crate) struct Oracle {
     pub(crate) ready: Ready,
 }
 
+struct OracleStartup {
+    child: Option<Child>,
+    directory: Option<std::path::PathBuf>,
+}
+
+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<Self, String> {
+        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<dyn SigningKey> {
     let der = STANDARD.decode(encoded).expect("decode test key");
     rustls::crypto::ring::default_provider()