fix(child_process): release timeout threads after child exit

changelog.d/9640-child-timeout-threads.md

@@ -0,0 +1,3 @@
+Fixed timed `child_process.spawn`, `exec`, and `execFile` calls retaining
+one sleeping OS thread until the full timeout after a child had already exited.
+Timeout workers now wake on child completion while preserving deadline kills.

crates/perry-runtime/src/child_process/reactor.rs

@@ -466,25 +466,6 @@ fn cp_spawn_reader<R: Read + Send + 'static>(handle: u64, mut pipe: R, fd: usize
     });
 }
 
-/// Spawn the waiter thread that reaps `child` and reports its exit status.
-fn cp_spawn_waiter(handle: u64, waiter: CpWaiter) {
-    std::thread::spawn(move || {
-        let (code, signal) = waiter();
-        cp_push_event(CpEvent::Exited {
-            handle,
-            code,
-            signal,
-        });
-    });
-}
-
-fn cp_spawn_timeout(handle: u64, timeout: Duration, signal: i32) {
-    std::thread::spawn(move || {
-        std::thread::sleep(timeout);
-        cp_push_event(CpEvent::Timeout { handle, signal });
-    });
-}
-
 /// IPC reader (#1933): read newline-delimited JSON from the parent socket and
 /// push each line for main-thread parse + `'message'` delivery. For
 /// `serialization: 'advanced'` (#2130) the framing is instead a 4-byte
@@ -773,10 +754,11 @@ fn cp_register_live_child_parts(
     for (fd, _, pipe) in extra_pipes {
         cp_spawn_reader(handle, pipe, fd);
     }
-    cp_spawn_waiter(handle, waiter);
-    if let Some(timeout) = timeout {
-        cp_spawn_timeout(handle, timeout, kill_signal);
-    }
+    cp_spawn_waiter(
+        handle,
+        waiter,
+        timeout.map(|timeout| (timeout, kill_signal)),
+    );
     #[cfg(any(unix, windows))]
     {
         if let Some(sock) = ipc {
@@ -1390,10 +1372,11 @@ pub(super) fn cp_exec_async(
                 }
                 Err(_) => (Some(-1), None),
             });
-            cp_spawn_waiter(handle, waiter);
-            if let Some(timeout) = timeout {
-                cp_spawn_timeout(handle, timeout, kill_signal);
-            }
+            cp_spawn_waiter(
+                handle,
+                waiter,
+                timeout.map(|timeout| (timeout, kill_signal)),
+            );
             crate::event_pump::js_notify_main_thread();
             cp
         }
@@ -1961,9 +1944,11 @@ pub(super) fn cp_live_stdin_queue_callback(handle: u64, callback_bits: u64) -> b
 mod integration;
 mod kill;
 mod stdin_drain;
+mod timeout;
 #[cfg(all(test, windows))]
 #[path = "reactor/windows_kill_tests.rs"]
 mod windows_kill_tests;
 pub(crate) use integration::*;
 pub(crate) use kill::*;
 use stdin_drain::*;
+use timeout::*;

crates/perry-runtime/src/child_process/reactor/timeout.rs

