worker_threads: bind this to the port in MessagePort on()/once() listeners

src/js/node/worker_threads.ts

@@ -128,25 +128,26 @@ function injectFakeEmitter(Class) {
     return event.detail;
   }
 
-  function wrapped(run, listener) {
+  // node-style listeners run with `this` = the object on()/once() was called on.
+  function wrapped(target, run, listener) {
     return function (event) {
-      return listener(run(event));
+      return listener.$call(target, run(event));
     };
   }
 
-  function functionForEventType(event, listener) {
+  function functionForEventType(target, event, listener) {
     switch (event) {
       case "error":
       case "messageerror": {
-        return wrapped(errorEventHandler, listener);
+        return wrapped(target, errorEventHandler, listener);
       }
 
       case "message": {
-        return wrapped(messageEventHandler, listener);
+        return wrapped(target, messageEventHandler, listener);
       }
 
       default: {
-        return wrapped(customEventHandler, listener);
+        return wrapped(target, customEventHandler, listener);
       }
     }
   }
@@ -173,7 +174,7 @@ function injectFakeEmitter(Class) {
   }
 
   function on(event, listener) {
-    register(this, event, listener, functionForEventType(event, listener), undefined);
+    register(this, event, listener, functionForEventType(this, event, listener), undefined);
     return this;
   }
 
@@ -190,8 +191,8 @@ function injectFakeEmitter(Class) {
   }
 
   function once(event, listener) {
-    const wrapper = functionForEventType(event, listener);
     const target = this;
+    const wrapper = functionForEventType(target, event, listener);
     // EventTarget drops a {once:true} listener natively, without telling the
     // registry — so purge it here or listenerCount()/eventNames() keep counting
     // a listener that already fired.

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

@@ -1418,6 +1418,132 @@ test("off() removes a pending once() listener", () => {
   port2.close();
 });
 
+// node invokes node-style listeners with `this` bound to the port. The on()/once()
+// wrappers used to call the user's function bare, so `this` was undefined. Covers
+// every way a wrapper gets invoked: a native MessageEvent, a native close Event
+// (on, once and close(cb), which is once("close", cb)), and emit(). The last
+// listener registered for each event is the one that resolves, so every earlier
+// one has already run by the time the promise settles.
+test("on()/once() listeners are called with `this` bound to the port", async () => {
+  const { port1, port2 } = new MessageChannel();
+  const calls: unknown[] = [];
+  const { promise: messaged, resolve: onMessaged } = Promise.withResolvers<void>();
+  const { promise: closed, resolve: onClosed } = Promise.withResolvers<void>();
+
+  port1.on("message", function (this: unknown, data) {
+    calls.push(["on message", this === port1, data]);
+  });
+  port1.once("message", function (this: unknown, data) {
+    calls.push(["once message", this === port1, data]);
+    onMessaged();
+  });
+  port1.on("close", function (this: unknown) {
+    calls.push(["on close", this === port1]);
+  });
+  port1.once("close", function (this: unknown) {
+    calls.push(["once close", this === port1]);
+  });
+  port1.on("custom", function (this: unknown, arg) {
+    calls.push(["on custom", this === port1, arg]);
+  });
+  port1.once("custom", function (this: unknown, arg) {
+    calls.push(["once custom", this === port1, arg]);
+  });
+
+  port1.emit("custom", 42);
+  port2.postMessage("hi");
+  await messaged;
+  port1.close(function (this: unknown) {
+    calls.push(["close(cb)", this === port1]);
+    onClosed();
+  });
+  await closed;
+  port2.close();
+
+  expect(calls).toEqual([
+    ["on custom", true, 42],
+    ["once custom", true, 42],
+    ["on message", true, "hi"],
+    ["once message", true, "hi"],
+    ["on close", true],
+    ["once close", true],
+    ["close(cb)", true],
+  ]);
+});
+
+test("parentPort.on() listeners are called with `this` bound to parentPort", async () => {
+  const w = new Worker(
+    `const { parentPort } = require("node:worker_threads");
+     parentPort.once("message", function (data) {
+       this.postMessage({ data, thisIsParentPort: this === parentPort });
+     });`,
+    { eval: true },
+  );
+  w.postMessage("ping");
+  const [reply] = await once(w, "message");
+  await w.terminate();
+  expect(reply).toEqual({ data: "ping", thisIsParentPort: true });
+});
+
+// In a plain globalThis.Worker, parentPort is a stand-in object whose
+// addEventListener forwards to the worker's global scope, so EventTarget invokes
+// the wrapper with the global scope as `this`. The listener must still see the
+// object it was registered on.
+test("the stand-in parentPort of a non-node Worker also binds `this` to parentPort", async () => {
+  const url = URL.createObjectURL(
+    new Blob([
+      `const { parentPort } = require("worker_threads");
+       parentPort.on("message", function (data) {
+         self.postMessage({ data, thisIsParentPort: this === parentPort, thisIsGlobal: this === self });
+       });`,
+    ]),
+  );
+  const w = new globalThis.Worker(url);
+  try {
+    const { promise, resolve, reject } = Promise.withResolvers<unknown>();
+    w.onmessage = e => resolve(e.data);
+    w.onerror = e => reject(new Error(e.message));
+    w.postMessage("ping");
+    expect(await promise).toEqual({ data: "ping", thisIsParentPort: true, thisIsGlobal: false });
+  } finally {
+    w.terminate();
+    URL.revokeObjectURL(url);
+  }
+});
+
+// The wrappers hand the listener's return value back to EventTarget, which is
+// how a rejecting async on()/once() listener becomes an uncaughtException, as in
+// node. Spawned because the test runner would otherwise see the exceptions itself.
+test("a rejecting async on()/once() listener is reported as an uncaughtException", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `const { MessageChannel } = require("worker_threads");
+       const { port1, port2 } = new MessageChannel();
+       const seen = [];
+       process.on("uncaughtException", err => {
+         seen.push(err.message);
+         if (seen.length === 2) {
+           console.log(JSON.stringify(seen));
+           port1.close();
+         }
+       });
+       port1.on("message", async () => { throw new Error("from on"); });
+       port1.once("message", async () => { throw new Error("from once"); });
+       port2.postMessage(1);`,
+    ],
+    env: bunEnv,
+    stderr: "pipe",
+  });
+  const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]);
+  expect({ stdout: stdout.trim(), stderr, exitCode }).toEqual({
+    stdout: JSON.stringify(["from on", "from once"]),
+    stderr: "",
+    exitCode: 0,
+  });
+});
+
 test("close(cb) interleaves with other close listeners in registration order", async () => {
   // node's mechanism is `this.once('close', cb)`, so cb interleaves with other
   // close listeners in the order they were registered (verified against node).