From 0221805a7396a3ae6ba02432d2e50420101f6436 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 10 Apr 2026 15:20:41 -0400 Subject: [PATCH] fix(cli): let detached workers exit after post-run shutdown Move worker control stdin handling off Tokio's blocking shutdown path so subprocess workers can exit cleanly after success or cooperative cancellation even when the parent still holds stdin open. Add regression coverage for retro-enabled success and SIGTERM-driven cancellation with stdin intentionally left open. --- .../fabro-cli/src/commands/run/runner.rs | 126 +++++++++-- lib/crates/fabro-cli/tests/it/cmd/runner.rs | 197 +++++++++++++++++- 2 files changed, 299 insertions(+), 24 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index 4803153ce..fe6cbffc3 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -1,3 +1,4 @@ +use std::io::{BufRead as StdBufRead, BufReader as StdBufReader}; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; @@ -16,10 +17,9 @@ use fabro_workflow::event::{Emitter, RunEventSink}; use fabro_workflow::operations::{self, StartServices}; use fabro_workflow::run_control::RunControlState; use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle}; -use tokio::io::{self, AsyncBufReadExt, AsyncRead, BufReader}; #[cfg(unix)] use tokio::signal::unix::{SignalKind, signal}; -use tokio::sync::Mutex; +use tokio::sync::{Mutex, mpsc}; use tokio::time::sleep; use crate::args::RunWorkerMode; @@ -72,11 +72,7 @@ pub(crate) async fn execute( ))); let interviewer = Arc::new(ControlInterviewer::new()); let cancel_token = Arc::new(AtomicBool::new(false)); - tokio::spawn(read_worker_control_stream( - io::stdin(), - Arc::clone(&interviewer), - Arc::clone(&cancel_token), - )); + spawn_worker_control_stream(Arc::clone(&interviewer), Arc::clone(&cancel_token))?; let run_control = RunControlState::new(); install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?; let github_app = maybe_build_github_app_credentials(&run_record.settings)?; @@ -112,24 +108,79 @@ pub(crate) async fn execute( Ok(()) } -async fn read_worker_control_stream( - reader: R, +#[derive(Debug, PartialEq, Eq)] +enum WorkerControlStreamEvent { + Line(String), + Eof, +} + +fn spawn_worker_control_stream( interviewer: Arc, cancel_token: Arc, +) -> Result<()> { + let (event_tx, event_rx) = mpsc::unbounded_channel(); + tokio::spawn(handle_worker_control_stream_events( + interviewer, + cancel_token, + event_rx, + )); + std::thread::Builder::new() + .name("fabro-worker-control".to_string()) + .spawn(move || { + read_worker_control_stream_blocking(StdBufReader::new(std::io::stdin()), event_tx); + }) + .context("failed to spawn worker control reader thread")?; + Ok(()) +} + +fn read_worker_control_stream_blocking( + mut reader: R, + event_tx: mpsc::UnboundedSender, ) where - R: AsyncRead + Unpin, + R: StdBufRead, { - let mut lines = BufReader::new(reader).lines(); + let mut line = String::new(); loop { - if let Ok(Some(line)) = lines.next_line().await { - apply_worker_control_line(&interviewer, &cancel_token, &line).await; - } else { - interviewer.interrupt_all().await; - break; + line.clear(); + match reader.read_line(&mut line) { + Ok(0) => { + let _ = event_tx.send(WorkerControlStreamEvent::Eof); + break; + } + Ok(_) => { + let line = line.trim_end_matches(['\r', '\n']).to_string(); + if event_tx.send(WorkerControlStreamEvent::Line(line)).is_err() { + break; + } + } + Err(_) => { + let _ = event_tx.send(WorkerControlStreamEvent::Eof); + break; + } } } } +async fn handle_worker_control_stream_events( + interviewer: Arc, + cancel_token: Arc, + mut event_rx: mpsc::UnboundedReceiver, +) { + while let Some(event) = event_rx.recv().await { + match event { + WorkerControlStreamEvent::Line(line) => { + apply_worker_control_line(&interviewer, &cancel_token, &line).await; + } + WorkerControlStreamEvent::Eof => { + interviewer.interrupt_all().await; + return; + } + } + } + + interviewer.interrupt_all().await; +} + async fn apply_worker_control_line( interviewer: &ControlInterviewer, cancel_token: &AtomicBool, @@ -494,9 +545,9 @@ mod tests { use std::sync::atomic::{AtomicBool, Ordering}; use super::{ - MissingArtifactUploadTokenUploader, WorkerTitlePhase, apply_worker_control_line, - initial_worker_title_phase, read_worker_control_stream, worker_title, - worker_title_phase_for_event, + MissingArtifactUploadTokenUploader, WorkerControlStreamEvent, WorkerTitlePhase, + apply_worker_control_line, handle_worker_control_stream_events, initial_worker_title_phase, + read_worker_control_stream_blocking, worker_title, worker_title_phase_for_event, }; use crate::args::RunWorkerMode; use fabro_interview::{AnswerValue, ControlInterviewer, Interviewer, Question, QuestionType}; @@ -664,18 +715,49 @@ mod tests { } #[tokio::test] - async fn worker_control_stream_eof_interrupts_pending_interviews() { + async fn blocking_worker_control_stream_emits_lines_and_eof() { + let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel(); + + read_worker_control_stream_blocking( + std::io::Cursor::new( + b"{\"v\":1,\"type\":\"run.cancel\"}\n{\"v\":1,\"type\":\"interview.answer\",\"qid\":\"q-1\",\"answer\":{\"kind\":\"yes\"}}\n", + ), + event_tx, + ); + + assert_eq!( + event_rx.try_recv(), + Ok(WorkerControlStreamEvent::Line( + r#"{"v":1,"type":"run.cancel"}"#.to_string() + )) + ); + assert_eq!( + event_rx.try_recv(), + Ok(WorkerControlStreamEvent::Line( + r#"{"v":1,"type":"interview.answer","qid":"q-1","answer":{"kind":"yes"}}"# + .to_string() + )) + ); + assert_eq!(event_rx.try_recv(), Ok(WorkerControlStreamEvent::Eof)); + } + + #[tokio::test] + async fn worker_control_event_loop_eof_interrupts_pending_interviews() { let interviewer = Arc::new(ControlInterviewer::new()); let cancel_token = Arc::new(AtomicBool::new(false)); let mut question = Question::new("Approve?", QuestionType::YesNo); question.id = "q-1".to_string(); let ask_interviewer = Arc::clone(&interviewer); let answer_task = tokio::spawn(async move { ask_interviewer.ask(question).await }); + let (event_tx, event_rx) = tokio::sync::mpsc::unbounded_channel(); - read_worker_control_stream( - tokio::io::empty(), + event_tx.send(WorkerControlStreamEvent::Eof).unwrap(); + drop(event_tx); + + handle_worker_control_stream_events( Arc::clone(&interviewer), Arc::clone(&cancel_token), + event_rx, ) .await; diff --git a/lib/crates/fabro-cli/tests/it/cmd/runner.rs b/lib/crates/fabro-cli/tests/it/cmd/runner.rs index 36a049425..11d50bed6 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/runner.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/runner.rs @@ -1,9 +1,16 @@ +use std::io::Read; +use std::process::{Child, ExitStatus, Output, Stdio}; +use std::time::{Duration, Instant}; + use fabro_store::EventEnvelope; use fabro_test::{fabro_snapshot, test_context}; -use fabro_types::{EventBody, RunEvent}; +use fabro_types::{EventBody, RunEvent, StatusReason}; use httpmock::MockServer; -use super::support::{output_stderr, run_events, run_state, server_target}; +use super::support::{ + output_stderr, run_events, run_state, server_target, wait_for_event_names, wait_for_status, + write_gated_workflow, +}; use crate::support::{fabro_json_snapshot, unique_run_id}; const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); @@ -28,6 +35,69 @@ fn assert_worker_succeeded(run_dir: &std::path::Path, stdout: &[u8]) { ))); } +fn spawn_worker_process( + context: &fabro_test::TestContext, + server: &str, + run_dir: &std::path::Path, + run_id: &str, + mode: &str, +) -> Child { + let mut cmd = std::process::Command::new(env!("CARGO_BIN_EXE_fabro")); + cmd.current_dir(&context.temp_dir); + cmd.env("NO_COLOR", "1"); + cmd.env("HOME", &context.home_dir); + cmd.env("FABRO_NO_UPGRADE_CHECK", "true"); + cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64"); + cmd.env("FABRO_TEST_IN_MEMORY_STORE", "1"); + cmd.args([ + "__run-worker", + "--server", + server, + "--run-dir", + run_dir.to_str().unwrap(), + "--run-id", + run_id, + "--mode", + mode, + ]); + cmd.stdin(Stdio::piped()); + cmd.stdout(Stdio::piped()); + cmd.stderr(Stdio::piped()); + cmd.spawn().expect("worker should spawn") +} + +fn wait_for_child_exit(child: &mut Child, timeout: Duration) -> ExitStatus { + let deadline = Instant::now() + timeout; + loop { + if let Some(status) = child.try_wait().expect("worker wait should succeed") { + return status; + } + assert!( + Instant::now() < deadline, + "timed out waiting for worker to exit" + ); + std::thread::sleep(Duration::from_millis(50)); + } +} + +fn child_output(mut child: Child, status: ExitStatus) -> Output { + let mut stdout = Vec::new(); + let mut stderr = Vec::new(); + if let Some(mut pipe) = child.stdout.take() { + pipe.read_to_end(&mut stdout) + .expect("worker stdout should be readable"); + } + if let Some(mut pipe) = child.stderr.take() { + pipe.read_to_end(&mut stderr) + .expect("worker stderr should be readable"); + } + Output { + status, + stdout, + stderr, + } +} + fn server_endpoint(storage_dir: &std::path::Path) -> (reqwest::Client, String) { let target = server_target(storage_dir); if target.starts_with('/') { @@ -557,3 +627,126 @@ fn detached_run_answers_pending_question_without_interview_scratch_files() { if props.question_id == question_id && props.answer == "A" ))); } + +#[test] +fn worker_exits_with_retro_enabled_even_when_stdin_stays_open() { + let context = test_context!(); + let run_id = unique_run_id(); + let workflow_path = context.temp_dir.join("retro-success.fabro"); + + context.write_temp( + "fabro.toml", + r#"_version = 1 + +[run.execution] +retros = true +"#, + ); + context.write_temp( + "retro-success.fabro", + r#"digraph RetroSuccess { + graph [goal="Finish successfully with retro enabled"] + start [shape=Mdiamond, label="Start"] + exit [shape=Msquare, label="Exit"] + work [shape=parallelogram, label="Work", script="true"] + start -> work -> exit +} +"#, + ); + + context + .command() + .args([ + "create", + "--dry-run", + "--auto-approve", + "--run-id", + run_id.as_str(), + workflow_path.to_str().unwrap(), + ]) + .assert() + .success(); + + let run_dir = context.find_run_dir(&run_id); + let server = server_target(&context.storage_dir); + let mut child = spawn_worker_process(&context, &server, &run_dir, &run_id, "start"); + let stdin = child.stdin.take().expect("worker stdin should be piped"); + + wait_for_event_names(&run_dir, &["run.completed", "retro.completed"]); + let status = wait_for_child_exit(&mut child, SHARED_DAEMON_TIMEOUT); + drop(stdin); + let output = child_output(child, status); + + assert!( + output.status.success(), + "worker should exit successfully after retro even with stdin open:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + let events = stored_worker_events(&run_dir); + let run_completed_index = events + .iter() + .position(|event| matches!(&event.body, EventBody::RunCompleted(_))) + .expect("run.completed should be present"); + let retro_completed_index = events + .iter() + .position(|event| matches!(&event.body, EventBody::RetroCompleted(_))) + .expect("retro.completed should be present"); + assert!( + run_completed_index < retro_completed_index, + "retro should still run after run.completed" + ); +} + +#[cfg(unix)] +#[test] +fn worker_exits_after_sigterm_cancel_even_when_stdin_stays_open() { + let context = test_context!(); + let run_id = unique_run_id(); + let workflow_path = context.temp_dir.join("cancel-gated.fabro"); + let _gate = write_gated_workflow(&workflow_path, "cancel_gated", "Wait for cancellation"); + + context + .command() + .args([ + "create", + "--auto-approve", + "--no-retro", + "--sandbox", + "local", + "--run-id", + run_id.as_str(), + workflow_path.to_str().unwrap(), + ]) + .assert() + .success(); + + let run_dir = context.find_run_dir(&run_id); + let server = server_target(&context.storage_dir); + let mut child = spawn_worker_process(&context, &server, &run_dir, &run_id, "start"); + let stdin = child.stdin.take().expect("worker stdin should be piped"); + + wait_for_event_names(&run_dir, &["run.running"]); + let worker_pid = child.id(); + assert!(worker_pid > 0, "worker pid should be present"); + fabro_proc::sigterm(worker_pid); + + wait_for_status(&run_dir, &["failed"]); + let status = wait_for_child_exit(&mut child, SHARED_DAEMON_TIMEOUT); + drop(stdin); + let output = child_output(child, status); + + assert!( + output.status.success(), + "worker should exit cleanly after SIGTERM cancellation:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + let status_record = run_state(&run_dir) + .status + .expect("cancelled run should have a status record"); + assert_eq!(status_record.status.to_string(), "failed"); + assert_eq!(status_record.reason, Some(StatusReason::Cancelled)); +}