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)); +}