worker: discard queued messages on termination

Close the outside port and remove its message forwarding listeners
synchronously. Closing the port alone is asynchronous and can leave
queued messages dispatching when terminate() runs inside a listener.

Allow the current event to finish while discarding subsequent messages,
including those drained when the backing thread exits.

Signed-off-by: Filip Skokan <panva.ip@gmail.com>
Assisted-by: Codex
PR-URL: #66354
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Aviv Keller <me@aviv.sh>

Author: panva

lib/internal/webworker.js

@@ -109,6 +109,7 @@ const kName = Symbol('kName');
 const kNavigator = Symbol('kNavigator');
 const kNavigatorBrand = Symbol('kNavigatorBrand');
 const kOutsidePort = Symbol('kOutsidePort');
+const kRemoveMessageListeners = Symbol('kRemoveMessageListeners');
 const kType = Symbol('kType');
 const kURL = Symbol('kURL');
 const kWorker = Symbol('kWorker');
@@ -133,15 +134,21 @@ function createErrorEvent(init) {
 // "If this throws an exception, catch it, fire an event named messageerror
 // at messageEventTarget, using MessageEvent, and then return."
 function forwardMessageEvents(port, target) {
-  port.on('message', (data) =>
+  const onMessage = (data) =>
     target.dispatchEvent(lazyMessageEvent('message', {
       data,
       ports: port[kCurrentlyReceivingPorts],
-    })));
-  port.on('messageerror', (data) =>
+    }));
+  const onMessageError = (data) =>
     target.dispatchEvent(lazyMessageEvent('messageerror', {
       data,
-    })));
+    }));
+  port.on('message', onMessage);
+  port.on('messageerror', onMessageError);
+  return () => {
+    port.removeListener('message', onMessage);
+    port.removeListener('messageerror', onMessageError);
+  };
 }
 
 // Web IDL [Replaceable]: the setter replaces the accessor with an own
@@ -846,7 +853,7 @@ class Worker extends EventTarget {
     });
 
     this[kOutsidePort] = this[kWorker][kPublicPort];
-    forwardMessageEvents(this[kOutsidePort], this);
+    this[kRemoveMessageListeners] = forwardMessageEvents(this[kOutsidePort], this);
 
     // "Set notHandled to the result of firing an event named error at
     // workerObject, using ErrorEvent, with the cancelable attribute
@@ -865,6 +872,10 @@ class Worker extends EventTarget {
     validateThisInternalField(this, kWorker, 'Worker');
     // "The terminate() method steps are to terminate a worker given this's
     // worker."
+    // Closing a port is asynchronous. Remove its forwarding listeners too,
+    // since terminate() can run while the backing thread drains queued messages.
+    this[kRemoveMessageListeners]?.();
+    this[kOutsidePort].close();
     this[kWorker]?.terminate();
   }
 

test/parallel/test-webworker-postmessage-lifecycle.js

@@ -28,3 +28,51 @@ function checkSerialization(worker) {
     thread.once('exit', common.mustCall(() => checkSerialization(worker)));
   }));
 }
+
+// Queue messages before terminating so the test does not depend on thread speed.
+{
+  const source = `
+    onmessage = ({ data }) => {
+      for (let i = 0; i < 3; i++) postMessage(i);
+      const state = new Int32Array(data);
+      Atomics.store(state, 0, 1);
+      Atomics.notify(state, 0);
+    };
+  `;
+  const worker = new Worker(`data:text/javascript,${encodeURIComponent(source)}`);
+  worker.onerror = common.mustNotCall('worker failed');
+  worker.onmessage = common.mustNotCall('message delivered after terminate()');
+  const state = new Int32Array(new SharedArrayBuffer(4));
+  worker.postMessage(state.buffer);
+  assert.notStrictEqual(Atomics.wait(state, 0, 0, common.platformTimeout(10000)), 'timed-out');
+  worker.terminate();
+  worker.terminate();
+  checkSerialization(worker);
+}
+
+// terminate() can run during message dispatch, including while the backing
+// thread is draining messages on exit. Finish this event but discard later ones.
+for (const closeAfterPosting of [false, true]) {
+  const source = `
+    onmessage = ({ data }) => {
+      for (let i = 0; i < 3; i++) postMessage(i);
+      const state = new Int32Array(data);
+      Atomics.store(state, 0, 1);
+      Atomics.notify(state, 0);
+      if (${closeAfterPosting}) close();
+    };
+  `;
+  const worker = new Worker(`data:text/javascript,${encodeURIComponent(source)}`);
+  worker.onerror = common.mustNotCall('worker failed');
+  worker.addEventListener('message', common.mustCall(({ data }) => {
+    assert.strictEqual(data, 0);
+    worker.terminate();
+    checkSerialization(worker);
+  }));
+  worker.addEventListener('message', common.mustCall(({ data }) => {
+    assert.strictEqual(data, 0);
+  }));
+  const state = new Int32Array(new SharedArrayBuffer(4));
+  worker.postMessage(state.buffer);
+  assert.notStrictEqual(Atomics.wait(state, 0, 0, common.platformTimeout(10000)), 'timed-out');
+}