fix(compression): terminate idle workers instead of dropping them

changelog.d/fixes/12542-compression-terminate-idle-workers.md

@@ -0,0 +1 @@
+- **fix(compression):** Terminate idle compression workers instead of only dropping them from the pool, so each idle cycle no longer leaks a live thread and V8 isolate while `dispatch()` spawns a replacement (#12542 — thanks @pacocartones)

open-sse/services/compression/compressionWorkerPool.ts

@@ -131,7 +131,7 @@ export class CompressionWorkerPool {
   }
   async close(): Promise<void> {
     for (const job of this.queue.splice(0)) job.resolve(unchanged(job.originalBody));
-    await Promise.all([...this.workers].map((slot) => this.remove(slot, true)));
+    await Promise.all([...this.workers].map((slot) => this.remove(slot)));
   }
   private spawn(): PoolWorker {
     const slot: PoolWorker = {
@@ -185,21 +185,26 @@ export class CompressionWorkerPool {
     slot.timeout = null;
     slot.job = null;
     job.resolve(result);
-    slot.idle = setTimeout(() => void this.remove(slot, false), this.idleMs);
+    slot.idle = setTimeout(() => void this.remove(slot), this.idleMs);
     slot.idle.unref();
     this.dispatch();
   }
   private fail(slot: PoolWorker): void {
     const job = slot.job;
     if (job) job.resolve(unchanged(job.originalBody));
     slot.job = null;
-    void this.remove(slot, true).finally(() => this.dispatch());
+    void this.remove(slot).finally(() => this.dispatch());
   }
-  private async remove(slot: PoolWorker, terminate: boolean): Promise<void> {
+  /**
+   * Drop the slot from the pool AND terminate its thread. Dropping alone is not enough:
+   * the worker keeps a `parentPort` listener, so its event loop (and V8 isolate) stays
+   * alive forever while `dispatch()` respawns a replacement (#11804).
+   */
+  private async remove(slot: PoolWorker): Promise<void> {
     if (!this.workers.delete(slot)) return;
     if (slot.timeout) clearTimeout(slot.timeout);
     if (slot.idle) clearTimeout(slot.idle);
-    if (terminate) await slot.worker.terminate().catch(() => undefined);
+    await slot.worker.terminate().catch(() => undefined);
   }
 }
 

tests/unit/compression/compression-worker.test.ts

@@ -1,5 +1,6 @@
 import assert from "node:assert/strict";
 import { after, describe, it } from "node:test";
+import { Worker } from "node:worker_threads";
 import {
   isCompressionWorkerEligible,
   isStrictlySerializable,
@@ -136,6 +137,40 @@ describe("compression worker execution", () => {
     }
   });
 
+  it("terminates an idle worker instead of only dropping it from the pool", async () => {
+    const spawned = new Set<Worker>();
+    const terminated: Promise<number>[] = [];
+    const originalPostMessage = Worker.prototype.postMessage;
+    const originalTerminate = Worker.prototype.terminate;
+    Worker.prototype.postMessage = function (this: Worker, ...args) {
+      spawned.add(this);
+      return originalPostMessage.apply(this, args);
+    };
+    Worker.prototype.terminate = function (this: Worker) {
+      const exit = originalTerminate.call(this);
+      terminated.push(exit);
+      return exit;
+    };
+    const messagePorts = () =>
+      process.getActiveResourcesInfo().filter((resource) => resource === "MessagePort").length;
+    const portsBefore = messagePorts();
+    const pool = new CompressionWorkerPool({ size: 1, idleMs: 50 });
+    try {
+      await pool.run(body, "stacked", { config });
+      await new Promise((resolve) => setTimeout(resolve, 300));
+      assert.equal(spawned.size, 1);
+      assert.equal(terminated.length, 1, "idle eviction must terminate the worker thread");
+      await Promise.all(terminated);
+      assert.ok(messagePorts() <= portsBefore, "idle eviction must not retain the worker's port");
+    } finally {
+      Worker.prototype.postMessage = originalPostMessage;
+      Worker.prototype.terminate = originalTerminate;
+      await pool.close();
+      // Reap anything the pool forgot so a regression fails instead of hanging the runner.
+      await Promise.all([...spawned].map((worker) => worker.terminate().catch(() => undefined)));
+    }
+  });
+
   it("keeps the parent event loop responsive while two workers overlap", async () => {
     const largeBody = {
       messages: Array.from({ length: 400 }, (_, index) => ({