Fire messageerror when a posted message fails to deserialize

src/js/node/worker_threads.ts

@@ -101,6 +101,11 @@ type NodeWorkerOptions = import("node:worker_threads").WorkerOptions;
 // after their Worker exits
 let urlRevokeRegistry: FinalizationRegistry<string> | undefined = undefined;
 
+// The native `messageerror` MessageEvent carries no error object; node hands listeners an Error.
+function deserializeError() {
+  return new TypeError("Unable to deserialize data.");
+}
+
 function injectFakeEmitter(Class) {
   // Per-instance registry mapping each event to (user listener -> wrapper), so
   // listenerCount/eventNames/removeAllListeners work over EventTarget's opaque
@@ -124,6 +129,10 @@ function injectFakeEmitter(Class) {
     return event.error;
   }
 
+  function messageErrorEventHandler(event: ErrorEvent | MessageEvent) {
+    return event instanceof MessageEvent ? deserializeError() : event.error;
+  }
+
   function customEventHandler(event) {
     return event.detail;
   }
@@ -136,11 +145,14 @@ function injectFakeEmitter(Class) {
 
   function functionForEventType(event, listener) {
     switch (event) {
-      case "error":
-      case "messageerror": {
+      case "error": {
         return wrapped(errorEventHandler, listener);
       }
 
+      case "messageerror": {
+        return wrapped(messageErrorEventHandler, listener);
+      }
+
       case "message": {
         return wrapped(messageEventHandler, listener);
       }
@@ -1394,8 +1406,7 @@ class Worker extends EventEmitter {
   }
 
   #onMessageError(event: MessageEvent) {
-    // TODO: is this right?
-    this.emit("messageerror", (event as any).error ?? event.data ?? event);
+    this.emit("messageerror", (event as any).error ?? deserializeError());
   }
 
   #onOpen() {

src/jsc/bindings/webcore/MessageEvent.cpp

@@ -94,8 +94,17 @@ auto MessageEvent::create(JSC::JSGlobalObject& globalObject, Ref<SerializedScrip
     bool didFail = false;
 
     auto deserialized = data->deserialize(globalObject, &globalObject, ports, SerializationErrorMode::NonThrowing, &didFail);
-    if (topExceptionScope.exception()) [[unlikely]]
+    if (topExceptionScope.exception()) [[unlikely]] {
+        // A termination exception is left pending for the caller; anything else is a
+        // deserialization failure and becomes a `messageerror` event.
+        if (!vm.hasPendingTerminationException()) {
+            topExceptionScope.clearException();
+            didFail = true;
+        }
         deserialized = jsUndefined();
+    }
+    if (didFail)
+        deserialized = jsNull();
 
     JSC::Strong<JSC::Unknown> strongData(vm, deserialized);
 

src/jsc/bindings/webcore/MessagePort.cpp

@@ -367,6 +367,10 @@ void MessagePort::dispatchOneMessage(ScriptExecutionContext& context, MessageWit
     }
 
     auto event = MessageEvent::create(*context.jsGlobalObject(), message.message.releaseNonNull(), {}, {}, {}, WTF::move(ports));
+    if (scope.exception()) [[unlikely]] {
+        RELEASE_ASSERT(vm->hasPendingTerminationException());
+        return;
+    }
     dispatchEvent(event.event);
 }
 

src/jsc/bindings/webcore/SerializedScriptValue.cpp

@@ -5080,7 +5080,7 @@ JSValue SerializedScriptValue::deserialize(JSGlobalObject& lexicalGlobalObject,
     // Rethrow is a bit simpler here since we don't deal with return codes.
     RETURN_IF_EXCEPTION(scope, {});
 
-    return result.first;
+    return result.first ? result.first : jsNull();
 }
 
 } // namespace WebCore

src/jsc/bindings/webcore/WorkerMessagingProxy.cpp

@@ -317,6 +317,8 @@ static bool drainInbox(WorkerMessagingProxy::MessageInbox& inbox, Zig::GlobalObj
             auto message = batch.takeFirst();
             auto ports = MessagePort::entanglePorts(context, WTF::move(message.transferredPorts));
             auto event = MessageEvent::create(globalObject, message.message.releaseNonNull(), nullptr, WTF::move(ports));
+            if (globalObject.vm().hasPendingTerminationException()) [[unlikely]]
+                return false;
             dispatch(event.event);
             if (globalObject.drainMicrotasks())
                 return false; // termination pending

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

@@ -1085,6 +1085,50 @@ test("onmessageerror alone does not ref the port", () => {
   port1.close();
 });
 
+describe.concurrent("a message that fails to deserialize emits 'messageerror'", () => {
+  // Serializes fine but fails to deserialize: the DataView's offset is only in bounds
+  // after a getter resized the buffer, and the buffer was serialized before that.
+  const undeserializable = `(() => { const ab = new ArrayBuffer(8, { maxByteLength: 65536 }); return { ab, get grow() { ab.resize(65536); return 1; }, get view() { return new DataView(ab, 4096, 16); } }; })()`;
+
+  async function run(script: string) {
+    await using proc = Bun.spawn({ cmd: [bunExe(), "-e", script], env: bunEnv, stdout: "pipe", stderr: "inherit" });
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+    return { stdout, exitCode };
+  }
+
+  test("on a MessagePort, and later messages still arrive", async () => {
+    const { stdout, exitCode } = await run(
+      `const { MessageChannel } = require("node:worker_threads");
+       const { port1, port2 } = new MessageChannel();
+       const seen = [];
+       const record = s => { seen.push(s); if (seen.length === 2) { console.log(seen.join(",")); port1.close(); port2.close(); } };
+       port2.on("messageerror", e => record("messageerror:" + e.constructor.name + ":" + e.message));
+       port2.on("message", m => record("message:" + m));
+       port1.postMessage(${undeserializable});
+       port1.postMessage("after");`,
+    );
+    expect(stdout).toBe("messageerror:TypeError:Unable to deserialize data.,message:after\n");
+    expect(exitCode).toBe(0);
+  });
+
+  test("on the Worker when parentPort posts it, and later messages still arrive", async () => {
+    const { stdout, exitCode } = await run(
+      `const { Worker } = require("node:worker_threads");
+       const w = new Worker(
+         'const { parentPort } = require("node:worker_threads"); parentPort.postMessage(${undeserializable}); parentPort.postMessage("after"); setInterval(() => {}, 1000);',
+         { eval: true },
+       );
+       const seen = [];
+       const record = s => { seen.push(s); if (seen.length === 2) { console.log(seen.join(",")); w.terminate(); } };
+       w.on("error", e => { console.log("error", e); process.exit(1); });
+       w.on("messageerror", e => record("messageerror:" + e.constructor.name + ":" + e.message));
+       w.on("message", m => record("message:" + m));`,
+    );
+    expect(stdout).toBe("messageerror:TypeError:Unable to deserialize data.,message:after\n");
+    expect(exitCode).toBe(0);
+  });
+});
+
 // Collecting the unreferenced peer must not look like a peer close: node never
 // closes a channel because a port was garbage-collected, so ref() still works.
 test("hasRef() survives collection of the unreferenced peer", () => {

test/js/web/broadcastchannel/broadcast-channel.test.ts

@@ -1,5 +1,33 @@
+import { bunEnv, bunExe } from "harness";
 import util from "util";
 
+// A payload that serializes fine but fails to deserialize: the DataView's offset is only
+// in bounds after a getter resized the buffer, and the buffer was serialized before that.
+const undeserializable = `(() => { const ab = new ArrayBuffer(8, { maxByteLength: 65536 });
+  return { ab, get grow() { ab.resize(65536); return 1; }, get view() { return new DataView(ab, 4096, 16); } }; })()`;
+
+test("a message that fails to deserialize fires messageerror and later messages still arrive", async () => {
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `const a = new BroadcastChannel("undeserializable"), b = new BroadcastChannel("undeserializable");
+       const seen = [];
+       const record = s => { seen.push(s); if (seen.length === 2) { console.log(seen.join(",")); a.close(); b.close(); } };
+       b.onmessageerror = e => record("messageerror:" + e.data);
+       b.onmessage = e => record("message:" + e.data);
+       a.postMessage(${undeserializable});
+       a.postMessage("after");`,
+    ],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "inherit",
+  });
+  const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+  expect(stdout).toBe("messageerror:null,message:after\n");
+  expect(exitCode).toBe(0);
+});
+
 test("postMessage results in correct event", done => {
   let c1 = new BroadcastChannel("eventType");
   let c2 = new BroadcastChannel("eventType");

test/js/web/workers/message-channel.test.ts

@@ -1,3 +1,5 @@
+import { bunEnv, bunExe } from "harness";
+
 test("simple usage", done => {
   const channel = new MessageChannel();
   const port1 = channel.port1;
@@ -11,6 +13,31 @@ test("simple usage", done => {
   port1.postMessage("hello");
 });
 
+test("a message that fails to deserialize fires messageerror and later messages still arrive", async () => {
+  // Serializes fine but fails to deserialize: the DataView's offset is only in bounds
+  // after a getter resized the buffer, and the buffer was serialized before that.
+  await using proc = Bun.spawn({
+    cmd: [
+      bunExe(),
+      "-e",
+      `const { port1, port2 } = new MessageChannel();
+       const seen = [];
+       const record = s => { seen.push(s); if (seen.length === 2) { console.log(seen.join(",")); port1.close(); port2.close(); } };
+       port2.onmessageerror = e => record("messageerror:" + e.data);
+       port2.onmessage = e => record("message:" + e.data);
+       const ab = new ArrayBuffer(8, { maxByteLength: 65536 });
+       port1.postMessage({ ab, get grow() { ab.resize(65536); return 1; }, get view() { return new DataView(ab, 4096, 16); } });
+       port1.postMessage("after");`,
+    ],
+    env: bunEnv,
+    stdout: "pipe",
+    stderr: "inherit",
+  });
+  const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+  expect(stdout).toBe("messageerror:null,message:after\n");
+  expect(exitCode).toBe(0);
+});
+
 test("transfer message port", done => {
   const channel = new MessageChannel();
   const anotherChannel = new MessageChannel();

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

@@ -572,6 +572,34 @@ describe("web worker", () => {
       expect(exitCode).toBe(0);
     });
   });
+
+  test("a message that fails to deserialize fires messageerror in the worker and later messages still arrive", async () => {
+    // Serializes fine but fails to deserialize: the DataView's offset is only in bounds
+    // after a getter resized the buffer, and the buffer was serialized before that.
+    await using proc = Bun.spawn({
+      cmd: [
+        bunExe(),
+        "-e",
+        `const src = \`self.addEventListener("messageerror", e => postMessage("messageerror:" + e.data));
+                      self.onmessage = e => postMessage("message:" + e.data);\`;
+         const w = new Worker(URL.createObjectURL(new Blob([src])));
+         w.onerror = e => { console.log("error", e.message); process.exit(1); };
+         const seen = [];
+         w.onmessage = e => { seen.push(e.data); if (seen.length === 2) { console.log(seen.join(",")); w.terminate(); } };
+         w.addEventListener("open", () => {
+           const ab = new ArrayBuffer(8, { maxByteLength: 65536 });
+           w.postMessage({ ab, get grow() { ab.resize(65536); return 1; }, get view() { return new DataView(ab, 4096, 16); } });
+           w.postMessage("after");
+         });`,
+      ],
+      env: bunEnv,
+      stdout: "pipe",
+      stderr: "inherit",
+    });
+    const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]);
+    expect(stdout).toBe("messageerror:null,message:after\n");
+    expect(exitCode).toBe(0);
+  });
 });
 
 // TODO: move to node:worker_threads tests directory