diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index f9d221865a82..ee8c09f82a4d 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -101,6 +101,11 @@ type NodeWorkerOptions = import("node:worker_threads").WorkerOptions; // after their Worker exits let urlRevokeRegistry: FinalizationRegistry | 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() { diff --git a/src/jsc/bindings/webcore/MessageEvent.cpp b/src/jsc/bindings/webcore/MessageEvent.cpp index 381ba8d3ec4e..036a3061b074 100644 --- a/src/jsc/bindings/webcore/MessageEvent.cpp +++ b/src/jsc/bindings/webcore/MessageEvent.cpp @@ -94,8 +94,17 @@ auto MessageEvent::create(JSC::JSGlobalObject& globalObject, Refdeserialize(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 strongData(vm, deserialized); diff --git a/src/jsc/bindings/webcore/MessagePort.cpp b/src/jsc/bindings/webcore/MessagePort.cpp index 1e1681c17d85..39c216bb37e2 100644 --- a/src/jsc/bindings/webcore/MessagePort.cpp +++ b/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); } diff --git a/src/jsc/bindings/webcore/SerializedScriptValue.cpp b/src/jsc/bindings/webcore/SerializedScriptValue.cpp index a6ada2d00b66..af2a0522a34f 100644 --- a/src/jsc/bindings/webcore/SerializedScriptValue.cpp +++ b/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 diff --git a/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp b/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp index 22d8e944916c..0314897925bd 100644 --- a/src/jsc/bindings/webcore/WorkerMessagingProxy.cpp +++ b/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 diff --git a/test/js/node/worker_threads/worker_threads.test.ts b/test/js/node/worker_threads/worker_threads.test.ts index cc7acb16c723..4d5c11606325 100644 --- a/test/js/node/worker_threads/worker_threads.test.ts +++ b/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", () => { diff --git a/test/js/web/broadcastchannel/broadcast-channel.test.ts b/test/js/web/broadcastchannel/broadcast-channel.test.ts index fd8114dc58fb..d4841b128ea1 100644 --- a/test/js/web/broadcastchannel/broadcast-channel.test.ts +++ b/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"); diff --git a/test/js/web/workers/message-channel.test.ts b/test/js/web/workers/message-channel.test.ts index 09a4cb060c6d..014809699c56 100644 --- a/test/js/web/workers/message-channel.test.ts +++ b/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(); diff --git a/test/js/web/workers/worker.test.ts b/test/js/web/workers/worker.test.ts index 92d1011a59b0..aaf92ef48621 100644 --- a/test/js/web/workers/worker.test.ts +++ b/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