fix(eslint-plugin-runner): await stdio pipe close before terminating workers

On windows-latest, terminating a worker thread while its stdio pipes are still live lets libuv's concurrent pipe teardown fault below the JS layer (a native abort). rstest's forked test child intercepts uncaughtException/unhandledRejection/process.exit, so this surfaced as "Worker exited unexpectedly" in the high-terminate-churn worker-pool-e2e suite.

terminateWorker already destroyed the pipes first, but destroy() only initiates the close; a terminate() right after still races the in-flight teardown. Await each pipe's 'close' (bounded by a 1s grace) so the two are serialized — by the time we terminate, the pipes are gone and there is no concurrent teardown left to fault.

Also route the tests' raw worker.terminate() calls through terminateWorker so they exercise the same safe teardown as the production timeout/crash/shutdown paths.

Author: fansenze

packages/eslint-plugin-runner/src/worker-pool.ts

@@ -170,7 +170,43 @@ const WORKER_FILE = (() => {
 const WORKER_EXIT_GRACE_MS = 5_000;
 
 /**
- * Terminate a worker, closing its piped stdout/stderr FIRST.
+ * Upper bound on how long `terminateWorker` waits for a piped stream to
+ * emit `'close'` after `destroy()` before giving up and terminating
+ * anyway. A normal close lands within a tick; this only guards a wedged
+ * worker whose stream never closes, so teardown can never stall.
+ */
+const PIPE_CLOSE_GRACE_MS = 1_000;
+
+/**
+ * Destroy one worker stdio pipe and resolve once it has ACTUALLY closed
+ * — or immediately if the stream is absent / already destroyed, and at
+ * the latest after `PIPE_CLOSE_GRACE_MS` so a stream that never emits
+ * `'close'` can't stall the caller.
+ */
+async function closePipe(stream: Worker['stdout']): Promise<void> {
+  if (!stream || stream.destroyed) return;
+  await new Promise<void>((resolveClosed) => {
+    const done = (): void => {
+      clearTimeout(timer);
+      stream.off('close', done);
+      resolveClosed();
+    };
+    const timer = setTimeout(done, PIPE_CLOSE_GRACE_MS);
+    stream.once('close', done);
+    try {
+      stream.destroy();
+    } catch {
+      // `destroy()` is not expected to throw synchronously, but guard so
+      // a faulty stream still settles (clearing the timer + listener)
+      // instead of leaking the promise, the timer, and the listener.
+      done();
+    }
+  });
+}
+
+/**
+ * Terminate a worker, closing its piped stdout/stderr FIRST and WAITING
+ * for them to actually close before killing the thread.
  *
  * Mitigates a windows-latest-only crash: `worker.terminate()` abruptly
  * kills the thread while its stdio named pipes are still live, and
@@ -179,16 +215,21 @@ const WORKER_EXIT_GRACE_MS = 5_000;
  * `uncaughtException` / `unhandledRejection` / `process.exit`, so the
  * only way that child can "exit unexpectedly" is a native fault). It
  * surfaced only in the high-terminate-churn `worker-pool-e2e` suite.
- * Closing our read end first makes the pipe teardown deterministic and
- * removes that race — and is sound teardown hygiene on every platform.
+ *
+ * `destroy()` alone is NOT enough: it only *initiates* the close, which
+ * completes on a later tick, so a `terminate()` called synchronously
+ * after it still races the in-flight teardown. Awaiting each pipe's
+ * `'close'` SERIALIZES the two — by the time we terminate, the pipes are
+ * already gone and there is no concurrent teardown left to fault.
+ * Bounded by `PIPE_CLOSE_GRACE_MS` so a wedged worker can't stall here.
+ *
  * Trailing worker output is intentionally dropped (the worker is being
  * killed); `destroy()` errors are swallowed by the `ignorePipeError`
  * listeners wired in `spawnWorker`. Returns `terminate()`'s promise so
  * callers can chain `.then` / `.catch` / `.finally` unchanged.
  */
-async function terminateWorker(worker: Worker): Promise<number> {
-  worker.stdout?.destroy();
-  worker.stderr?.destroy();
+export async function terminateWorker(worker: Worker): Promise<number> {
+  await Promise.all([closePipe(worker.stdout), closePipe(worker.stderr)]);
   return worker.terminate();
 }
 

packages/eslint-plugin-runner/tests/regression-coverage.test.ts

@@ -16,7 +16,7 @@ import path from 'node:path';
 import * as espree from 'espree';
 import { parseSync } from 'oxc-parser';
 
-import { WorkerPool } from '../src/worker-pool.js';
+import { WorkerPool, terminateWorker } from '../src/worker-pool.js';
 import { createSourceCode } from '../src/source-code/source-code.js';
 import { lintFile } from '../src/linter/ecma-language-plugin.js';
 import type { LoadedPlugins } from '../src/plugin/plugin-loader.js';
@@ -96,7 +96,7 @@ describe('WorkerPool respawn rejection drains pendingQueue', () => {
     //    Pre-fix: pendingQueue stays full forever.
     //    Post-fix: drainQueueIfAllSlotsDegraded() in the reject
     //              callback resolves siblings with pool_degraded.
-    await internals.workers[0].worker.terminate();
+    await terminateWorker(internals.workers[0].worker);
 
     // 5. Race sibling against a generous-but-bounded timeout. Pre-fix
     //    the race resolves with 'timeout'; post-fix sibling settles
@@ -1167,7 +1167,7 @@ describe('WorkerPool shutdown does not wait for already-exited workers', () => {
     const internals = pool as any;
     // Terminate worker[0]; its 'exit' event fires now. With retryCap=0
     // the slot is NOT respawned — it stays in this.workers as dead.
-    await internals.workers[0].worker.terminate();
+    await terminateWorker(internals.workers[0].worker);
     // Drain pendingQueue cascade.
     await new Promise((r) => setTimeout(r, 100));
 
@@ -1913,8 +1913,8 @@ describe('drainQueueIfAllSlotsDegraded does not fire during mid-respawn', () =>
 
     // Terminate both nearly simultaneously to trigger respawn race.
     await Promise.all([
-      internals.workers[0].worker.terminate(),
-      internals.workers[1].worker.terminate(),
+      terminateWorker(internals.workers[0].worker),
+      terminateWorker(internals.workers[1].worker),
     ]);
     void spawnCallNum;
 
@@ -2525,7 +2525,7 @@ describe('WorkerPool: worker crashing mid-init is correctly respawned', () => {
     const internals = pool as any;
     const originalSpawn = internals.spawnWorker.bind(internals);
     let firstSlot: {
-      worker: { terminate(): Promise<number> };
+      worker: import('node:worker_threads').Worker;
       id: number;
     } | null = null;
     let callIdx = 0;
@@ -2563,7 +2563,7 @@ describe('WorkerPool: worker crashing mid-init is correctly respawned', () => {
     // slot 0 in it. Pre-fix `this.workers` was empty until the final
     // assignment, so the respawn's findIndex returned -1 and the
     // replacement got terminated as an orphan.
-    await firstSlot!.worker.terminate();
+    await terminateWorker(firstSlot!.worker);
 
     // Give the respawn a moment to complete.
     await new Promise((r) => setTimeout(r, 500));
@@ -3006,7 +3006,7 @@ describe('WorkerPool: shutdown awaits in-flight respawn', () => {
     };
 
     // Crash the only worker → exit handler kicks off the (slow) respawn.
-    await internals.workers[0].worker.terminate();
+    await terminateWorker(internals.workers[0].worker);
     // Give the exit handler a tick to enter the respawn branch.
     await new Promise((r) => setTimeout(r, 50));
     expect(respawnStarted).toBe(true);

packages/eslint-plugin-runner/tests/worker-pool-e2e-queue.test.ts

@@ -1,6 +1,6 @@
 import { describe, test, expect } from '@rstest/core';
 
-import { WorkerPool } from '../src/worker-pool.js';
+import { WorkerPool, terminateWorker } from '../src/worker-pool.js';
 import type { LintTask } from '../src/worker-pool.js';
 
 import {
@@ -122,7 +122,7 @@ describe('WorkerPool end-to-end with a local fixture plugin', () => {
     // path is to terminate the worker (triggers a real 'exit') after
     // setting crashCount to the cap.
     internals.workers[0].crashCount = internals.opts.retryCap;
-    await internals.workers[0].worker.terminate();
+    await terminateWorker(internals.workers[0].worker);
 
     const result = await batchP;
     expect(result).toHaveLength(2);
@@ -293,7 +293,7 @@ describe('WorkerPool end-to-end with a local fixture plugin', () => {
     const firstBatch = pool.lintBatch([task('first.ts', 'const x = null;\n')]);
     await new Promise((r) => setTimeout(r, 30));
     internals.workers[0].crashCount = internals.opts.retryCap;
-    await internals.workers[0].worker.terminate();
+    await terminateWorker(internals.workers[0].worker);
 
     const firstResult = await firstBatch;
     expect(firstResult).toHaveLength(1);

packages/eslint-plugin-runner/tests/worker-pool-e2e-resilience.test.ts

@@ -1,7 +1,8 @@
 import { describe, test, expect } from '@rstest/core';
 import path from 'node:path';
+import type { Worker } from 'node:worker_threads';
 
-import { WorkerPool } from '../src/worker-pool.js';
+import { WorkerPool, terminateWorker } from '../src/worker-pool.js';
 import type { LintTask } from '../src/worker-pool.js';
 
 import {
@@ -124,11 +125,9 @@ describe('WorkerPool end-to-end with a local fixture plugin', () => {
     await pool.init();
 
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
-    const workers = (pool as any).workers as Array<{
-      worker: { terminate(): Promise<number> };
-    }>;
+    const workers = (pool as any).workers as Array<{ worker: Worker }>;
     expect(workers.length).toBeGreaterThan(0);
-    void workers[0].worker.terminate();
+    void terminateWorker(workers[0].worker);
 
     const shutdownStart = Date.now();
     await pool.shutdown();