From 7be6ab7ed076cde9444df6fdaf4e00025de24228 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Mon, 16 Feb 2026 14:52:24 +0700 Subject: [PATCH] fix: resolve pipe-buffer deadlock and add runner observability Read stdout and stderr concurrently (background thread for stderr) to prevent the classic deadlock where cargo blocks writing stderr while the parent blocks reading stdout. Timeout now actually kills the child process via kill -9. Added eprintln logging for all job lifecycle events so failures are visible in local-runner.log. Also switches .ci.yml to use $HOME instead of ~ for shell portability. Authored by Claude, lovingly guided by Zachery Aaron Shores-Chmielewski --- .ci.yml | 6 +- crates/ci/src/local_runner.rs | 173 +++++++++++++++++++++++----------- crates/ci/src/yaml.rs | 1 + 3 files changed, 124 insertions(+), 56 deletions(-) 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"]); } + }