@@ -0,0 +1,89 @@
+//! Cancellable child-process timeout waiters.
+
+use super::*;
+use std::sync::mpsc::{self, RecvTimeoutError};
+
+/// Spawn the process waiter and, when configured, a timeout waiter. The
+/// process waiter owns the sole cancellation sender, so a completed child
+/// wakes its timeout waiter instead of leaving an OS thread asleep until the
+/// original deadline.
+pub(super) fn cp_spawn_waiter(handle: u64, waiter: CpWaiter, timeout: Option<(Duration, i32)>) {
+    let timeout_cancel = timeout.map(|(timeout, signal)| {
+        let (cancel, cancelled) = mpsc::channel();
+        std::thread::spawn(move || {
+            if matches!(
+                cancelled.recv_timeout(timeout),
+                Err(RecvTimeoutError::Timeout)
+            ) {
+                cp_push_event(CpEvent::Timeout { handle, signal });
+            }
+        });
+        cancel
+    });
+
+    std::thread::spawn(move || {
+        let (code, signal) = waiter();
+        if let Some(cancel) = timeout_cancel {
+            let _ = cancel.send(());
+        }
+        cp_push_event(CpEvent::Exited {
+            handle,
+            code,
+            signal,
+        });
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::{Arc, Barrier};
+    use std::time::Instant;
+
+    #[test]
+    fn completed_children_wake_fifty_long_timeout_waiters() {
+        const CHILDREN: usize = 50;
+        let ready = Arc::new(Barrier::new(CHILDREN + 1));
+        let mut cancels = Vec::with_capacity(CHILDREN);
+        let mut completions = Vec::with_capacity(CHILDREN);
+
+        for _ in 0..CHILDREN {
+            let (cancel, cancelled) = mpsc::channel();
+            let (completed, completion) = mpsc::channel();
+            let ready = Arc::clone(&ready);
+            std::thread::spawn(move || {
+                ready.wait();
+                let timed_out = matches!(
+                    cancelled.recv_timeout(Duration::from_secs(60)),
+                    Err(RecvTimeoutError::Timeout)
+                );
+                let _ = completed.send(timed_out);
+            });
+            cancels.push(cancel);
+            completions.push(completion);
+        }
+
+        ready.wait();
+        let started = Instant::now();
+        for cancel in cancels {
+            cancel.send(()).expect("timeout waiter should be alive");
+        }
+        for completion in completions {
+            assert_eq!(
+                completion.recv_timeout(Duration::from_secs(1)),
+                Ok(false),
+                "timeout waiter did not stop after child completion"
+            );
+        }
+        assert!(started.elapsed() < Duration::from_secs(1));
+    }
+
+    #[test]
+    fn elapsed_deadline_still_selects_the_timeout_arm() {
+        let (_cancel, cancelled) = mpsc::channel::<()>();
+        assert!(matches!(
+            cancelled.recv_timeout(Duration::from_millis(10)),
+            Err(RecvTimeoutError::Timeout)
+        ));
+    }
+}

test-files/test_gap_9592_child_timeout_threads.ts

@@ -0,0 +1,49 @@
+// #9592 — a timed spawn must not retain its timeout OS thread after the child
+// exits. The Linux thread census catches the old one-sleeper-per-timeout leak;
+// the slow-child arm keeps the actual timeout behavior covered everywhere.
+import { spawn } from "node:child_process";
+import { readdirSync } from "node:fs";
+
+/** Resolve after the child process and its stdio handles have closed. */
+function close(child: any): Promise<void> {
+  return new Promise((resolve) => child.on("close", () => resolve()));
+}
+
+/** Count this process's live OS threads when Linux exposes the task census. */
+function threadCount(): number {
+  return process.platform === "linux" ? readdirSync("/proc/self/task").length : 0;
+}
+
+const baseline = threadCount();
+const quickChildren: Promise<void>[] = [];
+for (let i = 0; i < 50; i++) {
+  quickChildren.push(
+    close(spawn("/bin/true", [], { stdio: "ignore", timeout: 60_000 })),
+  );
+}
+await Promise.all(quickChildren);
+
+let timeoutThreadsReleased = process.platform !== "linux";
+const releaseDeadline = Date.now() + 1_000;
+while (!timeoutThreadsReleased && Date.now() < releaseDeadline) {
+  timeoutThreadsReleased = threadCount() <= baseline + 5;
+  if (!timeoutThreadsReleased) {
+    await new Promise<void>((resolve) => setTimeout(resolve, 20));
+  }
+}
+console.log("timeout threads released:", timeoutThreadsReleased);
+
+const started = Date.now();
+const slow = spawn("/bin/sleep", ["30"], {
+  stdio: "ignore",
+  timeout: 100,
+});
+await close(slow);
+const elapsed = Date.now() - started;
+console.log(
+  "slow child killed on time:",
+  slow.killed &&
+    slow.signalCode === "SIGTERM" &&
+    elapsed >= 50 &&
+    elapsed < 5_000,
+);