Close a MessagePort transferred to a worker that never starts

src/jsc/bindings/webcore/WorkerMessagingProxy.cpp

@@ -318,8 +318,7 @@ static bool drainInbox(WorkerMessagingProxy::MessageInbox& inbox, Zig::GlobalObj
             remaining -= batch.size();
 
         while (!batch.isEmpty()) {
-            // The receiving VM is being stopped: nothing more is delivered (the
-            // rest is dropped with the proxy).
+            // The receiving VM is being stopped: nothing more is delivered, and what is left is dropped unread.
             if (context.isJSExecutionForbidden())
                 return false;
             auto message = batch.takeFirst();
@@ -540,6 +539,19 @@ void WorkerMessagingProxy::releaseWorkerThread()
     deref();
 }
 
+void WorkerMessagingProxy::dropUndeliveredWorkerMessages()
+{
+    // Ports a worker that never started did not take. Closing its parentPort end also closes the ports queued on it.
+    auto droppedDataPorts = std::exchange(m_options.dataMessagePorts, {});
+    Deque<MessageWithMessagePorts> droppedMessages;
+    {
+        Locker locker { m_toWorker.lock };
+        droppedMessages = std::exchange(m_toWorker.queue, {});
+        m_toWorker.drainScheduled = false;
+    }
+    // Destroyed here, outside the lock: ~TransferredMessagePort closes its pipe side and notifies the peer.
+}
+
 void WorkerMessagingProxy::workerGlobalScopeDestroyedInternal(int32_t exitCode, bool stoppedByParent)
 {
     ASSERT(m_scriptExecutionContext && m_scriptExecutionContext->isContextThread());
@@ -558,6 +570,7 @@ void WorkerMessagingProxy::workerGlobalScopeDestroyedInternal(int32_t exitCode,
         m_state.store(State::Closing);
         m_pendingTasks.clear();
     }
+    dropUndeliveredWorkerMessages();
     rejectAllCrossVMRequests();
 
     // Everything the worker posted before it exited is delivered before 'close' (Node: before
@@ -592,6 +605,8 @@ void WorkerMessagingProxy::parentContextWillDestroy()
         m_pendingCrossVMRequests.clear();
     }
     releaseWorkerThread();
+    // After the join: a live worker thread can still be taking its ports (createNodeWorkerThreadsBinding).
+    dropUndeliveredWorkerMessages();
     m_scriptExecutionContext = nullptr;
 }
 

src/jsc/bindings/webcore/WorkerMessagingProxy.h

@@ -124,6 +124,7 @@ class WorkerMessagingProxy final : public ThreadSafeRefCounted<WorkerMessagingPr
 
     void workerGlobalScopeDestroyedInternal(int32_t exitCode, bool stoppedByParent);
     void releaseWorkerThread();
+    void dropUndeliveredWorkerMessages();
     void drainMessagesToWorkerObject(ScriptExecutionContext&, DrainBudget);
     void rejectAllCrossVMRequests();
     void postMessageErrorToWorkerObject(String&& message);

test/js/node/worker_threads/worker_threads.test.ts

@@ -2221,6 +2221,66 @@ describe("'online' precedes the worker's first message", () => {
     expect(order).toEqual(["online", "error:MODULE_NOT_FOUND"]);
     expect(code).toBe(1);
   });
