diff --git a/src/main.rs b/src/main.rs index 6244e8e..9edda13 100644 --- a/src/main.rs +++ b/src/main.rs @@ -749,11 +749,9 @@ async fn forge_main() -> Result<(), Box> { // Validate parallel stages validate_parallel_stages(&config)?; - // Check for circular dependencies in stages - // TODO: Implement circular dependency check - + // Circular / missing dependencies (Kahn's algorithm in resolve_stage_dependencies) if !config.stages.is_empty() { - let _ = resolve_stage_dependencies(&config.stages)?; + resolve_stage_dependencies(&config.stages)?; } println!("{}", "Configuration is valid!".green().bold()); diff --git a/src/runner/mod.rs b/src/runner/mod.rs index 5d58b24..8edba16 100644 --- a/src/runner/mod.rs +++ b/src/runner/mod.rs @@ -1,11 +1,11 @@ pub mod monitor; use bollard::Docker; -use futures_util::stream::{FuturesUnordered, StreamExt}; use std::collections::{HashMap, HashSet}; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::sync::{Mutex, Semaphore}; +use tokio::task::JoinHandle; use crate::config::{CacheConfig, Stage, Step}; use crate::docker::{ @@ -112,7 +112,9 @@ pub async fn run_step_parallel( .clone() }; - let _permit = image_lock.acquire().await.unwrap(); + let _permit = image_lock.acquire().await.map_err(|_| { + std::io::Error::other("Image pull lock closed unexpectedly during parallel execution") + })?; let container_ctx = ContainerRuntimeContext { workspace_dir: &runtime.workspace_dir, cache_dir: &runtime.cache_dir, @@ -155,6 +157,43 @@ pub async fn run_step_parallel( wait_result.map(|_| ()) } +type StepResult = Result<(), Box>; + +/// Await parallel step tasks; on first failure abort siblings so containers do not leak. +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; + } + + match handle.await { + Ok(Ok(())) => {} + Ok(Err(e)) => error_result = Some(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}" + )))) + } + } + } + + if let Some(err) = error_result { + return Err(err); + } + Ok(()) +} + pub async fn run_stage_parallel( docker: &Docker, steps: &[Step], @@ -169,7 +208,7 @@ pub async fn run_stage_parallel( image_pull_locks: Arc::new(Mutex::new(HashMap::new())), }); - let mut tasks = FuturesUnordered::new(); + let mut handles = Vec::with_capacity(steps.len()); for (index, step) in steps.iter().enumerate() { let docker = docker.clone(); @@ -183,7 +222,7 @@ pub async fn run_stage_parallel( let runtime = runtime.clone(); let monitor = Arc::clone(&monitor); - tasks.push(tokio::spawn(async move { + handles.push(tokio::spawn(async move { run_step_parallel( &docker, &step, verbose, &cache, &temp_dir, task, &runtime, monitor, ) @@ -191,35 +230,16 @@ pub async fn run_stage_parallel( })); } - // Collect results - fail fast on first error - let mut error_result = None; - while let Some(result) = tasks.next().await { - match result { - Ok(Ok(_)) => continue, - Ok(Err(e)) => { - error_result = Some(e); - break; - } - Err(e) => { - error_result = Some(e.into()); - break; - } - } - } + let run_result = collect_parallel_results(handles).await; - // Cleanup containers + // Always cleanup tracked containers (including siblings aborted after a failure). let ids = { ctx.container_ids.lock().await.clone() }; cleanup_containers(docker, &ctx.container_ids, verbose).await; for id in ids { monitor.on_container_destroyed(&id); } - // Return error if any task failed - if let Some(err) = error_result { - return Err(err); - } - - Ok(()) + run_result } pub fn resolve_stage_dependencies( diff --git a/tests/parallel_cleanup_tests.rs b/tests/parallel_cleanup_tests.rs new file mode 100644 index 0000000..e46bb95 --- /dev/null +++ b/tests/parallel_cleanup_tests.rs @@ -0,0 +1,57 @@ +use forge::runner::collect_parallel_results; +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; +use std::time::Duration; +use tokio::task::JoinHandle; + +type StepResult = Result<(), Box>; + +#[tokio::test] +async fn collect_parallel_results_succeeds_when_all_tasks_ok() { + let handles: Vec> = vec![ + tokio::spawn(async { Ok(()) }), + tokio::spawn(async { Ok(()) }), + ]; + + collect_parallel_results(handles).await.expect("all tasks should succeed"); +} + +#[tokio::test] +async fn collect_parallel_results_aborts_remaining_tasks_on_failure() { + let started = Arc::new(AtomicUsize::new(0)); + let finished = Arc::new(AtomicUsize::new(0)); + + let started_fail = Arc::clone(&started); + 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, + ) + }), + 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 err = collect_parallel_results(handles) + .await + .expect_err("first failing task should fail the stage"); + assert!(err.to_string().contains("step failed")); + assert_eq!(started.load(Ordering::SeqCst), 2); + assert_eq!( + finished.load(Ordering::SeqCst), + 0, + "slow sibling must be aborted before completion" + ); +} \ No newline at end of file