fix: Restore task completion semantics

crates/turborepo-process/src/child.rs

@@ -18,9 +18,9 @@
 const CHILD_POLL_INTERVAL: Duration = Duration::from_micros(50);
 const POST_EXIT_OUTPUT_DRAIN_TIMEOUT: Duration = Duration::from_millis(100);
 #[cfg(any(unix, windows))]
 const PROCESS_TREE_DRAIN_POLL_INTERVAL: Duration = Duration::from_millis(10);
 #[cfg(windows)]
 const WINDOWS_DESCENDANT_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
 
 use std::{
     fmt,
@@ -581,7 +581,7 @@
     }
 
     #[cfg(windows)]
     fn has_active_windows_job(&self) -> bool {
         self._job
             .as_ref()
             .is_some_and(|job| match job.active_processes() {
@@ -922,6 +922,8 @@
 #[derive(Clone, Debug)]
 pub struct Child {
     pid: Option<u32>,
+    #[cfg(unix)]
+    target_identity: Option<TargetIdentity>,
     command_channel: ChildCommandChannel,
     exit_channel: watch::Receiver<Option<ChildExit>>,
     stdin: Arc<Mutex<Option<ChildInput>>>,
@@ -984,6 +986,8 @@
         }?;
 
         let pid = child.pid();
+        #[cfg(unix)]
+        let target_identity = child.target_identity;
 
         let (command_tx, mut command_rx) = ChildCommandChannel::new();
 
@@ -1012,45 +1016,7 @@
                 }
                 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 @@
 
         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 @@
         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<ChildExit>) {
+        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 @@
 
         let (status, write_result) = tokio::join!(self.wait(), writer_fut);
         write_result?;
+        self.cleanup_if_successful(status);
 
         Ok(status)
     }
@@ -1324,7 +1332,9 @@
         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 {

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::<libc::pid_t>()
+            {
+                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;
     }
 }

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];