From 1ef6ef1cc95b277e56fa95e5e3ae8f19938954cf Mon Sep 17 00:00:00 2001 From: Anthony Shew Date: Thu, 21 May 2026 14:50:55 -0600 Subject: [PATCH] fix: Restore task completion semantics --- crates/turborepo-process/src/child.rs | 90 ++++++++++--------- crates/turborepo-process/src/lib.rs | 59 ++++++------ .../test/scripts/normal_exit_worker.js | 13 ++- 3 files changed, 88 insertions(+), 74 deletions(-) diff --git a/crates/turborepo-process/src/child.rs b/crates/turborepo-process/src/child.rs index 96505a2b3ce4c..9a8cd42f023ac 100644 --- a/crates/turborepo-process/src/child.rs +++ b/crates/turborepo-process/src/child.rs @@ -922,6 +922,8 @@ struct ChildStateManager { #[derive(Clone, Debug)] pub struct Child { pid: Option, + #[cfg(unix)] + target_identity: Option, command_channel: ChildCommandChannel, exit_channel: watch::Receiver>, stdin: Arc>>, @@ -984,6 +986,8 @@ impl Child { }?; let pid = child.pid(); + #[cfg(unix)] + let target_identity = child.target_identity; let (command_tx, mut command_rx) = ChildCommandChannel::new(); @@ -1012,45 +1016,7 @@ impl Child { } status = child.wait() => { drop(controller); - let should_drain_process_tree = matches!(&status, Ok(Some(0))); - - #[cfg(unix)] - let killed_during_drain = if should_drain_process_tree - && let Some(pid) = pid - { - let mut command_rx_open = true; - child - .wait_for_process_group_exit( - pid as libc::pid_t, - None, - &mut command_rx, - &mut command_rx_open, - ) - .await - == ChildExit::Killed - } else { - false - }; - - #[cfg(windows)] - let killed_during_drain = if should_drain_process_tree { - let mut command_rx_open = true; - child - .wait_for_job_exit(&mut command_rx, &mut command_rx_open) - .await - == ChildExit::Killed - } else { - false - }; - - #[cfg(not(any(unix, windows)))] - let killed_during_drain = false; - - if killed_during_drain { - manager.exit_tx.send(Some(ChildExit::Killed)).ok(); - } else { - manager.handle_child_exit(status).await; - } + manager.handle_child_exit(status).await; } } @@ -1059,6 +1025,8 @@ impl Child { Ok(Self { pid, + #[cfg(unix)] + target_identity, command_channel: command_tx, exit_channel: exit_rx, stdin: Arc::new(Mutex::new(stdin)), @@ -1106,6 +1074,45 @@ impl Child { self.pid } + #[cfg(unix)] + fn cleanup_process_scope_after_success(&self) { + let Some(identity) = self.target_identity else { + return; + }; + + let Some(pid) = self.pid else { + return; + }; + + if process_group_matches_identity(pid as libc::pid_t, identity) { + debug!( + "cleaning up remaining process group after successful task: {}", + identity.process_group_id + ); + signal_process_group(identity.process_group_id, libc::SIGKILL); + } + } + + #[cfg(windows)] + fn cleanup_process_scope_after_success(&self) { + let Some(pid) = self.pid else { + return; + }; + + if let Err(err) = super::job_object::terminate_descendant_processes(pid) { + debug!("failed to clean up descendants after successful task {pid}: {err}"); + } + } + + #[cfg(not(any(unix, windows)))] + fn cleanup_process_scope_after_success(&self) {} + + fn cleanup_if_successful(&self, status: Option) { + if status == Some(ChildExit::Finished(Some(0))) { + self.cleanup_process_scope_after_success(); + } + } + pub(crate) fn has_exited(&self) -> bool { self.exit_channel.borrow().is_some() } @@ -1216,6 +1223,7 @@ impl Child { let (status, write_result) = tokio::join!(self.wait(), writer_fut); write_result?; + self.cleanup_if_successful(status); Ok(status) } @@ -1324,7 +1332,9 @@ impl Child { debug_assert!(stdout_buffer.is_empty(), "buffer should be empty"); debug_assert!(stderr_buffer.is_empty(), "buffer should be empty"); - Ok(exit_status.or(self.wait().await)) + let status = exit_status.or(self.wait().await); + self.cleanup_if_successful(status); + Ok(status) } pub fn label(&self) -> &str { diff --git a/crates/turborepo-process/src/lib.rs b/crates/turborepo-process/src/lib.rs index 7c7b82d9ce366..8177be04fbcba 100644 --- a/crates/turborepo-process/src/lib.rs +++ b/crates/turborepo-process/src/lib.rs @@ -708,13 +708,18 @@ mod test { let _ = fs::remove_file(pid_file); } + #[cfg(unix)] #[tokio::test] - async fn test_normal_exit_waits_for_worker_before_completing() { + async fn test_successful_task_cleans_up_long_lived_worker_after_output_drain() { let manager = ProcessManager::new(false); - let marker_file = - std::env::temp_dir().join(format!("turbo-normal-exit-marker-{}", std::process::id())); - let pid_file = - std::env::temp_dir().join(format!("turbo-normal-exit-worker-{}", std::process::id())); + let marker_file = std::env::temp_dir().join(format!( + "turbo-normal-exit-marker-never-{}", + std::process::id() + )); + let pid_file = std::env::temp_dir().join(format!( + "turbo-normal-exit-worker-never-{}", + std::process::id() + )); let _ = fs::remove_file(&marker_file); let _ = fs::remove_file(&pid_file); @@ -723,49 +728,43 @@ mod test { std::ffi::OsString::from("./test/scripts/normal_exit_worker.js"), marker_file.as_os_str().to_os_string(), pid_file.as_os_str().to_os_string(), - std::ffi::OsString::from("800"), + std::ffi::OsString::from("never"), ]); let mut child = manager .spawn(command, Duration::from_secs(30), test_task_id()) .unwrap() .unwrap(); + let mut worker_pid = None; for _ in 0..50 { - if pid_file.exists() { + if let Ok(contents) = fs::read_to_string(&pid_file) + && let Ok(pid) = contents.trim().parse::() + { + worker_pid = Some(pid); break; } sleep(Duration::from_millis(20)).await; } + let worker_pid = worker_pid.expect("worker pid file should have been written"); + assert!( - pid_file.exists(), - "worker pid file should have been written" + process_exists(worker_pid), + "worker process {worker_pid} should be alive before output drain" ); - let start = Instant::now(); - let exit = child.wait().await; - let elapsed = start.elapsed(); - let marker_existed_when_wait_returned = marker_file.exists(); - - if !marker_existed_when_wait_returned { - for _ in 0..100 { - if marker_file.exists() { - break; - } - sleep(Duration::from_millis(20)).await; - } - } + let mut output = Vec::new(); + let exit = tokio::time::timeout( + Duration::from_secs(5), + child.wait_with_piped_outputs(&mut output), + ) + .await + .expect("successful output drain should not hang on a long-lived worker") + .expect("output drain should succeed"); let _ = fs::remove_file(&marker_file); let _ = fs::remove_file(&pid_file); assert_eq!(exit, Some(ChildExit::Finished(Some(0)))); - assert!( - marker_existed_when_wait_returned, - "child.wait() returned before the worker wrote its output" - ); - assert!( - elapsed >= Duration::from_millis(700), - "child.wait() should have blocked for the worker; only waited {elapsed:?}" - ); + wait_for_process_to_exit(worker_pid).await; } } diff --git a/crates/turborepo-process/test/scripts/normal_exit_worker.js b/crates/turborepo-process/test/scripts/normal_exit_worker.js index bfdbcdfbd7972..ba366f62edeb5 100644 --- a/crates/turborepo-process/test/scripts/normal_exit_worker.js +++ b/crates/turborepo-process/test/scripts/normal_exit_worker.js @@ -5,11 +5,16 @@ const mode = process.argv[2]; if (mode === "worker") { const markerPath = process.argv[3]; - const delayMs = Number(process.argv[4]); - setTimeout(() => { - fs.writeFileSync(markerPath, "done\n"); - }, delayMs); + if (process.argv[4] === "never") { + setInterval(() => {}, 1000); + } else { + const delayMs = Number(process.argv[4]); + + setTimeout(() => { + fs.writeFileSync(markerPath, "done\n"); + }, delayMs); + } } else { const markerPath = process.argv[2]; const pidPath = process.argv[3];