From 88b46b7efd7af16a2cb1c7dd6dc299432b54ca21 Mon Sep 17 00:00:00 2001 From: Allen Elzayn Date: Wed, 23 Sep 2026 16:23:09 +0000 Subject: [PATCH] fix(runner): use select_all for true concurrent fail-fast in parallel steps Replace sequential await loop in collect_parallel_results with futures_util::future::select_all so that a failing task aborts all siblings immediately, regardless of task order in the handle list. The previous implementation awaited handles one-by-one, meaning a slow task at index 0 would block detection of a fast failure at index N. Also adds: - Test for slow-first / fast-fail-last concurrent scenario - Test for empty handle input edge case --- src/runner/mod.rs | 53 +++++++++++++++++----------- tests/parallel_cleanup_tests.rs | 61 +++++++++++++++++++++++++++++---- 2 files changed, 87 insertions(+), 27 deletions(-) diff --git a/src/runner/mod.rs b/src/runner/mod.rs index 8edba16..b31bfa0 100644 --- a/src/runner/mod.rs +++ b/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>; /// 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>, ) -> Result<(), Box> { - let mut error_result: Option> = 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(()) } diff --git a/tests/parallel_cleanup_tests.rs b/tests/parallel_cleanup_tests.rs index e46bb95..a7cbc3a 100644 --- a/tests/parallel_cleanup_tests.rs +++ b/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> = vec![ tokio::spawn(async move { started_fail.fetch_add(1, Ordering::SeqCst); - Err( - Box::new(std::io::Error::other("step failed")) - as Box, - ) + Err(Box::new(std::io::Error::other("step failed")) + as Box) }), 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" ); -} \ No newline at end of file +} + +/// 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> = 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) + }), + ]; + + 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"); +}