fix(runner): abort parallel siblings on step failure

[Nueramarcos]

Parallel fail-fast left sibling tokio tasks running, orphaning containers.
This change aborts siblings and always runs cleanup after parallel steps.

---

Airport upstream lane · Nueramarcos

[gemini-code-assist]

Code Review

This pull request refactors the parallel step execution logic to collect task handles and abort sibling tasks upon encountering a failure, preventing container leaks. It also replaces a discarded result in the dependency resolution check with proper error propagation and adds unit tests for the parallel cleanup behavior. The review feedback correctly identifies that the new collect_parallel_results function awaits task handles sequentially in a loop, which blocks on earlier tasks and defeats the intended fail-fast behavior if a later task fails first. To address this, the reviewer suggests using futures_util::future::select_all to monitor tasks concurrently and recommends updating the test cases to place the slow task first to properly validate this concurrent fail-fast behavior.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Awaiting the JoinHandles sequentially in a for loop blocks on each task in the order they were spawned. If a later task (e.g., the second one) fails early, the loop will remain blocked awaiting the first task to complete before it can detect the failure and abort the others. This defeats the intended fail-fast behavior.

To achieve true concurrent monitoring and immediate fail-fast, you can use futures_util::future::select_all to await the first completed task.

pub async fn collect_parallel_results(
    mut 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;

    while !handles.is_empty() {
        let (res, _index, remaining) = futures_util::future::select_all(handles).await;
        handles = remaining;

        match res {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                error_result = Some(e);
                break;
            }
            Err(join_err) => {
                if !join_err.is_cancelled() {
                    error_result = Some(Box::new(std::io::Error::other(format!(
                        "Parallel step task panicked: {join_err}"
                    ))));
                }
                break;
            }
        }
    }

    for handle in &handles {
        handle.abort();
    }
    for handle in handles {
        let _ = handle.await;
    }

    if let Some(err) = error_result {
        return Err(err);
    }
    Ok(())
}

[gemini-code-assist]

The reason this test passed even with the sequential implementation is that the failing task was placed first in the handles vector (handles[0]). If the slow task were placed first, the sequential implementation would block on it for 500ms before checking the failing task, failing the test.

To make the test robust and guarantee true concurrent fail-fast behavior, we should place the slow task first in the vector.

Suggested change

	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>,
	)
	}),
	tokio::spawn({
	let started = Arc::clone(&started);
	let finished = Arc::clone(&finished);
	async move {
	started.fetch_add(1, Ordering::SeqCst);
	tokio::time::sleep(Duration::from_millis(500)).await;
	finished.fetch_add(1, Ordering::SeqCst);
	Ok(())
	}
	}),
	];
	let handles: Vec<JoinHandle<StepResult>> = vec![
	tokio::spawn({
	let started = Arc::clone(&started);
	let finished = Arc::clone(&finished);
	async move {
	started.fetch_add(1, Ordering::SeqCst);
	tokio::time::sleep(Duration::from_millis(500)).await;
	finished.fetch_add(1, Ordering::SeqCst);
	Ok(())
	}
	}),
	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>,
	)
	}),
	];