diff --git a/.ci.yml b/.ci.yml index 996a6d9..3d30e19 100644 --- a/.ci.yml +++ b/.ci.yml @@ -5,12 +5,12 @@ pipelines: branches: ["*"] jobs: check: - run: . ~/.cargo/env && cargo check --workspace + run: echo "HOME=$HOME PATH=$PATH" && . "$HOME/.cargo/env" && which cargo && cargo check --workspace timeout: 600 clippy: - run: . ~/.cargo/env && cargo clippy --workspace --all-targets -- -D warnings + run: . "$HOME/.cargo/env" && cargo clippy --workspace --all-targets -- -D warnings timeout: 600 test: needs: [check] - run: . ~/.cargo/env && cargo xtask test essential + run: . "$HOME/.cargo/env" && cargo xtask test essential timeout: 900 diff --git a/crates/ci/src/local_runner.rs b/crates/ci/src/local_runner.rs index 9412def..0cd0b7f 100644 --- a/crates/ci/src/local_runner.rs +++ b/crates/ci/src/local_runner.rs @@ -4,7 +4,7 @@ //! is ready to execute. use std::io::BufRead; -use std::process::{Command, Stdio}; +use std::process::{Child, Command, Stdio}; use std::sync::{Arc, Mutex}; use swactor::actor::{ActorAddress, ActorInterface, Ctx}; @@ -41,7 +41,18 @@ impl LocalRunner { let work_dir = &self.start_job.work_dir; let timeout_secs = self.start_job.job_def.timeout_secs; + eprintln!( + "[runner] job {}/{} starting ({} commands, timeout {}s, workdir {})", + job_id.pipeline_id.0, + job_id.job_name, + self.start_job.job_def.run.len(), + timeout_secs, + work_dir, + ); + for cmd_str in &self.start_job.job_def.run { + eprintln!("[runner] exec: {cmd_str}"); + // Send progress: command being run. let _ = ctx.send( self.coordinator_addr, @@ -64,6 +75,7 @@ impl LocalRunner { let mut child = match child_result { Ok(c) => c, Err(e) => { + eprintln!("[runner] spawn failed: {e}"); let _ = ctx.send( self.coordinator_addr, LocalCoordinatorMsg::JobComplete(JobComplete { @@ -77,70 +89,36 @@ impl LocalRunner { } }; - // Timeout mechanism: share child handle, spawn thread that kills after timeout. + // Timeout: spawn a thread that kills the child after timeout_secs. + let child_id = child.id(); let kill_flag = Arc::new(Mutex::new(false)); let kill_flag_clone = Arc::clone(&kill_flag); - // Use a pipe to signal the timeout thread when the command finishes. let (done_tx, done_rx) = std::sync::mpsc::channel::<()>(); let timeout_handle = std::thread::spawn(move || { - // Wait for either timeout or command completion. - if done_rx.recv_timeout(std::time::Duration::from_secs(timeout_secs)).is_err() { + if done_rx + .recv_timeout(std::time::Duration::from_secs(timeout_secs)) + .is_err() + { *kill_flag_clone.lock().unwrap() = true; + // Actually kill the child process so the pipe readers unblock. + let _ = std::process::Command::new("kill") + .args(["-9", &child_id.to_string()]) + .status(); } }); - // Read stdout line-by-line. - let stdout = child.stdout.take(); - let stderr = child.stderr.take(); - - let mut last_lines: Vec = Vec::new(); - - if let Some(stdout) = stdout { - let reader = std::io::BufReader::new(stdout); - for line in reader.lines() { - if let Ok(line) = line { - let _ = ctx.send( - self.coordinator_addr, - LocalCoordinatorMsg::JobProgress(JobProgress { - job_id: job_id.clone(), - output_line: line.clone(), - }), - ); - last_lines.push(line); - if last_lines.len() > 50 { - last_lines.remove(0); - } - } - } - } - - if let Some(stderr) = stderr { - let reader = std::io::BufReader::new(stderr); - for line in reader.lines() { - if let Ok(line) = line { - let _ = ctx.send( - self.coordinator_addr, - LocalCoordinatorMsg::JobProgress(JobProgress { - job_id: job_id.clone(), - output_line: format!("[stderr] {line}"), - }), - ); - last_lines.push(line); - if last_lines.len() > 50 { - last_lines.remove(0); - } - } - } - } + // Read stdout and stderr concurrently to avoid pipe-buffer deadlock. + let (last_lines, timed_out) = + drain_child_output(&mut child, job_id, ctx, self.coordinator_addr, &kill_flag); let status = child.wait(); - // Signal timeout thread that command finished. + // Signal timeout thread that we're done. let _ = done_tx.send(()); let _ = timeout_handle.join(); - // Check if killed by timeout. - if *kill_flag.lock().unwrap() { + if timed_out || *kill_flag.lock().unwrap() { + eprintln!("[runner] command timed out after {timeout_secs}s"); let _ = ctx.send( self.coordinator_addr, LocalCoordinatorMsg::JobComplete(JobComplete { @@ -155,10 +133,14 @@ impl LocalRunner { match status { Ok(exit) if exit.success() => { - // Command passed, continue to next. + eprintln!("[runner] command succeeded"); } Ok(exit) => { let exit_code = exit.code().unwrap_or(-1); + eprintln!("[runner] command failed (exit {exit_code})"); + for line in &last_lines { + eprintln!("[runner] {line}"); + } let _ = ctx.send( self.coordinator_addr, LocalCoordinatorMsg::JobComplete(JobComplete { @@ -174,6 +156,7 @@ impl LocalRunner { return; } Err(e) => { + eprintln!("[runner] wait failed: {e}"); let _ = ctx.send( self.coordinator_addr, LocalCoordinatorMsg::JobComplete(JobComplete { @@ -188,7 +171,10 @@ impl LocalRunner { } } - // All commands passed. + eprintln!( + "[runner] job {}/{} passed", + job_id.pipeline_id.0, job_id.job_name + ); let _ = ctx.send( self.coordinator_addr, LocalCoordinatorMsg::JobComplete(JobComplete { @@ -201,6 +187,87 @@ impl LocalRunner { } } +/// Drain stdout and stderr from a child process concurrently. +/// +/// Spawns a background thread for stderr so that both pipes are consumed +/// in parallel, preventing the classic pipe-buffer deadlock where the child +/// blocks writing to a full stderr while the parent blocks reading stdout. +/// +/// Returns (last_lines, timed_out). +fn drain_child_output( + child: &mut Child, + job_id: &crate::JobId, + ctx: &Ctx, + coordinator_addr: ActorAddress, + kill_flag: &Arc>, +) -> (Vec, bool) { + let stdout = child.stdout.take(); + let stderr = child.stderr.take(); + + // Collect stderr on a background thread. + let stderr_job_id = job_id.clone(); + let stderr_kill = Arc::clone(kill_flag); + let stderr_handle = std::thread::spawn(move || { + let mut lines = Vec::new(); + if let Some(stderr) = stderr { + let reader = std::io::BufReader::new(stderr); + for line in reader.lines() { + if *stderr_kill.lock().unwrap() { + break; + } + if let Ok(line) = line { + lines.push(line); + } + } + } + lines + }); + + // Read stdout on the current thread, streaming progress. + let mut last_lines: Vec = Vec::new(); + if let Some(stdout) = stdout { + let reader = std::io::BufReader::new(stdout); + for line in reader.lines() { + if let Ok(line) = line { + let _ = ctx.send( + coordinator_addr, + LocalCoordinatorMsg::JobProgress(JobProgress { + job_id: job_id.clone(), + output_line: line.clone(), + }), + ); + last_lines.push(line); + if last_lines.len() > 50 { + last_lines.remove(0); + } + } + } + } + + // Join stderr thread and stream its lines as progress. + let timed_out = *kill_flag.lock().unwrap(); + let stderr_lines = stderr_handle.join().unwrap_or_default(); + for line in &stderr_lines { + let _ = ctx.send( + coordinator_addr, + LocalCoordinatorMsg::JobProgress(JobProgress { + job_id: stderr_job_id.clone(), + output_line: format!("[stderr] {line}"), + }), + ); + } + + // Merge stderr into last_lines tail. + for line in stderr_lines { + last_lines.push(line); + if last_lines.len() > 50 { + last_lines.remove(0); + } + } + + (last_lines, timed_out) +} + impl ActorInterface for LocalRunner { type Incoming = LocalRunnerMsg; type Response = (); diff --git a/crates/ci/src/yaml.rs b/crates/ci/src/yaml.rs index de50472..2bf5df4 100644 --- a/crates/ci/src/yaml.rs +++ b/crates/ci/src/yaml.rs @@ -429,4 +429,5 @@ pipelines: let multi = to_job_definition("multi", &check.jobs["multi"]); assert_eq!(multi.run, vec!["cargo fmt -- --check", "cargo test"]); } + }