fix(core): prevent pool.close() hang on forks pool (#1285)

Author: fi3ework

packages/core/src/pool/poolRunner.ts

@@ -13,7 +13,6 @@ import {
 import type { PoolTask } from './types';
 
 const WORKER_START_TIMEOUT_MS = 90_000;
-const WORKER_STOP_TIMEOUT_MS = 60_000;
 const MAX_STDERR_MESSAGE_BYTES = 64 * 1024;
 
 function formatCapturedStderr(text: string): string {
@@ -90,8 +89,6 @@ export class PoolRunner {
   private startDeferred: Deferred | undefined;
   private stopDeferred: Deferred | undefined;
   private startTimer: NodeJS.Timeout | undefined;
-  private stopTimer: NodeJS.Timeout | undefined;
-  private forceKillTimer: NodeJS.Timeout | undefined;
   private lastFatalError: Error | undefined;
   /**
    * Set when the worker reports `fatal_error` or a transport error. The
@@ -180,14 +177,23 @@ export class PoolRunner {
     return this.runTaskInternal('collect', task) as Promise<CollectTaskResult>;
   }
 
+  /**
+   * Host owns termination — no IPC handshake. Per-task teardown runs in
+   * `runInPool`'s own `finally` before `runFinished`, so by `stop()` there
+   * is nothing process-level to drain. Relying on the worker's own
+   * `process.exit()` was the rstest#1275 hang.
+   */
   stop(options?: { force?: boolean }): Promise<void> {
     return this.runOperation(async () => {
       switch (this.state) {
         case 'STOPPED':
         case 'IDLE':
           return;
         case 'STOPPING': {
-          // Wait for the in-flight stop, then optionally force.
+          // Wait for the in-flight stop to settle. If the caller asks for
+          // `force` and the prior stop was graceful, escalate to SIGKILL —
+          // the prior `await` may have resolved without actually killing
+          // the child (e.g. SIGTERM masked).
           if (this.stopDeferred) {
             await this.stopDeferred.promise;
           }
@@ -216,34 +222,7 @@ export class PoolRunner {
       this.state = 'STOPPING';
       this.stopDeferred = createDeferred();
 
-      // Best-effort graceful stop. The worker defers its own exit until
-      // after any in-flight task completes teardown.
-      try {
-        this.worker.send({ type: 'stop' });
-      } catch {
-        // ignore: worker may already be down
-      }
-
-      // Escalate to SIGTERM after the budget expires, then SIGKILL shortly
-      // after. On Windows, SIGTERM is unconditionally fatal (the process
-      // cannot trap it), so sending it immediately would kill workers with
-      // in-flight tasks. Instead, rely on the IPC `stop` message for
-      // graceful shutdown and only signal when the worker fails to exit in
-      // time.
-      this.stopTimer = setTimeout(() => {
-        void this.worker.stop().catch(() => undefined);
-      }, WORKER_STOP_TIMEOUT_MS);
-      this.stopTimer.unref();
-      // Arm a hard SIGKILL fallback independently so it fires even if
-      // SIGTERM is trapped/ignored and `worker.stop()` never resolves.
-      this.forceKillTimer = setTimeout(() => {
-        void this.worker.stop({ force: true }).catch(() => undefined);
-      }, WORKER_STOP_TIMEOUT_MS + 5_000);
-      this.forceKillTimer.unref();
-
-      if (options?.force) {
-        await this.worker.stop({ force: true });
-      }
+      await this.worker.stop({ force: options?.force ?? false });
       await this.stopDeferred.promise;
     });
   }
@@ -351,10 +330,6 @@ export class PoolRunner {
       case 'collectFinished':
         this.resolveTask('collect', response.taskId, response.result);
         return;
-      case 'stopped':
-        // Worker acknowledged graceful shutdown — actual transition happens
-        // in `handleExit`.
-        return;
       case 'fatal_error': {
         const error = deserializeError(response.error);
         // Mark as crashed BEFORE rejecting. The host's dispatch unwinds via
@@ -384,7 +359,6 @@ export class PoolRunner {
   }
 
   private handleExit(code: number | null, signal: NodeJS.Signals | null): void {
-    this.clearStopTimer();
     this.clearStartTimer();
 
     const wasStopping = this.state === 'STOPPING';
@@ -470,15 +444,4 @@ export class PoolRunner {
     clearTimeout(this.startTimer);
     this.startTimer = undefined;
   }
-
-  private clearStopTimer(): void {
-    if (this.stopTimer) {
-      clearTimeout(this.stopTimer);
-      this.stopTimer = undefined;
-    }
-    if (this.forceKillTimer) {
-      clearTimeout(this.forceKillTimer);
-      this.forceKillTimer = undefined;
-    }
-  }
 }

packages/core/src/pool/protocol.ts

@@ -20,8 +20,7 @@ export type WorkerRequest =
       type: 'collect';
       taskId: number;
       options: RunWorkerOptions['options'];
-    }
-  | { type: 'stop' };
+    };
 
 export type CollectTaskResult = {
   tests: Test[];
@@ -49,7 +48,6 @@ export type WorkerResponse =
       result: CollectTaskResult;
       memory?: WorkerMemoryReport;
     }
-  | { type: 'stopped' }
   | {
       type: 'fatal_error';
       error: SerializedError;

packages/core/src/pool/workers/forksPoolWorker.ts

@@ -10,6 +10,12 @@ const BENIGN_IPC_ERROR_CODES = new Set([
   'ERR_STREAM_WRITE_AFTER_END',
 ]);
 
+/**
+ * SIGKILL escalation budget after SIGTERM. Defensive guard for native
+ * modules that mask SIGTERM.
+ */
+const SIGKILL_FALLBACK_MS = 500;
+
 /**
  * IPC errors that surface during shutdown but reflect the channel already
  * going away, not a genuine failure. Windows additionally surfaces
@@ -93,8 +99,7 @@ export class ForksPoolWorker extends BasePoolWorker {
       // Use `exit`, not `close`. `close` waits for all processes holding
       // the stdio pipes to release them — if a test spawns a subprocess
       // that inherits the worker's stdout/stderr, `close` blocks until
-      // that grandchild exits too, stalling slot reclaim and potentially
-      // hanging the pool until WORKER_STOP_TIMEOUT_MS force-kills.
+      // that grandchild exits too, stalling slot reclaim.
       child.on('exit', (code, signal) => {
         this.exited = true;
         this.emitter.emit('exit', code, signal);
@@ -113,10 +118,11 @@ export class ForksPoolWorker extends BasePoolWorker {
 
   async stop(options?: { force?: boolean }): Promise<void> {
     if (!this.hasLiveChild()) return;
-    await killAndWait(
-      this.childProcess!,
-      options?.force ? 'SIGKILL' : 'SIGTERM',
-    );
+    if (options?.force) {
+      await killAndWait(this.childProcess!, 'SIGKILL');
+      return;
+    }
+    await killAndWait(this.childProcess!, 'SIGTERM', SIGKILL_FALLBACK_MS);
   }
 
   sendRaw(envelope: Envelope): void {

packages/core/src/runtime/worker/index.ts

@@ -14,14 +14,6 @@ const send = (response: WorkerResponse): void => {
 };
 
 let currentTaskId: number | undefined;
-let stopRequested = false;
-/**
- * Set when a stop arrived mid-task. The task handler drains teardown in its
- * own `finally` first; only then do we flush `stopped` and exit. Without the
- * deferral `process.exit(0)` would orphan env/coverage/mock cleanup and the
- * `process.exit` / `process.kill` restoration.
- */
-let exitOnTaskIdle = false;
 /**
  * Set when a task handler has reported `fatal_error` and is on its way to
  * exit. Suppresses the bottom-of-the-stack `fatalExit` from racing in with a
@@ -36,11 +28,6 @@ const sendFatalError = (err: unknown): void => {
   });
 };
 
-const finalizeStop = (): void => {
-  send({ type: 'stopped' });
-  setImmediate(() => process.exit(0));
-};
-
 /**
  * Last-resort handlers. The runtime's `runInPool` registers its own
  * uncaught/unhandled handlers that capture errors thrown WHILE a test is
@@ -110,30 +97,12 @@ const runTask = async (
   }
 
   currentTaskId = undefined;
-  if (exitOnTaskIdle) finalizeStop();
-};
-
-const requestGracefulStop = (): void => {
-  if (stopRequested) return;
-  stopRequested = true;
-  if (currentTaskId !== undefined) {
-    // Defer the ack + exit until the task's `finally { await teardown(); }`
-    // runs. PoolRunner's WORKER_STOP_TIMEOUT_MS escalates to SIGKILL if
-    // teardown truly hangs, so this deferral is bounded.
-    exitOnTaskIdle = true;
-    return;
-  }
-  finalizeStop();
 };
 
-// SIGTERM shares the same shutdown path under the forks pool. `PoolRunner.stop`
-// sends SIGTERM alongside the `stop` envelope; without this handler Node's
-// default SIGTERM behavior would terminate the process immediately and skip
-// teardown. Under the threads pool the host uses `worker.terminate()` instead
-// of signals, so this handler is a no-op there. `setup.ts` may install a
-// profiling-specific SIGTERM handler (`--cpu-prof` et al.) that runs first and
-// preempts ours, preserving existing profiling semantics.
-process.on('SIGTERM', requestGracefulStop);
+// No SIGTERM handler — the host owns termination and SIGTERM (default action:
+// exit) is what gets us out. Any handler that didn't unconditionally exit
+// would defeat that contract (rstest#1275). `setup.ts` may install a
+// profiling-specific handler that calls `process.exit()`, which is compatible.
 
 channel.on((message: unknown) => {
   if (!isWorkerRequestEnvelope(message)) {
@@ -152,8 +121,5 @@ channel.on((message: unknown) => {
     case 'collect':
       void runTask('collect', request);
       break;
-    case 'stop':
-      requestGracefulStop();
-      break;
   }
 });

packages/core/tests/pool/fixtures/testWorker.mjs

@@ -79,32 +79,18 @@ const makeRunResult = (request, extra) => ({
   ...extra,
 });
 
-// Mirrors the real worker: track whether a task is in-flight and defer
-// stop until the task completes.
-let taskInFlight = false;
-let exitOnTaskIdle = false;
-
-const finalizeStop = () => {
-  send({ type: 'stopped' });
-  setTimeout(() => process.exit(0), 10);
-};
-
 const handleRun = (request) => {
   const mode = request.options?.__testMode;
-  taskInFlight = true;
 
   const finish = (extra) => {
     send({
       type: 'runFinished',
       taskId: request.taskId,
       result: makeRunResult(request, extra),
     });
-    taskInFlight = false;
-    if (exitOnTaskIdle) finalizeStop();
   };
 
   if (mode === 'fatal') {
-    taskInFlight = false;
     send({
       type: 'fatal_error',
       error: {
@@ -189,18 +175,6 @@ const handleCollect = (request) => {
   });
 };
 
-let stopRequested = false;
-const requestGracefulStop = () => {
-  if (stopRequested) return;
-  stopRequested = true;
-  if (taskInFlight) {
-    // Defer until the in-flight task finishes, just like the real worker.
-    exitOnTaskIdle = true;
-    return;
-  }
-  finalizeStop();
-};
-
 onHostMessage((message) => {
   if (!message || message[REQ_TAG] !== true) return;
   const request = message.request;
@@ -216,13 +190,7 @@ onHostMessage((message) => {
     case 'collect':
       handleCollect(request);
       break;
-    case 'stop':
-      requestGracefulStop();
-      break;
   }
 });
 
-// SIGTERM is meaningful only under the forks pool. In threads mode signals
-// are delivered to the parent process, not the worker thread, so this
-// handler is a harmless no-op there.
-process.on('SIGTERM', requestGracefulStop);
+// No SIGTERM handler — mirrors worker/index.ts.

packages/core/tests/pool/pool.test.ts

@@ -244,7 +244,7 @@ describe('Pool - exit-based lifecycle (not close)', () => {
       // Task 1: worker spawns a 30s grandchild that inherits stdout/stderr,
       // then sends its result and exits. With a `close`-based lifecycle this
       // would block slot reclaim until the grandchild exits (30s), causing
-      // task 2 to hang until WORKER_STOP_TIMEOUT_MS.
+      // task 2 to hang.
       const start = Date.now();
       const r1 = await pool.runTest(
         createTask('run', { __testMode: 'spawn-orphan' }),
@@ -312,33 +312,29 @@ describe('Pool - failure recovery', () => {
 // ── close() behavior ──────────────────────────────────────────────────────
 
 describe('Pool - close()', () => {
-  it('should not drop in-flight task result', async () => {
-    // Use isolate: false so the warm-up run establishes a reusable worker.
-    // The slow task then dispatches instantly on the existing process,
-    // eliminating the fork+handshake race that makes timer-based
-    // synchronization unreliable on slow CI.
-    const pool = new Pool(createPoolOptions({ isolate: false, minWorkers: 1 }));
-    // Warm up: ensure the worker is started and idle.
-    await pool.runTest(createTask());
-
-    const taskPromise = pool.runTest(
-      createTask('run', { __testMode: 'slow', __delayMs: 500 }),
-    );
-    // The reused worker receives the run request on an already-open IPC
-    // channel, so a short delay is enough for the message to arrive.
-    await new Promise((r) => setTimeout(r, 50));
-    const closePromise = pool.close();
-    // The task should still resolve (worker sends result before stopping).
-    const result = await taskPromise;
-    expect(result.status).toBe('pass');
-    await closePromise;
-  });
-
   it('should reject subsequent submissions after close()', async () => {
     const pool = new Pool(createPoolOptions());
     await pool.close();
     await expect(pool.runTest(createTask())).rejects.toThrow(/closed/);
   });
+
+  // Regression: rstest#1275. The host owns worker termination via SIGTERM
+  // and does not wait for any IPC ack. Previously the host waited the full
+  // 60s WORKER_STOP_TIMEOUT_MS plus a 5s SIGKILL escalation for workers
+  // that couldn't self-exit (e.g. rspack tokio threads ref'ing the loop),
+  // producing the 60–65s hang reported in the issue.
+  it('should close promptly under the forks pool (rstest#1275)', async () => {
+    const pool = new Pool(createPoolOptions({ isolate: false, minWorkers: 1 }));
+    await pool.runTest(createTask()); // warm up: worker is now alive and idle
+
+    const start = Date.now();
+    await pool.close();
+    const elapsed = Date.now() - start;
+
+    // Idle close should be SIGTERM-fast (sub-second). Allow generous CI
+    // headroom but require it to be well below the 60s graceful budget.
+    expect(elapsed).toBeLessThan(5000);
+  });
 });
 
 // ── maxWorkers capacity ─────────────────────────────────────────────────────