+
+  test.each(["workerData", "postMessage"] as const)(
+    "a %s transferred MessagePort closes when the worker entry does not resolve",
+    async transferRoute => {
+      using dir = tempDir("worker-missing-entry-transferred-port", {});
+      const { port1, port2 } = new MessageChannel();
+      const worker = new Worker(
+        join(String(dir), "missing.js"),
+        transferRoute === "workerData" ? { workerData: { port: port2 }, transferList: [port2] } : undefined,
+      );
+      const events: string[] = [];
+      worker.on("online", () => events.push("online"));
+      worker.on("error", error => events.push(`error:${error.code}`));
+      const portClosed = once(port1, "close").then(() => events.push("port-close"));
+      const exit = new Promise<number>(resolve =>
+        worker.on("exit", code => {
+          events.push(`exit:${code}`);
+          resolve(code);
+        }),
+      );
+      if (transferRoute === "postMessage") {
+        worker.postMessage({ port: port2 }, [port2]);
+      }
+
+      const code = await exit;
+      await portClosed;
+
+      expect({ code, events }).toEqual({
+        code: 1,
+        events: ["online", "error:MODULE_NOT_FOUND", "exit:1", "port-close"],
+      });
+    },
+  );
+});
+
+// terminate() in the same tick as the constructor stops the thread before it takes its
+// workerData ports or reads parentPort. The thread takes them before any user code runs, so
+// nothing can hold it back, and a thread that wins that race closes them as it exits. So only
+// the close is asserted, not its order against 'exit'. The tests above, where the entry does
+// not resolve, are the ones that cannot pass without the drop.
+describe("a transferred MessagePort closes when terminate() stops the worker before it starts", () => {
+  // Stays referenced, as a Worker in a pool does: a collected Worker drops its ports too.
+  let worker: Worker;
+
+  test.each(["workerData", "postMessage"] as const)("transferred through %s", async transferRoute => {
+    const { port1, port2 } = new MessageChannel();
+    worker = new Worker(
+      "setInterval(() => {}, 1000)",
+      transferRoute === "workerData"
+        ? { eval: true, workerData: { port: port2 }, transferList: [port2] }
+        : { eval: true },
+    );
+    const portClosed = once(port1, "close").then(() => "port-close");
+    if (transferRoute === "postMessage") {
+      worker.postMessage({ port: port2 }, [port2]);
+    }
+
+    await worker.terminate();
+    expect(await portClosed).toBe("port-close");
+  });
 });
 
 // ─── worker teardown vs. work still in flight ────────────────────────────────

test/js/web/workers/worker.test.ts

@@ -403,6 +403,42 @@ describe("web worker", () => {
     });
   });
 
+  // A message the worker never took from its inbox is dropped when the worker is gone. A port in
+  // that message is closed with it, so the port's peer hears 'close'.
+  describe("a MessagePort posted to a worker that never reads it closes", () => {
+    // Stays referenced, as a Worker in a pool does: a collected Worker drops its inbox too.
+    let worker: Worker;
+
+    test("the entry point does not resolve", async () => {
+      using dir = tempDir("web-worker-missing-entry-port", {});
+      const { port1, port2 } = new MessageChannel();
+      worker = new Worker(path.join(String(dir), "missing.js"));
+      const events: string[] = [];
+      worker.addEventListener("error", () => events.push("error"));
+      worker.addEventListener("close", e => events.push(`close:${e.code}`));
+      const portClosed = once(port1, "close").then(() => events.push("port-close"));
+      worker.postMessage({ port: port2 }, [port2]);
+
+      await portClosed;
+      expect(events).toEqual(["error", "close:1", "port-close"]);
+    });
+
+    // The entry never returns, so the worker never reads its inbox, whether terminate() lands
+    // before the thread starts or while the entry runs.
+    test("terminate() stops a worker whose entry is still running", async () => {
+      const { port1, port2 } = new MessageChannel();
+      worker = new Worker("data:text/javascript,for(;;){}");
+      const events: string[] = [];
+      worker.addEventListener("close", () => events.push("close"));
+      const portClosed = once(port1, "close").then(() => events.push("port-close"));
+      worker.postMessage({ port: port2 }, [port2]);
+      worker.terminate();
+
+      await portClosed;
+      expect(events).toEqual(["close", "port-close"]);
+    });
+  });
+
   // As in browsers (and Node's Web Worker), the worker's implicit port opens once the entry's
   // synchronous part has run: a message dispatched while no 'message' handler exists is dropped.
   // node:worker_threads' parentPort is what queues until a listener is attached. #40141