fix(runner): use select_all for true concurrent fail-fast

src/runner/mod.rs

@@ -1,6 +1,7 @@
 pub mod monitor;
 
 use bollard::Docker;
+use futures_util::future::select_all;
 use std::collections::{HashMap, HashSet};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
@@ -160,37 +161,49 @@ pub async fn run_step_parallel(
 type StepResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
 
 /// Await parallel step tasks; on first failure abort siblings so containers do not leak.
+/// Uses `select_all` to concurrently monitor all tasks and abort immediately when any fails,
+/// regardless of task order.
 pub async fn collect_parallel_results(
     handles: Vec<JoinHandle<StepResult>>,
 ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let mut error_result: Option<Box<dyn std::error::Error + Send + Sync>> = None;
-
-    for handle in handles {
-        if error_result.is_some() {
-            handle.abort();
-            if let Err(join_err) = handle.await && !join_err.is_cancelled() {
-                error_result.get_or_insert(Box::new(std::io::Error::other(format!(
-                    "Parallel step task join error after cancellation: {join_err}"
-                ))));
-            }
-            continue;
-        }
+    if handles.is_empty() {
+        return Ok(());
+    }
 
-        match handle.await {
+    let mut remaining = handles;
+
+    loop {
+        let (result, _index, rest) = select_all(remaining).await;
+        remaining = rest;
+
+        match result {
             Ok(Ok(())) => {}
-            Ok(Err(e)) => error_result = Some(e),
+            Ok(Err(e)) => {
+                // A step failed — abort all remaining sibling tasks immediately
+                for handle in remaining {
+                    handle.abort();
+                    // Drain the handle so tokio cleans up properly
+                    let _ = handle.await;
+                }
+                return Err(e);
+            }
             Err(join_err) if join_err.is_cancelled() => {}
             Err(join_err) => {
-                error_result = Some(Box::new(std::io::Error::other(format!(
-                    "Parallel step task panicked or was cancelled: {join_err}"
-                ))))
+                for handle in remaining {
+                    handle.abort();
+                    let _ = handle.await;
+                }
+                return Err(Box::new(std::io::Error::other(format!(
+                    "Parallel step task panicked: {join_err}"
+                ))));
             }
         }
-    }
 
-    if let Some(err) = error_result {
-        return Err(err);
+        if remaining.is_empty() {
+            break;
+        }
     }
+
     Ok(())
 }
 

tests/parallel_cleanup_tests.rs

@@ -1,4 +1,4 @@
-use forge::runner::collect_parallel_results;
+use forge_runner::runner::collect_parallel_results;
 use std::sync::{
     Arc,
     atomic::{AtomicUsize, Ordering},
@@ -15,7 +15,9 @@ async fn collect_parallel_results_succeeds_when_all_tasks_ok() {
         tokio::spawn(async { Ok(()) }),
     ];
 
-    collect_parallel_results(handles).await.expect("all tasks should succeed");
+    collect_parallel_results(handles)
+        .await
+        .expect("all tasks should succeed");
 }
 
 #[tokio::test]
@@ -27,10 +29,8 @@ async fn collect_parallel_results_aborts_remaining_tasks_on_failure() {
     let handles: Vec<JoinHandle<StepResult>> = vec![
         tokio::spawn(async move {
             started_fail.fetch_add(1, Ordering::SeqCst);
-            Err(
-                Box::new(std::io::Error::other("step failed"))
-                    as Box<dyn std::error::Error + Send + Sync>,
-            )
+            Err(Box::new(std::io::Error::other("step failed"))
+                as Box<dyn std::error::Error + Send + Sync>)
         }),
         tokio::spawn({
             let started = Arc::clone(&started);
@@ -54,4 +54,51 @@ async fn collect_parallel_results_aborts_remaining_tasks_on_failure() {
         0,
         "slow sibling must be aborted before completion"
     );
-}
+}
+
+/// Validates true concurrent fail-fast: slow task is first in the list,
+/// fast failing task is last. With `select_all`, the fast failure should
+/// abort the slow task without waiting for it to complete.
+#[tokio::test]
+async fn collect_parallel_results_aborts_slow_first_task_when_later_task_fails() {
+    let finished = Arc::new(AtomicUsize::new(0));
+
+    let finished_clone = Arc::clone(&finished);
+    let handles: Vec<JoinHandle<StepResult>> = vec![
+        // Slow task first — would block sequential await for 500ms
+        tokio::spawn({
+            let finished = Arc::clone(&finished);
+            async move {
+                tokio::time::sleep(Duration::from_millis(500)).await;
+                finished.fetch_add(1, Ordering::SeqCst);
+                Ok(())
+            }
+        }),
+        // Fast failing task last
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            finished_clone.fetch_add(1, Ordering::SeqCst);
+            Err(Box::new(std::io::Error::other("fast task failed"))
+                as Box<dyn std::error::Error + Send + Sync>)
+        }),
+    ];
+
+    let err = collect_parallel_results(handles)
+        .await
+        .expect_err("fast failing task should abort the stage");
+    assert!(err.to_string().contains("fast task failed"));
+    // Only the fast failing task incremented finished (value=1);
+    // the slow task must have been aborted before its sleep completed.
+    assert_eq!(
+        finished.load(Ordering::SeqCst),
+        1,
+        "slow first task must be aborted before it completes"
+    );
+}
+
+#[tokio::test]
+async fn collect_parallel_results_handles_empty_input() {
+    collect_parallel_results(vec![])
+        .await
+        .expect("empty handles should succeed immediately");
+}