fix(worker): route parentPort listener throws and handled Node-style errors like the web surface

A parentPort listener that threw was dispatched without rethrowing, so the error went to the uncaught-error reporter and never reached the worker's onerror or its parent. The relay now dispatches the way worker-global message delivery does.

The worker_threads Worker's onerror handler returned nothing, so an error its 'error' listeners took was also reported to the parent's global scope as unhandled. It returns whether a listener ran, which cancels the event.

Author: adrian-niculescu

test-app/app/src/main/assets/app/tests/messaging/parentPortThrowingWorker.js

@@ -0,0 +1,4 @@
+var parentPort = require("node:worker_threads").parentPort;
+parentPort.on("message", function () {
+    throw new Error("thrown by a parentPort listener");
+});

test-app/app/src/main/assets/app/tests/testMessaging.js

@@ -255,6 +255,48 @@ describe("Messaging runtime edges", function () {
                 }
             };
         });
+
+        it("lets a node:worker_threads 'error' listener consume the error", function (done) {
+            var wt = require("node:worker_threads");
+            var globalErrors = [];
+            var listener = function (event) {
+                globalErrors.push(event.message);
+                event.preventDefault();
+            };
+            addEventListener("error", listener);
+            var worker = new wt.Worker("~/tests/messaging/throwingWorker.js");
+            worker.on("error", function (error) {
+                setTimeout(function () {
+                    removeEventListener("error", listener);
+                    expect(error.message).toContain("boom from worker");
+                    expect(globalErrors).toEqual([]);
+                    worker.terminate();
+                    done();
+                }, SETTLE);
+            });
+        });
+
+        it("routes a throw from a parentPort listener to the parent's 'error' listeners", function (done) {
+            var wt = require("node:worker_threads");
+            var worker = new wt.Worker("~/tests/messaging/parentPortThrowingWorker.js");
+            var messages = [];
+            var finish = function () {
+                expect(messages.length).toBe(1);
+                expect(messages[0]).toContain("thrown by a parentPort listener");
+                worker.terminate();
+                done();
+            };
+            // Nothing else settles the spec when the error never arrives.
+            var guard = setTimeout(finish, 10000);
+            worker.on("error", function (error) {
+                messages.push(error.message);
+                if (messages.length === 1) {
+                    clearTimeout(guard);
+                    setTimeout(finish, SETTLE);
+                }
+            });
+            worker.postMessage("go");
+        });
     });
 
     describe("AbortSignal handler attribute accounting", function () {

test-app/runtime/src/main/cpp/js/node-worker-threads.js

@@ -44,6 +44,7 @@ const { BroadcastChannel } = require("internal/broadcast-channel");
 const {
   EventTarget,
   defineEventHandler,
+  dispatchEventRethrowing,
   globalEventTarget,
 } = require("internal/events");
 
@@ -63,7 +64,6 @@ const globalPostMessage = g.postMessage;
 
 const addEventListener = EventTarget.prototype.addEventListener;
 const removeEventListener = EventTarget.prototype.removeEventListener;
-const dispatchEvent = EventTarget.prototype.dispatchEvent;
 
 // Runs `fn` after the caller returns. Node reports 'online' and 'exit' from
 // the thread's own lifecycle; the runtime's Worker has no equivalent signal,
@@ -123,10 +123,11 @@ class WorkerEmitter {
     return this.removeListener(type, listener);
   }
 
+  // Whether a listener was registered, as Node's EventEmitter reports it.
   emit(type, arg) {
     const list = this.#listeners[type];
-    if (list === undefined) {
-      return;
+    if (list === undefined || list.length === 0) {
+      return false;
     }
     const snapshot = ArrayPrototypeSlice(list);
     for (let i = 0; i < snapshot.length; i++) {
@@ -139,6 +140,7 @@ class WorkerEmitter {
       }
       FunctionPrototypeCall(entry.listener, this, arg);
     }
+    return true;
   }
 }
 
@@ -179,8 +181,10 @@ class Worker extends WorkerEmitter {
     worker.onmessageerror = function (event) {
       self.emit("messageerror", event.data);
     };
+    // A truthy return cancels the error, so one an 'error' listener took is
+    // not reported to the parent's global scope as well.
     worker.onerror = function (error) {
-      self.emit("error", error);
+      return self.emit("error", error);
     };
     soon(function () {
       self.emit("online", undefined);
@@ -312,9 +316,11 @@ ObjectDefineProperty(ParentPort.prototype, SymbolToStringTag, {
 let parentPort = null;
 if (!isMainThread) {
   parentPort = new ParentPort();
+  // Rethrowing, so a listener that throws reaches the worker's error chain
+  // (the scope's onerror, then the parent's Worker) the way a throwing
+  // onmessage on the global scope does.
   const relay = function (event) {
-    FunctionPrototypeCall(
-      dispatchEvent,
+    dispatchEventRethrowing(
       parentPort,
       new (getMessageEvent())(event.type, { data: event.data, ports: event.ports })
     );