fix(manager): report a spawned run as finished, not failed_to_start

implementation/src/manager/run.rs

@@ -305,6 +305,7 @@ struct SingleGenVMContext {
     consumed_result: tokio::sync::OnceCell<Vec<u8>>,
 
     process_handle: tokio::sync::Mutex<Option<tokio::process::Child>>,
+    spawned: AtomicBool,
     cancel_requested: AtomicBool,
     cancel_notify: tokio::sync::Notify,
     finish_cause: std::sync::Mutex<Option<FinishCause>>,
@@ -331,6 +332,17 @@ impl SingleGenVMContext {
             .unwrap_or_else(|_| String::new())
     }
 
+    /// Records that `proc.spawn()` returned, so a later failure is classified
+    /// by the protocol's own rule (a process was created) rather than by which
+    /// arm of a `select!` happened to win the race.
+    fn mark_spawned(&self) {
+        self.spawned.store(true, Ordering::SeqCst);
+    }
+
+    fn was_spawned(&self) -> bool {
+        self.spawned.load(Ordering::SeqCst)
+    }
+
     fn request_finish(&self, cause: FinishCause) {
         let should_notify = !self.cancel_requested.swap(true, Ordering::SeqCst);
         if let Ok(mut stored_cause) = self.finish_cause.lock() {
@@ -2091,6 +2103,7 @@ impl Ctx {
                 id: genvm_id,
                 host_genvm_id: req.host_genvm_id.clone(),
                 process_handle: tokio::sync::Mutex::new(None),
+                spawned: AtomicBool::new(false),
                 started_at: chrono::Utc::now(),
                 strict_deadline: strict_deadline_from_request(&req),
 
@@ -2269,8 +2282,32 @@ async fn supervise_genvm(
     if let Err(e) =
         supervise_genvm_inner(full_ctx, exec_ctx.clone(), req, modules_lock, permits).await
     {
-        fail_to_start(&exec_ctx, e);
+        report_run_failure(&exec_ctx, e).await;
+    }
+}
+
+/// Publishes the terminal event for a run that ended in an error.
+///
+/// The protocol separates `failed_to_start` from `finished` by whether a
+/// process was created, not by which code path noticed the error
+/// (`docs/website/src/impl-spec/appendix/manager-socket.rst`). Once
+/// `proc.spawn()` has returned, the run is a `finished` however it failed, so
+/// the spawned flag decides and the error only decides which of the two to
+/// report. Before the flag existed, an `EPIPE` from the stdin writer and a
+/// completed `child.wait()` raced in the same `select!`, and the loser was
+/// reported as never having started even though a process existed.
+async fn report_run_failure(exec: &SingleGenVMContext, error: anyhow::Error) {
+    if !exec.was_spawned() {
+        fail_to_start(exec, error);
+        return;
     }
+
+    // No exit status exists: the failure happened before `child.wait()` could
+    // return one, which the protocol allows (`exit_code` is null when the run
+    // was killed before its exit code was known). The error string stays in the
+    // log, since `finished` has no field to carry it.
+    log_warn_into!(@operator, &LoggerWithId, genvm_id:id = exec.id.0, error:ah = error; "executor was spawned but the run failed before its exit status was known");
+    let _ = finish_execution(exec, None, FinishCause::Exited).await;
 }
 
 async fn supervise_genvm_inner(
@@ -2494,6 +2531,11 @@ async fn run_genvm_process(
     let process = async {
         // Spawn child process, then drop child-side FDs
         let mut child = proc.spawn()?;
+        // From here a process exists. The protocol separates `started` from
+        // `failed_to_start` on exactly this fact, so record it before anything
+        // else can fail: a later error is a `finished`, not a
+        // `failed_to_start`, whichever arm of the stdin `select!` reports it.
+        exec_ctx.mark_spawned();
         drop(write_fd);
         log_debug_into!(&LoggerWithId, genvm_id:id = genvm_id.0, pid:? = child.id(); "genvm process started");
         std::mem::drop(module_child_fds);

implementation/src/manager/run_test.rs

@@ -54,6 +54,7 @@ fn fake_execution_with_host_id(
         log_sink: Arc::new(LogSinkInner::new(false)),
         consumed_result: tokio::sync::OnceCell::new(),
         process_handle: tokio::sync::Mutex::new(None),
+        spawned: AtomicBool::new(false),
         cancel_requested: AtomicBool::new(false),
         cancel_notify: tokio::sync::Notify::new(),
         finish_cause: std::sync::Mutex::new(None),
@@ -751,6 +752,44 @@ async fn terminal_event_is_published_once() {
     assert!(matches!(rx.has_changed(), Ok(false)));
 }
 
+#[tokio::test]
+async fn a_failure_after_spawn_is_a_finished_not_a_failed_to_start() {
+    // The protocol distinguishes the two terminal events by whether a process
+    // was created, not by which arm of the stdin `select!` noticed the error.
+    // `report_run_failure` is what production calls for every error out of a
+    // run, so exercising it covers the classification itself.
+    let error = || anyhow::anyhow!("failed to write execution data to child stdin: EPIPE");
+
+    // Spawned: a process exists, so this is a `finished` with no exit code
+    // (the protocol allows a null `exit_code` when it was killed before the
+    // code was known). Reporting `failed_to_start` here told the host no
+    // process ever ran, which is false.
+    let spawned = fake_execution(GenVMId(1), None);
+    spawned.mark_spawned();
+    let mut rx = spawned.events.subscribe();
+    report_run_failure(&spawned, error()).await;
+    match rx.borrow_and_update().clone() {
+        Snapshot::Event(Event::Finished {
+            exit_code, cause, ..
+        }) => {
+            assert_eq!(exit_code, None);
+            assert_eq!(cause, FinishCause::Exited);
+        }
+        other => panic!("spawned run reported {other:?}, expected a finished event"),
+    }
+
+    // Control: the same error with no process is still a `failed_to_start`.
+    // Without this the test would also pass if the branch were removed and
+    // every failure became a `finished`.
+    let never_spawned = fake_execution(GenVMId(2), None);
+    let mut rx = never_spawned.events.subscribe();
+    report_run_failure(&never_spawned, error()).await;
+    assert!(matches!(
+        rx.borrow_and_update().clone(),
+        Snapshot::Event(Event::FailedToStart { .. })
+    ));
+}
+
 #[tokio::test]
 async fn process_completion_waits_for_final_logs_on_success_and_error() {
     use std::io::Write;