From 5855a14de4c5f942c6d0f491ac70f6ee1f5b4b7d Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 21:28:06 +0000 Subject: [PATCH 01/13] MessagePort: implement NodeEventTarget on the native listener list MessagePort's EventEmitter surface (.on/.once/.off/.emit/listenerCount/ eventNames/removeAllListeners/set|getMaxListeners) was a JS shim grafted onto the prototype when node:worker_threads first loaded. It kept its own (event -> userFn -> wrapper) registry alongside WebCore's EventTarget listener list, so: - the surface was absent until worker_threads was imported somewhere - .on(t, f) registered a wrapper via addEventListener, so the same function added via both APIs ran twice, listenerCount/eventNames didn't see addEventListener listeners, removeEventListener(f) couldn't remove a .on(f), removeAllListeners left addEventListener listeners live, and getEventListeners returned the wrapper not f - emit('message', payload) built MessageEvent(event, payload) with the payload in the init-dict slot, so listeners got null, a primitive payload threw, and emit returned this instead of a boolean Node's model (lib/internal/event_target.js): MessagePort IS a NodeEventTarget. .on/.addListener is addEventListener with kIsNodeStyleListener storing f itself in the one listener list; once is the same + once:true; listenerCount/eventNames/removeAllListeners read or clear that map; emit(type, arg) = had = listenerCount > 0, hybrid dispatch (node-style listeners get arg by identity, EventTarget-style get a lazily-built MessageEvent/CustomEvent), returns had. This change: - adds an isNodeStyleListener bit to RegisteredEventListener / AddEventListenerOptions, keyed off a private kIsNodeStyleListener symbol so only internal code can set it - in innerInvokeEventListeners, flagged listeners are invoked via JSEventListener::handleEventNodeStyle, which passes event.data / .detail / .error instead of the Event wrapper - puts the NodeEventTarget methods natively on an intermediate prototype between MessagePort.prototype and EventTarget.prototype (so Object.getOwnPropertyNames(MessagePort.prototype) matches node) and implements them over the one native listener list, including removeAllEventListenersForType - emit() builds a MessageEvent{data:arg} or CustomEvent{detail:arg} and dispatches it; node-style listeners recover arg by identity - wires CustomEventInterfaceType into EventFactory so a natively-created CustomEvent is wrapped as JSCustomEvent (its .detail was unreachable) - deletes the ~180-line injectFakeEmitter shim; parentPort's stand-in forwards on/once/off to self.addEventListener with the private flag BroadcastChannel is a plain EventTarget in node too; it keeps no shim (unchanged). --- src/js/builtins.d.ts | 6 + src/js/builtins/BunBuiltinNames.h | 1 + src/js/node/worker_threads.ts | 220 ++-------- .../webcore/AddEventListenerOptions.h | 5 + src/jsc/bindings/webcore/EventFactory.cpp | 4 +- src/jsc/bindings/webcore/EventHeaders.h | 4 +- src/jsc/bindings/webcore/EventListener.h | 6 + src/jsc/bindings/webcore/EventListenerMap.cpp | 17 + src/jsc/bindings/webcore/EventListenerMap.h | 1 + src/jsc/bindings/webcore/EventTarget.cpp | 19 +- src/jsc/bindings/webcore/EventTarget.h | 1 + .../webcore/JSAddEventListenerOptions.cpp | 6 + src/jsc/bindings/webcore/JSEventListener.cpp | 43 +- src/jsc/bindings/webcore/JSEventListener.h | 6 + src/jsc/bindings/webcore/JSMessagePort.cpp | 220 +++++++++- src/jsc/bindings/webcore/MessagePort.h | 4 + .../webcore/RegisteredEventListener.h | 7 +- .../message-port-node-event-target.test.ts | 404 ++++++++++++++++++ 18 files changed, 773 insertions(+), 201 deletions(-) create mode 100644 test/js/node/worker_threads/message-port-node-event-target.test.ts diff --git a/src/js/builtins.d.ts b/src/js/builtins.d.ts index 7cc91f81fc46..eacfcb73cc8a 100644 --- a/src/js/builtins.d.ts +++ b/src/js/builtins.d.ts @@ -475,6 +475,12 @@ declare interface AddEventListenerOptions { * internal `kResistStopPropagation`. */ $kResistStopPropagation?: boolean; + /** + * Private symbol read by the native EventTarget. A listener registered with it is + * invoked with the event's underlying value (data / detail / error) instead of the + * Event wrapper. Mirrors Node.js's internal `kIsNodeStyleListener`. + */ + $kIsNodeStyleListener?: boolean; } declare class OutOfMemoryError { diff --git a/src/js/builtins/BunBuiltinNames.h b/src/js/builtins/BunBuiltinNames.h index b6495c2eea2b..cf86a6694a6b 100644 --- a/src/js/builtins/BunBuiltinNames.h +++ b/src/js/builtins/BunBuiltinNames.h @@ -119,6 +119,7 @@ using namespace JSC; macro(isUntransferable) \ macro(join) \ macro(json) \ + macro(kIsNodeStyleListener) \ macro(kResistStopPropagation) \ macro(key) \ macro(lazy) \ diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 921df1a621d2..081dd4f10f53 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -4,7 +4,6 @@ declare const self: typeof globalThis; type WebWorker = InstanceType; const EventEmitter = require("node:events"); -const { SafeMap } = require("internal/primordials"); const Readable = require("internal/streams/readable"); const Writable = require("internal/streams/writable"); const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared"); @@ -101,189 +100,12 @@ type NodeWorkerOptions = import("node:worker_threads").WorkerOptions; // after their Worker exits let urlRevokeRegistry: FinalizationRegistry | undefined = undefined; -function injectFakeEmitter(Class) { - // Per-instance registry mapping each event to (user listener -> wrapper), so - // listenerCount/eventNames/removeAllListeners work over EventTarget's opaque - // internal map and off() can find the wrapper a given listener registered. - // SafeMap: its prototype is a frozen, null-proto snapshot of Map.prototype, so - // .get/.set/.size/.values()/iteration all bypass a user-replaced Map.prototype. - // (It has no @get/@set private names, so the $-intrinsics don't apply to it.) - // Keyed by a module-local symbol, not a WeakMap — WeakMap has neither defence. - const kListenerRegistry = Symbol("listenerRegistry"); - function registryFor(target, create) { - let map = target[kListenerRegistry]; - if (!map && create) target[kListenerRegistry] = map = new SafeMap(); - return map; - } - - function messageEventHandler(event: MessageEvent) { - return event.data; - } - - function errorEventHandler(event: ErrorEvent) { - return event.error; - } - - function customEventHandler(event) { - return event.detail; - } - - function wrapped(run, listener) { - return function (event) { - return listener(run(event)); - }; - } - - function functionForEventType(event, listener) { - switch (event) { - case "error": - case "messageerror": { - return wrapped(errorEventHandler, listener); - } - - case "message": { - return wrapped(messageEventHandler, listener); - } - - default: { - return wrapped(customEventHandler, listener); - } - } - } - - function EventClass(eventName) { - if (eventName === "error" || eventName === "messageerror") { - return ErrorEvent; - } - - return MessageEvent; - } - - // EventTarget dedupes on (type, callback), so in node the FIRST registration of - // a listener wins outright -- including its once-ness -- and later adds of the - // same function are no-ops. Keying wrappers per listener reproduces that. - function register(target, event, listener, wrapper, options) { - const map = registryFor(target, true)!; - let byListener = map.get(event); - if (!byListener) map.set(event, (byListener = new SafeMap())); - if (byListener.has(listener)) return false; - target.addEventListener(event, wrapper, options); - byListener.set(listener, wrapper); - return true; - } - - function on(event, listener) { - register(this, event, listener, functionForEventType(event, listener), undefined); - return this; - } - - function off(event, listener) { - if (listener) { - const byListener = registryFor(this, false)?.get(event); - const wrapper = byListener?.get(listener) ?? listener; - this.removeEventListener(event, wrapper); - byListener?.delete(listener); - } else { - this.removeEventListener(event); - } - return this; - } - - function once(event, listener) { - const wrapper = functionForEventType(event, listener); - const target = this; - // 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. - function onceWrapper(ev) { - registryFor(target, false)?.get(event)?.delete(listener); - return wrapper(ev); - } - register(this, event, listener, onceWrapper, { once: true }); - return this; - } - - function emit(event, ...args) { - switch (event) { - case "error": - case "messageerror": - case "message": - this.dispatchEvent(new (EventClass(event))(event, ...args)); - break; - default: - // Non-standard events surface as CustomEvent (detail = first arg) to - // addEventListener and as the raw argument to .on(), matching node. - this.dispatchEvent(new CustomEvent(event, { detail: args[0] })); - break; - } - return this; - } - - const kMaxListeners = Symbol("kMaxListeners"); - function setMaxListeners(n) { - this[kMaxListeners] = n; - return this; - } - function getMaxListeners() { - return this[kMaxListeners] ?? 10; - } - function listenerCount(type) { - return registryFor(this, false)?.get(type)?.size ?? 0; - } - function eventNames() { - const map = registryFor(this, false); - if (!map) return []; - const out: string[] = []; - for (const [k, v] of map) if (v.size > 0) out.push(k); - return out; - } - function removeAllListeners(type) { - const map = registryFor(this, false); - if (!map) return this; - const removeType = t => { - const byListener = map.get(t); - if (byListener) { - for (const w of byListener.values()) this.removeEventListener(t, w); - map.delete(t); - } - }; - if (arguments.length === 0) { - // removeType only deletes `t`, and a Map iterator tolerates deleting the - // entry it just yielded — so no snapshot copy is needed here. - for (const t of map.keys()) removeType(t); - } else { - removeType(type); - } - return this; - } - - // node inherits these from NodeEventTarget.prototype (a curated subset of - // EventEmitter, not EventEmitter itself); use an intermediate prototype so - // Object.getOwnPropertyNames(MessagePort.prototype) matches node. - const proto = Class.prototype; - const inherited = Object.create(Object.getPrototypeOf(proto)); - const emitterMethods: [string, Function][] = [ - ["on", on], - ["off", off], - ["once", once], - ["emit", emit], - ["addListener", on], - ["removeListener", off], - ["listenerCount", listenerCount], - ["eventNames", eventNames], - ["removeAllListeners", removeAllListeners], - ["setMaxListeners", setMaxListeners], - ["getMaxListeners", getMaxListeners], - ]; - for (const [methodName, value] of emitterMethods) { - Object.defineProperty(inherited, methodName, { value, writable: true, enumerable: false, configurable: true }); - } - Object.setPrototypeOf(proto, inherited); -} - +// MessagePort is a NodeEventTarget: the node-style emitter surface (.on / +// .once / .off / .emit / listenerCount / eventNames / removeAllListeners / +// set|getMaxListeners) lives on MessagePort.prototype natively and shares the +// same listener list as addEventListener, so no JS-side registry or wrapper +// is needed. const _MessagePort = globalThis.MessagePort; -injectFakeEmitter(_MessagePort); - const MessagePort = _MessagePort; // node's close(cb) registers cb as a one-time "close" listener before the native close. @@ -814,15 +636,29 @@ function fakeParentPort() { value: self.removeEventListener.bind(self), }); - Object.defineProperty(fake, "removeListener", { - value: self.removeEventListener.bind(self), - enumerable: false, - }); - - Object.defineProperty(fake, "addListener", { - value: self.addEventListener.bind(self), - enumerable: false, - }); + // The NodeEventTarget surface on MessagePort.prototype requires a real + // MessagePort receiver; this stand-in forwards to the global scope instead. + function on(this: any, type: string, listener: any) { + self.addEventListener(type, listener, { $kIsNodeStyleListener: true } as AddEventListenerOptions); + return this; + } + function once(this: any, type: string, listener: any) { + self.addEventListener(type, listener, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); + return this; + } + function off(this: any, type: string, listener: any) { + self.removeEventListener(type, listener); + return this; + } + for (const [name, fn] of [ + ["on", on], + ["addListener", on], + ["once", once], + ["off", off], + ["removeListener", off], + ] as const) { + Object.defineProperty(fake, name, { value: fn, enumerable: false, configurable: true, writable: true }); + } return fake; } diff --git a/src/jsc/bindings/webcore/AddEventListenerOptions.h b/src/jsc/bindings/webcore/AddEventListenerOptions.h index 377792966ecf..d3335b98c68e 100644 --- a/src/jsc/bindings/webcore/AddEventListenerOptions.h +++ b/src/jsc/bindings/webcore/AddEventListenerOptions.h @@ -48,6 +48,11 @@ struct AddEventListenerOptions : EventListenerOptions { // modules, mirroring Node.js's kResistStopPropagation: a listener registered // with it still runs after another listener called stopImmediatePropagation(). bool resistStopPropagation { false }; + + // Not part of the DOM standard. Mirrors Node.js's kIsNodeStyleListener: when + // set, the listener is invoked with the event's underlying value (data/detail/ + // error) instead of the Event wrapper. MessagePort.prototype.on/.once set it. + bool isNodeStyleListener { false }; }; } // namespace WebCore diff --git a/src/jsc/bindings/webcore/EventFactory.cpp b/src/jsc/bindings/webcore/EventFactory.cpp index 062a6aac8e82..56d6777702a9 100644 --- a/src/jsc/bindings/webcore/EventFactory.cpp +++ b/src/jsc/bindings/webcore/EventFactory.cpp @@ -88,8 +88,8 @@ JSC::JSValue toJSNewlyCreated(JSC::JSGlobalObject*, JSDOMGlobalObject* globalObj } // case CompositionEventInterfaceType: // return createWrapper(globalObject, WTF::move(impl)); - // case CustomEventInterfaceType: - // return createWrapper(globalObject, WTF::move(impl)); + case CustomEventInterfaceType: + return createWrapper(globalObject, WTF::move(impl)); // #if ENABLE(DEVICE_ORIENTATION) // case DeviceMotionEventInterfaceType: // return createWrapper(globalObject, WTF::move(impl)); diff --git a/src/jsc/bindings/webcore/EventHeaders.h b/src/jsc/bindings/webcore/EventHeaders.h index 6692b9234a80..6f2561297e01 100644 --- a/src/jsc/bindings/webcore/EventHeaders.h +++ b/src/jsc/bindings/webcore/EventHeaders.h @@ -78,8 +78,8 @@ #include "JSCloseEvent.h" // #include "CompositionEvent.h" // #include "JSCompositionEvent.h" -// #include "CustomEvent.h" -// #include "JSCustomEvent.h" +#include "CustomEvent.h" +#include "JSCustomEvent.h" // #if ENABLE(DEVICE_ORIENTATION) // #include "DeviceMotionEvent.h" // #include "JSDeviceMotionEvent.h" diff --git a/src/jsc/bindings/webcore/EventListener.h b/src/jsc/bindings/webcore/EventListener.h index c164c0f170f1..81373b68aada 100644 --- a/src/jsc/bindings/webcore/EventListener.h +++ b/src/jsc/bindings/webcore/EventListener.h @@ -52,6 +52,12 @@ class EventListener : public RefCounted, public CanMakeWeakPtrmarkAsRemoved(); + m_entries.removeAt(i); + return true; + } + } + + return false; +} + EventListenerVector* EventListenerMap::find(const AtomString& eventType) { for (auto& entry : m_entries) { diff --git a/src/jsc/bindings/webcore/EventListenerMap.h b/src/jsc/bindings/webcore/EventListenerMap.h index af32801205e6..7ba52bb9c3ff 100644 --- a/src/jsc/bindings/webcore/EventListenerMap.h +++ b/src/jsc/bindings/webcore/EventListenerMap.h @@ -61,6 +61,7 @@ class EventListenerMap { void replace(const AtomString& eventType, EventListener& oldListener, Ref&& newListener, const RegisteredEventListener::Options&); RegisteredEventListener* add(const AtomString& eventType, Ref&&, const RegisteredEventListener::Options&); bool remove(const AtomString& eventType, EventListener&, bool useCapture); + bool removeAll(const AtomString& eventType); WEBCORE_EXPORT EventListenerVector* find(const AtomString& eventType); const EventListenerVector* find(const AtomString& eventType) const { return const_cast(this)->find(eventType); } Vector eventTypes() const; diff --git a/src/jsc/bindings/webcore/EventTarget.cpp b/src/jsc/bindings/webcore/EventTarget.cpp index e3d74425215e..9c13850c22ab 100644 --- a/src/jsc/bindings/webcore/EventTarget.cpp +++ b/src/jsc/bindings/webcore/EventTarget.cpp @@ -103,7 +103,7 @@ bool EventTarget::addEventListener(const AtomString& eventType, Refcallback().handleEvent(context, event); + if (registeredListener->isNodeStyleListener()) [[unlikely]] + registeredListener->callback().handleEventNodeStyle(context, event); + else + registeredListener->callback().handleEvent(context, event); // InspectorInstrumentation::didHandleEvent(context, event, *registeredListener); if (registeredListener->isPassive()) @@ -366,6 +369,18 @@ const EventListenerVector& EventTarget::eventListeners(const AtomString& eventTy return listenerVector ? *listenerVector : emptyVector.get(); } +void EventTarget::removeAllEventListenersForType(const AtomString& eventType) +{ + auto* data = eventTargetData(); + if (!data) + return; + if (data->eventListenerMap.removeAll(eventType)) { + if (this->onDidChangeListener) [[unlikely]] + this->onDidChangeListener(*this, eventType, OnDidChangeListenerKind::Clear); + eventListenersDidChange(); + } +} + void EventTarget::removeAllEventListeners() { // auto& threadData = threadGlobalData(); diff --git a/src/jsc/bindings/webcore/EventTarget.h b/src/jsc/bindings/webcore/EventTarget.h index 52b34ad09b54..4eacc61cf37e 100644 --- a/src/jsc/bindings/webcore/EventTarget.h +++ b/src/jsc/bindings/webcore/EventTarget.h @@ -114,6 +114,7 @@ class EventTarget : public ScriptWrappable, public CanMakeWeakPtrWithBitField AddEventListenerOptions convertDictionary(JS result.resistStopPropagation = convert(lexicalGlobalObject, resistStopPropagationValue); RETURN_IF_EXCEPTION(throwScope, {}); } + JSValue isNodeStyleListenerValue = object->get(&lexicalGlobalObject, builtinNames(vm).kIsNodeStyleListenerPrivateName()); + RETURN_IF_EXCEPTION(throwScope, {}); + if (!isNodeStyleListenerValue.isUndefined()) { + result.isNodeStyleListener = convert(lexicalGlobalObject, isNodeStyleListenerValue); + RETURN_IF_EXCEPTION(throwScope, {}); + } } return result; } diff --git a/src/jsc/bindings/webcore/JSEventListener.cpp b/src/jsc/bindings/webcore/JSEventListener.cpp index e1d7f4eb3eb7..0173a1f266c2 100644 --- a/src/jsc/bindings/webcore/JSEventListener.cpp +++ b/src/jsc/bindings/webcore/JSEventListener.cpp @@ -32,6 +32,9 @@ // #include "JSDocument.h" #include "JSEvent.h" #include "JSEventTarget.h" +#include "JSMessageEvent.h" +#include "CustomEvent.h" +#include "ErrorEvent.h" #include "WebCoreJSClientData.h" // #include "JSExecState.h" // #include "JSExecStateInstrumentation.h" @@ -43,6 +46,7 @@ #include #include #include +#include namespace WebCore { using namespace JSC; @@ -141,6 +145,43 @@ JSC_DEFINE_HOST_FUNCTION(jsFunctionEmitUncaughtExceptionNextTick, (JSC::JSGlobal return JSC::JSValue::encode(JSC::jsUndefined()); } +// A listener registered with Node.js's kIsNodeStyleListener receives the value +// the event carries rather than the Event wrapper: MessageEvent → .data, +// CustomEvent → .detail, ErrorEvent → .error; other events carry no value. +static JSC::JSValue nodeStyleArgumentForEvent(JSC::JSGlobalObject* lexicalGlobalObject, JSDOMGlobalObject* globalObject, Event& event) +{ + switch (event.eventInterface()) { + case MessageEventInterfaceType: { + auto& me = static_cast(event); + // Native MessagePort dispatch caches the deserialized payload on the + // wrapper; reuse it so node-style and EventTarget-style listeners see + // the same value by identity. + if (auto cached = me.cachedData().getValue({})) + return cached; + if (std::holds_alternative(me.data())) + return me.jsData().getValue(JSC::jsNull()); + // Serialized but not yet deserialized (e.g. BroadcastChannel): fall + // through to the wrapper's .data getter so the result is cached. + JSValue jsEvent = toJS(lexicalGlobalObject, globalObject, &event); + if (auto* wrapper = dynamicDowncast(jsEvent.getObject())) + return wrapper->data(*lexicalGlobalObject); + return JSC::jsNull(); + } + case CustomEventInterfaceType: + return static_cast(event).detail().getValue(JSC::jsNull()); + case ErrorEventInterfaceType: + return static_cast(event).error(*lexicalGlobalObject); + default: + return JSC::jsUndefined(); + } +} + +void JSEventListener::handleEventNodeStyle(ScriptExecutionContext& scriptExecutionContext, Event& event) +{ + SetForScope nodeStyle { m_invokeAsNodeStyle, true }; + handleEvent(scriptExecutionContext, event); +} + void JSEventListener::handleEvent(ScriptExecutionContext& scriptExecutionContext, Event& event) { if (scriptExecutionContext.isJSExecutionForbidden()) @@ -222,7 +263,7 @@ void JSEventListener::handleEvent(ScriptExecutionContext& scriptExecutionContext Ref protectedThis(*this); MarkedArgumentBuffer args; - args.append(toJS(lexicalGlobalObject, globalObject, &event)); + args.append(m_invokeAsNodeStyle ? nodeStyleArgumentForEvent(lexicalGlobalObject, globalObject, event) : toJS(lexicalGlobalObject, globalObject, &event)); ASSERT(!args.hasOverflowed()); // JSExecState::instrumentFunction(&scriptExecutionContext, callData); diff --git a/src/jsc/bindings/webcore/JSEventListener.h b/src/jsc/bindings/webcore/JSEventListener.h index fb8da1bb6296..f543a6b495dc 100644 --- a/src/jsc/bindings/webcore/JSEventListener.h +++ b/src/jsc/bindings/webcore/JSEventListener.h @@ -78,6 +78,7 @@ class JSEventListener : public EventListener { JSEventListener(JSC::JSObject* function, JSC::JSObject* wrapper, bool isAttribute, CreatedFromMarkup, DOMWrapperWorld&); void handleEvent(ScriptExecutionContext&, Event&) override; + void handleEventNodeStyle(ScriptExecutionContext&, Event&) final; void setWrapperWhenInitializingJSFunction(JSC::VM&, JSC::JSObject* wrapper) const { m_wrapper = JSC::Weak(wrapper); } private: @@ -85,6 +86,11 @@ class JSEventListener : public EventListener { bool m_wasCreatedFromMarkup : 1; mutable bool m_isInitialized : 1; + // Transient dispatch state: set by handleEventNodeStyle for the duration + // of a single handleEvent call so the shared invocation path knows to pass + // the event's carried value instead of the Event wrapper. Not a bitfield + // because SetForScope takes a reference. + bool m_invokeAsNodeStyle { false }; mutable JSC::Weak m_jsFunction; mutable JSC::Weak m_wrapper; diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index 6bc8c19950ed..3e7415d4cdec 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -22,6 +22,8 @@ #include "JSMessagePort.h" #include "ActiveDOMObject.h" +#include "AddEventListenerOptions.h" +#include "CustomEvent.h" #include "EventNames.h" #include "ExtendedDOMClientIsoSubspaces.h" #include "ExtendedDOMIsoSubspaces.h" @@ -32,14 +34,18 @@ #include "JSDOMConvertAny.h" #include "JSDOMConvertBase.h" #include "JSDOMConvertDictionary.h" +#include "JSDOMConvertEventListener.h" +#include "JSDOMConvertNullable.h" #include "JSDOMConvertObject.h" #include "JSDOMConvertSequences.h" +#include "JSDOMConvertStrings.h" #include "JSDOMExceptionHandling.h" #include "JSDOMGlobalObjectInlines.h" #include "JSDOMOperation.h" #include "JSDOMWrapperCache.h" #include "JSEventListener.h" #include "JSStructuredSerializeOptions.h" +#include "MessageEvent.h" #include "ScriptExecutionContext.h" #include "WebCoreJSClientData.h" // #include "WebCoreOpaqueRootInlines.h" @@ -65,6 +71,20 @@ static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_close); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_ref); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_unref); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_hasRef); +// NodeEventTarget surface (node lib/internal/event_target.js): a curated subset +// of EventEmitter that reads and writes the same listener list as EventTarget, +// so .on()/addEventListener() dedupe against each other, listenerCount()/ +// eventNames()/removeAllListeners() see every listener, and emit() returns a +// boolean while passing the raw argument to node-style listeners. +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_on); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_once); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_off); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_emit); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_listenerCount); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_eventNames); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_removeAllListeners); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_setMaxListeners); +static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_getMaxListeners); // Attributes @@ -138,6 +158,23 @@ static const HashTableValue JSMessagePortPrototypeTableValues[] = { { "hasRef"_s, static_cast(JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_hasRef, 0 } }, }; +// NodeEventTarget methods live on an intermediate prototype between +// MessagePort.prototype and EventTarget.prototype so that +// Object.getOwnPropertyNames(MessagePort.prototype) matches node. +static const HashTableValue JSMessagePortNodeEventTargetTableValues[] = { + { "on"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_on, 2 } }, + { "addListener"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_on, 2 } }, + { "once"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_once, 2 } }, + { "off"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_off, 2 } }, + { "removeListener"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_off, 2 } }, + { "emit"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_emit, 1 } }, + { "listenerCount"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_listenerCount, 1 } }, + { "eventNames"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_eventNames, 0 } }, + { "removeAllListeners"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_removeAllListeners, 0 } }, + { "setMaxListeners"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_setMaxListeners, 1 } }, + { "getMaxListeners"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_getMaxListeners, 0 } }, +}; + const ClassInfo JSMessagePortPrototype::s_info = { "MessagePort"_s, &Base::s_info, nullptr, nullptr, CREATE_METHOD_TABLE(JSMessagePortPrototype) }; void JSMessagePortPrototype::finishCreation(VM& vm) @@ -158,7 +195,13 @@ JSMessagePort::JSMessagePort(Structure* structure, JSDOMGlobalObject& globalObje JSObject* JSMessagePort::createPrototype(VM& vm, JSDOMGlobalObject& globalObject) { - auto* structure = JSMessagePortPrototype::createStructure(vm, &globalObject, JSEventTarget::prototype(vm, globalObject)); + auto* eventTargetPrototype = JSEventTarget::prototype(vm, globalObject); + auto* nodeEventTargetStructure = JSC::JSFinalObject::createStructure(vm, &globalObject, eventTargetPrototype, 0); + nodeEventTargetStructure->setMayBePrototype(true); + auto* nodeEventTargetPrototype = JSC::JSFinalObject::create(vm, nodeEventTargetStructure); + reifyStaticProperties(vm, JSMessagePort::info(), JSMessagePortNodeEventTargetTableValues, *nodeEventTargetPrototype); + + auto* structure = JSMessagePortPrototype::createStructure(vm, &globalObject, nodeEventTargetPrototype); structure->setMayBePrototype(true); return JSMessagePortPrototype::create(vm, &globalObject, structure); } @@ -402,6 +445,181 @@ JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_hasRef, (JSGlobalObject return IDLOperation::call(*lexicalGlobalObject, *callFrame, "hasRef"); } +// NodeEventTarget ------------------------------------------------------------ + +static inline JSC::EncodedJSValue messagePortNodeAddListener(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis, bool once) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + auto& impl = castedThis->wrapped(); + EnsureStillAliveScope argument0 = callFrame->argument(0); + auto type = convert>(*lexicalGlobalObject, argument0.value()); + RETURN_IF_EXCEPTION(throwScope, {}); + EnsureStillAliveScope argument1 = callFrame->argument(1); + auto listener = convert>>(*lexicalGlobalObject, argument1.value(), *castedThis, [](JSC::JSGlobalObject& lexicalGlobalObject, JSC::ThrowScope& scope) { throwArgumentMustBeObjectError(lexicalGlobalObject, scope, 1, "listener"_s, "MessagePort"_s, "on"_s); }); + RETURN_IF_EXCEPTION(throwScope, {}); + if (listener) { + AddEventListenerOptions options; + options.once = once; + options.isNodeStyleListener = true; + static_cast(impl).addEventListener(WTF::move(type), listener.releaseNonNull(), options); + vm.writeBarrier(&static_cast(*castedThis), argument1.value()); + } + return JSValue::encode(castedThis); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_onBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + return messagePortNodeAddListener(lexicalGlobalObject, callFrame, castedThis, false); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_on, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "on"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_onceBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + return messagePortNodeAddListener(lexicalGlobalObject, callFrame, castedThis, true); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_once, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "once"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_offBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + auto& impl = castedThis->wrapped(); + EnsureStillAliveScope argument0 = callFrame->argument(0); + auto type = convert>(*lexicalGlobalObject, argument0.value()); + RETURN_IF_EXCEPTION(throwScope, {}); + EnsureStillAliveScope argument1 = callFrame->argument(1); + auto listener = convert>>(*lexicalGlobalObject, argument1.value(), *castedThis, [](JSC::JSGlobalObject& lexicalGlobalObject, JSC::ThrowScope& scope) { throwArgumentMustBeObjectError(lexicalGlobalObject, scope, 1, "listener"_s, "MessagePort"_s, "off"_s); }); + RETURN_IF_EXCEPTION(throwScope, {}); + if (listener) + static_cast(impl).removeEventListener(WTF::move(type), *listener, {}); + return JSValue::encode(castedThis); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_off, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "off"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_emitBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + auto& impl = castedThis->wrapped(); + EnsureStillAliveScope argument0 = callFrame->argument(0); + auto type = convert>(*lexicalGlobalObject, argument0.value()); + RETURN_IF_EXCEPTION(throwScope, {}); + bool had = impl.hasEventListeners(type); + if (had) { + EnsureStillAliveScope argument1 = callFrame->argument(1); + JSValue arg = argument1.value(); + // node's MessagePort[kCreateEvent]: a MessageEvent for message / + // messageerror, a CustomEvent for everything else. Node-style + // listeners recover the raw argument by identity at invoke time. + if (type == eventNames().messageEvent || type == eventNames().messageerrorEvent) { + MessageEvent::Init init; + init.data = arg; + auto event = MessageEvent::create(type, WTF::move(init)); + impl.dispatchEvent(event.get()); + } else { + CustomEvent::Init init; + init.detail = arg; + auto event = CustomEvent::create(type, init); + impl.dispatchEvent(event.get()); + } + } + return JSValue::encode(jsBoolean(had)); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_emit, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "emit"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_listenerCountBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + EnsureStillAliveScope argument0 = callFrame->argument(0); + auto type = convert>(*lexicalGlobalObject, argument0.value()); + RETURN_IF_EXCEPTION(throwScope, {}); + return JSValue::encode(jsNumber(castedThis->wrapped().eventListeners(type).size())); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_listenerCount, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "listenerCount"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_eventNamesBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + UNUSED_PARAM(callFrame); + auto types = castedThis->wrapped().eventTypes(); + MarkedArgumentBuffer values; + for (auto& t : types) + values.append(jsString(vm, t.string())); + return JSValue::encode(constructArray(lexicalGlobalObject, static_cast(nullptr), values)); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_eventNames, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "eventNames"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_removeAllListenersBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + auto& impl = castedThis->wrapped(); + if (callFrame->argumentCount() == 0) { + impl.removeAllEventListeners(); + } else { + EnsureStillAliveScope argument0 = callFrame->uncheckedArgument(0); + auto type = convert>(*lexicalGlobalObject, argument0.value()); + RETURN_IF_EXCEPTION(throwScope, {}); + impl.removeAllEventListenersForType(type); + } + return JSValue::encode(castedThis); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_removeAllListeners, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "removeAllListeners"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_setMaxListenersBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + UNUSED_PARAM(lexicalGlobalObject); + castedThis->wrapped().setNodeMaxListeners(callFrame->argument(0).toUInt32(lexicalGlobalObject)); + return JSValue::encode(castedThis); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_setMaxListeners, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "setMaxListeners"); +} + +static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_getMaxListenersBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) +{ + UNUSED_PARAM(lexicalGlobalObject); + UNUSED_PARAM(callFrame); + return JSValue::encode(jsNumber(castedThis->wrapped().nodeMaxListeners())); +} + +JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_getMaxListeners, (JSGlobalObject * lexicalGlobalObject, CallFrame* callFrame)) +{ + return IDLOperation::call(*lexicalGlobalObject, *callFrame, "getMaxListeners"); +} + JSC::GCClient::IsoSubspace* JSMessagePort::subspaceForImpl(JSC::VM& vm) { return WebCore::subspaceForImpl( diff --git a/src/jsc/bindings/webcore/MessagePort.h b/src/jsc/bindings/webcore/MessagePort.h index 86d047d320b0..9ddd36b4d1cb 100644 --- a/src/jsc/bindings/webcore/MessagePort.h +++ b/src/jsc/bindings/webcore/MessagePort.h @@ -115,6 +115,9 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget, // Report the actual loop-ref state (matches Node's uv_has_ref), not the intent flag. bool jsHasRef() { return m_hasRef || m_listenerLoopRefActive; } + uint32_t nodeMaxListeners() const { return m_nodeMaxListeners; } + void setNodeMaxListeners(uint32_t n) { m_nodeMaxListeners = n; } + private: MessagePort(ScriptExecutionContext&, Ref&&, uint8_t side); @@ -165,6 +168,7 @@ class MessagePort final : public ContextDestructionObserver, public EventTarget, bool m_listenerLoopRefActive { false }; uint32_t m_messageEventCount { 0 }; + uint32_t m_nodeMaxListeners { 10 }; static void onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind); // Reconciles the listener event-loop ref with (m_isRefd && m_messageEventCount > 0). void updateListenerEventLoopRef(); diff --git a/src/jsc/bindings/webcore/RegisteredEventListener.h b/src/jsc/bindings/webcore/RegisteredEventListener.h index 6b7d847595e9..6dee20b6a17d 100644 --- a/src/jsc/bindings/webcore/RegisteredEventListener.h +++ b/src/jsc/bindings/webcore/RegisteredEventListener.h @@ -36,11 +36,12 @@ class WeakPtrImplWithEventTargetData; class RegisteredEventListener : public RefCounted { public: struct Options { - Options(bool capture = false, bool passive = false, bool once = false, bool resistStopPropagation = false) + Options(bool capture = false, bool passive = false, bool once = false, bool resistStopPropagation = false, bool isNodeStyleListener = false) : capture(capture) , passive(passive) , once(once) , resistStopPropagation(resistStopPropagation) + , isNodeStyleListener(isNodeStyleListener) { } @@ -48,6 +49,7 @@ class RegisteredEventListener : public RefCounted { bool passive; bool once; bool resistStopPropagation; + bool isNodeStyleListener; }; static Ref create(Ref&& listener, const Options& options) @@ -63,6 +65,7 @@ class RegisteredEventListener : public RefCounted { bool isOnce() const { return m_isOnce; } bool wasRemoved() const { return m_wasRemoved; } bool resistsStopPropagation() const { return m_resistStopPropagation; } + bool isNodeStyleListener() const { return m_isNodeStyleListener; } void markAsRemoved(); @@ -81,6 +84,7 @@ class RegisteredEventListener : public RefCounted { , m_isOnce(options.once) , m_wasRemoved(false) , m_resistStopPropagation(options.resistStopPropagation) + , m_isNodeStyleListener(options.isNodeStyleListener) , m_callback(WTF::move(listener)) { } @@ -90,6 +94,7 @@ class RegisteredEventListener : public RefCounted { bool m_isOnce : 1; bool m_wasRemoved : 1; bool m_resistStopPropagation : 1; + bool m_isNodeStyleListener : 1; uint32_t m_abortAlgorithmIdentifier { 0 }; Ref m_callback; WeakPtr m_abortSignal; diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts new file mode 100644 index 000000000000..38053f185c84 --- /dev/null +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -0,0 +1,404 @@ +import { test, expect, describe } from "bun:test"; +import { bunEnv, bunExe } from "harness"; + +// MessagePort is a NodeEventTarget in node: .on/.addListener/.once share the +// same listener list as addEventListener, so listenerCount/eventNames/ +// removeAllListeners/getEventListeners see both, cross-remove works, and +// emit() returns a boolean while passing the raw argument by identity. +// +// Each row of the coherence matrix runs in its own subprocess so that a throw +// in one row cannot mask another, and so row 15 can observe the surface +// without any prior import of node:worker_threads. +describe.concurrent("MessagePort NodeEventTarget", () => { + // Row 15 first so nothing else in this file can accidentally load + // node:worker_threads into a shared state that would mask the bug. + test("surface exists without importing node:worker_threads", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + if (require.cache["node:worker_threads"]) throw new Error("worker_threads preloaded"); + const { port1 } = new MessageChannel(); + const names = ["on","off","once","emit","addListener","removeListener","listenerCount","eventNames","removeAllListeners","setMaxListeners","getMaxListeners"]; + for (const n of names) { + if (typeof port1[n] !== "function") throw new Error("missing " + n + ": " + typeof port1[n]); + } + port1.close(); + process.stdout.write("OK"); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("OK"); + expect(exitCode).toBe(0); + }); + + test("same fn via .on + addEventListener is one listener, invoked once", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + let n = 0; + const fn = () => { n++ }; + port1.on("message", fn); + port1.addEventListener("message", fn); + port2.postMessage("x"); + setImmediate(() => { + port1.close(); + process.stdout.write(String(n)); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("1"); + expect(exitCode).toBe(0); + }); + + test("listenerCount counts both .on and addEventListener listeners", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + port1.on("message", () => {}); + port1.addEventListener("message", () => {}); + process.stdout.write(String(port1.listenerCount("message"))); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("2"); + expect(exitCode).toBe(0); + }); + + test("eventNames reports types registered via addEventListener", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + port1.addEventListener("foo", () => {}); + process.stdout.write(JSON.stringify(port1.eventNames())); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe('["foo"]'); + expect(exitCode).toBe(0); + }); + + test("removeEventListener removes a listener added via .on", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + const fn = () => {}; + port1.on("message", fn); + port1.removeEventListener("message", fn); + process.stdout.write(String(port1.listenerCount("message"))); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("0"); + expect(exitCode).toBe(0); + }); + + test("removeAllListeners clears addEventListener listeners too", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + let n = 0; + port1.addEventListener("message", () => { n++ }); + port1.on("message", () => { n++ }); + port1.removeAllListeners("message"); + port1.start(); + port2.postMessage("x"); + setImmediate(() => { + port1.close(); + process.stdout.write(String(n)); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("0"); + expect(exitCode).toBe(0); + }); + + test("emit passes the argument by identity to .on listeners", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + const payload = { a: 1 }; + let got; + port1.on("message", v => { got = v }); + port1.emit("message", payload); + process.stdout.write(String(got === payload)); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("true"); + expect(exitCode).toBe(0); + }); + + test("emit returns a boolean (true when listeners, false otherwise)", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + const r1 = port1.emit("nope", 1); + port1.on("hit", () => {}); + const r2 = port1.emit("hit", 1); + process.stdout.write(JSON.stringify([r1, r2])); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("[false,true]"); + expect(exitCode).toBe(0); + }); + + test("emit with a primitive payload does not throw", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + let got; + port1.on("message", v => { got = v }); + port1.emit("message", 1); + process.stdout.write(String(got)); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("1"); + expect(exitCode).toBe(0); + }); + + test("emit('error', err) passes the error by identity", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + const err = new Error("boom"); + let got; + port1.on("error", e => { got = e }); + port1.emit("error", err); + process.stdout.write(String(got === err)); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("true"); + expect(exitCode).toBe(0); + }); + + test(".once dedupes against .on for the same fn", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + const fn = () => {}; + port1.on("foo", fn); + port1.once("foo", fn); + process.stdout.write(String(port1.listenerCount("foo"))); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("1"); + expect(exitCode).toBe(0); + }); + + test("events.getEventListeners returns the user fn, not a wrapper", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { getEventListeners } = require("node:events"); + const { port1 } = new MessageChannel(); + const fn = () => {}; + port1.on("message", fn); + const list = getEventListeners(port1, "message"); + process.stdout.write(String(list.length === 1 && list[0] === fn)); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("true"); + expect(exitCode).toBe(0); + }); + + test("addEventListener listeners still receive an Event; .on listeners get the payload", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + let a, b; + port1.on("message", v => { a = v }); + port1.addEventListener("message", ev => { b = ev }); + port2.postMessage(42); + setImmediate(() => { + port1.close(); + process.stdout.write(JSON.stringify([a, b instanceof MessageEvent, b.data])); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("[42,true,42]"); + expect(exitCode).toBe(0); + }); + + test("emit with an addEventListener listener delivers an Event with matching data/detail", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + let msg, custom; + port1.addEventListener("message", ev => { msg = ev.data }); + port1.addEventListener("foo", ev => { custom = ev.detail }); + port1.emit("message", 7); + port1.emit("foo", 8); + process.stdout.write(JSON.stringify([msg, custom])); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("[7,8]"); + expect(exitCode).toBe(0); + }); + + test(".off removes a listener added via addEventListener", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1 } = new MessageChannel(); + const fn = () => {}; + port1.addEventListener("message", fn); + port1.off("message", fn); + process.stdout.write(String(port1.listenerCount("message"))); + port1.close(); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("0"); + expect(exitCode).toBe(0); + }); +}); From 86020db17c3ccfe426a27c57c52ae8d33799ae20 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Sat, 25 Jul 2026 21:30:48 +0000 Subject: [PATCH 02/13] [autofix.ci] apply automated fixes --- .../node/worker_threads/message-port-node-event-target.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts index 38053f185c84..c22daed38b08 100644 --- a/test/js/node/worker_threads/message-port-node-event-target.test.ts +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -1,4 +1,4 @@ -import { test, expect, describe } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe } from "harness"; // MessagePort is a NodeEventTarget in node: .on/.addListener/.once share the From c53eb03787321fd57962bdc67c0d947443eecf07 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 21:42:05 +0000 Subject: [PATCH 03/13] trim multi-line comments flagged by comment-cop --- src/js/node/worker_threads.ts | 8 +------- src/jsc/bindings/webcore/EventListener.h | 7 ++----- src/jsc/bindings/webcore/JSEventListener.cpp | 13 +++++-------- src/jsc/bindings/webcore/JSEventListener.h | 5 +---- src/jsc/bindings/webcore/JSMessagePort.cpp | 16 +++++----------- 5 files changed, 14 insertions(+), 35 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 081dd4f10f53..bb6eec262d93 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -100,11 +100,6 @@ type NodeWorkerOptions = import("node:worker_threads").WorkerOptions; // after their Worker exits let urlRevokeRegistry: FinalizationRegistry | undefined = undefined; -// MessagePort is a NodeEventTarget: the node-style emitter surface (.on / -// .once / .off / .emit / listenerCount / eventNames / removeAllListeners / -// set|getMaxListeners) lives on MessagePort.prototype natively and shares the -// same listener list as addEventListener, so no JS-side registry or wrapper -// is needed. const _MessagePort = globalThis.MessagePort; const MessagePort = _MessagePort; @@ -636,8 +631,7 @@ function fakeParentPort() { value: self.removeEventListener.bind(self), }); - // The NodeEventTarget surface on MessagePort.prototype requires a real - // MessagePort receiver; this stand-in forwards to the global scope instead. + // MessagePort.prototype.on/etc. require a real MessagePort receiver; forward to self. function on(this: any, type: string, listener: any) { self.addEventListener(type, listener, { $kIsNodeStyleListener: true } as AddEventListenerOptions); return this; diff --git a/src/jsc/bindings/webcore/EventListener.h b/src/jsc/bindings/webcore/EventListener.h index 81373b68aada..c3a8372b9922 100644 --- a/src/jsc/bindings/webcore/EventListener.h +++ b/src/jsc/bindings/webcore/EventListener.h @@ -52,11 +52,8 @@ class EventListener : public RefCounted, public CanMakeWeakPtr(event); - // Native MessagePort dispatch caches the deserialized payload on the - // wrapper; reuse it so node-style and EventTarget-style listeners see - // the same value by identity. + // Reuse the deserialized payload cached by MessagePort dispatch so + // .on listeners and event.data readers see the same value by identity. if (auto cached = me.cachedData().getValue({})) return cached; if (std::holds_alternative(me.data())) return me.jsData().getValue(JSC::jsNull()); - // Serialized but not yet deserialized (e.g. BroadcastChannel): fall - // through to the wrapper's .data getter so the result is cached. + // Still serialized: read through the wrapper so the result is cached. JSValue jsEvent = toJS(lexicalGlobalObject, globalObject, &event); if (auto* wrapper = dynamicDowncast(jsEvent.getObject())) return wrapper->data(*lexicalGlobalObject); diff --git a/src/jsc/bindings/webcore/JSEventListener.h b/src/jsc/bindings/webcore/JSEventListener.h index f543a6b495dc..98dffe4a4198 100644 --- a/src/jsc/bindings/webcore/JSEventListener.h +++ b/src/jsc/bindings/webcore/JSEventListener.h @@ -86,10 +86,7 @@ class JSEventListener : public EventListener { bool m_wasCreatedFromMarkup : 1; mutable bool m_isInitialized : 1; - // Transient dispatch state: set by handleEventNodeStyle for the duration - // of a single handleEvent call so the shared invocation path knows to pass - // the event's carried value instead of the Event wrapper. Not a bitfield - // because SetForScope takes a reference. + // Transient: set by handleEventNodeStyle around one handleEvent call. bool m_invokeAsNodeStyle { false }; mutable JSC::Weak m_jsFunction; mutable JSC::Weak m_wrapper; diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index 3e7415d4cdec..aad6f7cb546c 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -71,11 +71,8 @@ static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_close); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_ref); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_unref); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_hasRef); -// NodeEventTarget surface (node lib/internal/event_target.js): a curated subset -// of EventEmitter that reads and writes the same listener list as EventTarget, -// so .on()/addEventListener() dedupe against each other, listenerCount()/ -// eventNames()/removeAllListeners() see every listener, and emit() returns a -// boolean while passing the raw argument to node-style listeners. +// NodeEventTarget (node lib/internal/event_target.js): an EventEmitter subset +// that reads and writes the same listener list as addEventListener. static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_on); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_once); static JSC_DECLARE_HOST_FUNCTION(jsMessagePortPrototypeFunction_off); @@ -158,9 +155,7 @@ static const HashTableValue JSMessagePortPrototypeTableValues[] = { { "hasRef"_s, static_cast(JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_hasRef, 0 } }, }; -// NodeEventTarget methods live on an intermediate prototype between -// MessagePort.prototype and EventTarget.prototype so that -// Object.getOwnPropertyNames(MessagePort.prototype) matches node. +// On an intermediate prototype so getOwnPropertyNames(MessagePort.prototype) matches node. static const HashTableValue JSMessagePortNodeEventTargetTableValues[] = { { "on"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_on, 2 } }, { "addListener"_s, static_cast(JSC::PropertyAttribute::DontEnum | JSC::PropertyAttribute::Function), NoIntrinsic, { HashTableValue::NativeFunctionType, jsMessagePortPrototypeFunction_on, 2 } }, @@ -521,9 +516,8 @@ static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_emitBody(JSC::J if (had) { EnsureStillAliveScope argument1 = callFrame->argument(1); JSValue arg = argument1.value(); - // node's MessagePort[kCreateEvent]: a MessageEvent for message / - // messageerror, a CustomEvent for everything else. Node-style - // listeners recover the raw argument by identity at invoke time. + // node's MessagePort[kCreateEvent]: MessageEvent for message/messageerror, + // CustomEvent otherwise; node-style listeners recover arg at invoke time. if (type == eventNames().messageEvent || type == eventNames().messageerrorEvent) { MessageEvent::Init init; init.data = arg; From 54bac106624108f0c75b988d5d3e36c11a885052 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 22:02:08 +0000 Subject: [PATCH 04/13] address review: ThrowScope for setMaxListeners, reset flags on Clear, parentPort shadows - setMaxListeners: toUInt32 can enter user JS; declare a ThrowScope and RETURN_IF_EXCEPTION before mutating state. - onDidChangeListenerImpl: reset m_hasMessageEventListener / m_hasCloseEventListener on Clear so removeAllListeners() pauses delivery (re-buffers) instead of leaving the flags stale and draining to zero listeners. - fakeParentPort: shadow listenerCount / eventNames / removeAllListeners / emit / set|getMaxListeners so they don't fall through to the native prototype (which requires a real MessagePort receiver). Tracks parentPort-registered listeners in a local map so removeAllListeners() only removes what parentPort added (self carries internal listeners). Two new tests cover the re-buffer path and the parentPort surface. --- src/js/node/worker_threads.ts | 51 +++++++++++++- src/jsc/bindings/webcore/JSMessagePort.cpp | 7 +- src/jsc/bindings/webcore/MessagePort.cpp | 10 ++- .../message-port-node-event-target.test.ts | 67 +++++++++++++++++++ 4 files changed, 131 insertions(+), 4 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index bb6eec262d93..dcaf8ce5daf3 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -4,6 +4,7 @@ declare const self: typeof globalThis; type WebWorker = InstanceType; const EventEmitter = require("node:events"); +const { SafeMap } = require("internal/primordials"); const Readable = require("internal/streams/readable"); const Writable = require("internal/streams/writable"); const { throwNotImplemented, warnNotImplementedOnce } = require("internal/shared"); @@ -631,25 +632,73 @@ function fakeParentPort() { value: self.removeEventListener.bind(self), }); - // MessagePort.prototype.on/etc. require a real MessagePort receiver; forward to self. + // MessagePort.prototype.on/etc. require a real MessagePort receiver; shadow + // them so calls on this stand-in forward to the global scope. A local map + // tracks which listeners were added via parentPort so listenerCount/ + // eventNames/removeAllListeners see only those (self may carry internal + // listeners the user must not touch). + const byType = new SafeMap(); function on(this: any, type: string, listener: any) { self.addEventListener(type, listener, { $kIsNodeStyleListener: true } as AddEventListenerOptions); + let set = byType.get(type); + if (!set) byType.set(type, (set = new Set())); + set.add(listener); return this; } function once(this: any, type: string, listener: any) { self.addEventListener(type, listener, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); + let set = byType.get(type); + if (!set) byType.set(type, (set = new Set())); + set.add(listener); return this; } function off(this: any, type: string, listener: any) { self.removeEventListener(type, listener); + byType.get(type)?.delete(listener); return this; } + function listenerCount(type: string) { + return byType.get(type)?.size ?? 0; + } + function eventNames() { + const out: string[] = []; + for (const [k, v] of byType) if (v.size > 0) out.push(k); + return out; + } + function removeAllListeners(this: any, type?: string) { + const clear = (t: string) => { + const set = byType.get(t); + if (set) { + for (const fn of set) self.removeEventListener(t, fn); + byType.delete(t); + } + }; + if (arguments.length === 0) for (const t of [...byType.keys()]) clear(t); + else clear(type!); + return this; + } + function emit(type: string, arg?: any) { + const had = (byType.get(type)?.size ?? 0) > 0; + const ev = + type === "message" || type === "messageerror" + ? new MessageEvent(type, { data: arg }) + : new CustomEvent(type, { detail: arg }); + self.dispatchEvent(ev); + return had; + } + let maxListeners = 10; for (const [name, fn] of [ ["on", on], ["addListener", on], ["once", once], ["off", off], ["removeListener", off], + ["listenerCount", listenerCount], + ["eventNames", eventNames], + ["removeAllListeners", removeAllListeners], + ["emit", emit], + ["getMaxListeners", () => maxListeners], + ["setMaxListeners", n => ((maxListeners = n), fake)], ] as const) { Object.defineProperty(fake, name, { value: fn, enumerable: false, configurable: true, writable: true }); } diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index aad6f7cb546c..3262ad209be4 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -592,8 +592,11 @@ JSC_DEFINE_HOST_FUNCTION(jsMessagePortPrototypeFunction_removeAllListeners, (JSG static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_setMaxListenersBody(JSC::JSGlobalObject* lexicalGlobalObject, JSC::CallFrame* callFrame, typename IDLOperation::ClassParameter castedThis) { - UNUSED_PARAM(lexicalGlobalObject); - castedThis->wrapped().setNodeMaxListeners(callFrame->argument(0).toUInt32(lexicalGlobalObject)); + auto& vm = JSC::getVM(lexicalGlobalObject); + auto throwScope = DECLARE_THROW_SCOPE(vm); + auto n = callFrame->argument(0).toUInt32(lexicalGlobalObject); + RETURN_IF_EXCEPTION(throwScope, {}); + castedThis->wrapped().setNodeMaxListeners(n); return JSValue::encode(castedThis); } diff --git a/src/jsc/bindings/webcore/MessagePort.cpp b/src/jsc/bindings/webcore/MessagePort.cpp index a59c83b8c29b..7e35ff165afd 100644 --- a/src/jsc/bindings/webcore/MessagePort.cpp +++ b/src/jsc/bindings/webcore/MessagePort.cpp @@ -493,10 +493,17 @@ void MessagePort::updateListenerEventLoopRef() void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& eventType, OnDidChangeListenerKind kind) { + auto& port = static_cast(self); + // removeAllListeners() bypasses the removeEventListener override, so the + // m_has*EventListener flags are reconciled here for Clear. + if (eventType == eventNames().closeEvent) { + if (kind == Clear) + port.m_hasCloseEventListener.store(false, std::memory_order_release); + return; + } if (eventType != eventNames().messageEvent) return; - auto& port = static_cast(self); switch (kind) { case Add: port.m_messageEventCount++; @@ -507,6 +514,7 @@ void MessagePort::onDidChangeListenerImpl(EventTarget& self, const AtomString& e break; case Clear: port.m_messageEventCount = 0; + port.m_hasMessageEventListener = false; break; } port.updateListenerEventLoopRef(); diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts index c22daed38b08..fd22af4adc59 100644 --- a/test/js/node/worker_threads/message-port-node-event-target.test.ts +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -401,4 +401,71 @@ describe.concurrent("MessagePort NodeEventTarget", () => { expect(stdout).toBe("0"); expect(exitCode).toBe(0); }); + + test("removeAllListeners('message') re-buffers instead of dropping", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + require("node:worker_threads"); + const { port1, port2 } = new MessageChannel(); + port1.on("message", () => { throw new Error("should not fire") }); + port1.removeAllListeners("message"); + port2.postMessage("x"); + setImmediate(() => { + let got; + port1.on("message", v => { got = v }); + setImmediate(() => { + process.stdout.write(String(got)); + port1.close(); + }); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("x"); + expect(exitCode).toBe(0); + }); + + // Spawns a worker inside a debug subprocess; give it more than the default 5s. + test("parentPort.listenerCount / eventNames / removeAllListeners work in a worker", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", + ` + const { Worker } = require("node:worker_threads"); + const w = new Worker( + "const { parentPort } = require('node:worker_threads');" + + "const fn = () => {};" + + "parentPort.on('message', fn);" + + "const lc = parentPort.listenerCount('message');" + + "const en = parentPort.eventNames().includes('message');" + + "parentPort.removeAllListeners('message');" + + "const after = parentPort.listenerCount('message');" + + "const ml = parentPort.getMaxListeners();" + + "parentPort.postMessage([lc, en, after, ml]);", + { eval: true } + ); + w.on("message", m => { + process.stdout.write(JSON.stringify(m)); + w.terminate(); + }); + `, + ], + env: bunEnv, + stdout: "pipe", + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toBe("[1,true,0,10]"); + expect(exitCode).toBe(0); + }, 30_000); }); From 892d7ca5e3ddd52f5df43356424987838566703c Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 22:36:19 +0000 Subject: [PATCH 05/13] test: run the coherence matrix in-process Only the no-import row and the parentPort row need subprocess isolation; the rest now run directly against the debug build, which avoids 15 concurrent bun-debug spawns exceeding the default 5s timeout under ASAN. --- .../message-port-node-event-target.test.ts | 542 ++++++------------ 1 file changed, 172 insertions(+), 370 deletions(-) diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts index fd22af4adc59..1e7d96c507a0 100644 --- a/test/js/node/worker_threads/message-port-node-event-target.test.ts +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -1,18 +1,19 @@ -import { describe, expect, test } from "bun:test"; +import { test, expect, describe } from "bun:test"; import { bunEnv, bunExe } from "harness"; +import { once, getEventListeners } from "node:events"; + +// Load so that a build without the native NodeEventTarget surface falls back +// to the JS shim; with it present the require is a no-op for the prototype. +require("node:worker_threads"); // MessagePort is a NodeEventTarget in node: .on/.addListener/.once share the // same listener list as addEventListener, so listenerCount/eventNames/ // removeAllListeners/getEventListeners see both, cross-remove works, and // emit() returns a boolean while passing the raw argument by identity. -// -// Each row of the coherence matrix runs in its own subprocess so that a throw -// in one row cannot mask another, and so row 15 can observe the surface -// without any prior import of node:worker_threads. -describe.concurrent("MessagePort NodeEventTarget", () => { - // Row 15 first so nothing else in this file can accidentally load - // node:worker_threads into a shared state that would mask the bug. +describe("MessagePort NodeEventTarget", () => { test("surface exists without importing node:worker_threads", async () => { + // Only this row needs a fresh process so nothing in this file can + // retroactively install the surface by importing worker_threads. await using proc = Bun.spawn({ cmd: [ bunExe(), @@ -39,401 +40,202 @@ describe.concurrent("MessagePort NodeEventTarget", () => { }); test("same fn via .on + addEventListener is one listener, invoked once", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1, port2 } = new MessageChannel(); - let n = 0; - const fn = () => { n++ }; - port1.on("message", fn); - port1.addEventListener("message", fn); - port2.postMessage("x"); - setImmediate(() => { - port1.close(); - process.stdout.write(String(n)); - }); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("1"); - expect(exitCode).toBe(0); + const { port1, port2 } = new MessageChannel(); + try { + let n = 0; + const fn = () => n++; + port1.on("message", fn); + port1.addEventListener("message", fn); + port2.postMessage("x"); + await once(port1, "message"); + expect(n).toBe(1); + } finally { + port1.close(); + } }); - test("listenerCount counts both .on and addEventListener listeners", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - port1.on("message", () => {}); - port1.addEventListener("message", () => {}); - process.stdout.write(String(port1.listenerCount("message"))); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("2"); - expect(exitCode).toBe(0); + test("listenerCount counts both .on and addEventListener listeners", () => { + const { port1 } = new MessageChannel(); + try { + port1.on("message", () => {}); + port1.addEventListener("message", () => {}); + expect(port1.listenerCount("message")).toBe(2); + } finally { + port1.close(); + } }); - test("eventNames reports types registered via addEventListener", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - port1.addEventListener("foo", () => {}); - process.stdout.write(JSON.stringify(port1.eventNames())); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe('["foo"]'); - expect(exitCode).toBe(0); + test("eventNames reports types registered via addEventListener", () => { + const { port1 } = new MessageChannel(); + try { + port1.addEventListener("foo", () => {}); + expect(port1.eventNames()).toEqual(["foo"]); + } finally { + port1.close(); + } }); - test("removeEventListener removes a listener added via .on", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - const fn = () => {}; - port1.on("message", fn); - port1.removeEventListener("message", fn); - process.stdout.write(String(port1.listenerCount("message"))); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("0"); - expect(exitCode).toBe(0); + test("removeEventListener removes a listener added via .on", () => { + const { port1 } = new MessageChannel(); + try { + const fn = () => {}; + port1.on("message", fn); + port1.removeEventListener("message", fn); + expect(port1.listenerCount("message")).toBe(0); + } finally { + port1.close(); + } }); test("removeAllListeners clears addEventListener listeners too", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1, port2 } = new MessageChannel(); - let n = 0; - port1.addEventListener("message", () => { n++ }); - port1.on("message", () => { n++ }); - port1.removeAllListeners("message"); - port1.start(); - port2.postMessage("x"); - setImmediate(() => { - port1.close(); - process.stdout.write(String(n)); - }); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("0"); - expect(exitCode).toBe(0); + const { port1, port2 } = new MessageChannel(); + try { + let n = 0; + port1.addEventListener("message", () => n++); + port1.on("message", () => n++); + port1.removeAllListeners("message"); + port1.start(); + port2.postMessage("x"); + await new Promise(r => setImmediate(() => r())); + expect(n).toBe(0); + } finally { + port1.close(); + } }); - test("emit passes the argument by identity to .on listeners", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - const payload = { a: 1 }; - let got; - port1.on("message", v => { got = v }); - port1.emit("message", payload); - process.stdout.write(String(got === payload)); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("true"); - expect(exitCode).toBe(0); + test("emit passes the argument by identity to .on listeners", () => { + const { port1 } = new MessageChannel(); + try { + const payload = { a: 1 }; + let got; + port1.on("message", v => (got = v)); + port1.emit("message", payload); + expect(got).toBe(payload); + } finally { + port1.close(); + } }); - test("emit returns a boolean (true when listeners, false otherwise)", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - const r1 = port1.emit("nope", 1); - port1.on("hit", () => {}); - const r2 = port1.emit("hit", 1); - process.stdout.write(JSON.stringify([r1, r2])); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("[false,true]"); - expect(exitCode).toBe(0); + test("emit returns a boolean (true when listeners, false otherwise)", () => { + const { port1 } = new MessageChannel(); + try { + expect(port1.emit("nope", 1)).toBe(false); + port1.on("hit", () => {}); + expect(port1.emit("hit", 1)).toBe(true); + } finally { + port1.close(); + } }); - test("emit with a primitive payload does not throw", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - let got; - port1.on("message", v => { got = v }); - port1.emit("message", 1); - process.stdout.write(String(got)); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("1"); - expect(exitCode).toBe(0); + test("emit with a primitive payload does not throw", () => { + const { port1 } = new MessageChannel(); + try { + let got; + port1.on("message", v => (got = v)); + port1.emit("message", 1); + expect(got).toBe(1); + } finally { + port1.close(); + } }); - test("emit('error', err) passes the error by identity", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - const err = new Error("boom"); - let got; - port1.on("error", e => { got = e }); - port1.emit("error", err); - process.stdout.write(String(got === err)); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("true"); - expect(exitCode).toBe(0); + test("emit('error', err) passes the error by identity", () => { + const { port1 } = new MessageChannel(); + try { + const err = new Error("boom"); + let got; + port1.on("error", e => (got = e)); + port1.emit("error", err); + expect(got).toBe(err); + } finally { + port1.close(); + } }); - test(".once dedupes against .on for the same fn", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - const fn = () => {}; - port1.on("foo", fn); - port1.once("foo", fn); - process.stdout.write(String(port1.listenerCount("foo"))); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("1"); - expect(exitCode).toBe(0); + test(".once dedupes against .on for the same fn", () => { + const { port1 } = new MessageChannel(); + try { + const fn = () => {}; + port1.on("foo", fn); + port1.once("foo", fn); + expect(port1.listenerCount("foo")).toBe(1); + } finally { + port1.close(); + } }); - test("events.getEventListeners returns the user fn, not a wrapper", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { getEventListeners } = require("node:events"); - const { port1 } = new MessageChannel(); - const fn = () => {}; - port1.on("message", fn); - const list = getEventListeners(port1, "message"); - process.stdout.write(String(list.length === 1 && list[0] === fn)); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("true"); - expect(exitCode).toBe(0); + test("events.getEventListeners returns the user fn, not a wrapper", () => { + const { port1 } = new MessageChannel(); + try { + const fn = () => {}; + port1.on("message", fn); + const list = getEventListeners(port1, "message"); + expect(list.length).toBe(1); + expect(list[0]).toBe(fn); + } finally { + port1.close(); + } }); test("addEventListener listeners still receive an Event; .on listeners get the payload", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1, port2 } = new MessageChannel(); - let a, b; - port1.on("message", v => { a = v }); - port1.addEventListener("message", ev => { b = ev }); - port2.postMessage(42); - setImmediate(() => { - port1.close(); - process.stdout.write(JSON.stringify([a, b instanceof MessageEvent, b.data])); - }); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("[42,true,42]"); - expect(exitCode).toBe(0); + const { port1, port2 } = new MessageChannel(); + try { + let a, b; + port1.on("message", v => (a = v)); + port1.addEventListener("message", ev => (b = ev)); + port2.postMessage(42); + await once(port1, "message"); + expect(a).toBe(42); + expect(b).toBeInstanceOf(MessageEvent); + expect((b as MessageEvent).data).toBe(42); + } finally { + port1.close(); + } }); - test("emit with an addEventListener listener delivers an Event with matching data/detail", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - let msg, custom; - port1.addEventListener("message", ev => { msg = ev.data }); - port1.addEventListener("foo", ev => { custom = ev.detail }); - port1.emit("message", 7); - port1.emit("foo", 8); - process.stdout.write(JSON.stringify([msg, custom])); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("[7,8]"); - expect(exitCode).toBe(0); + test("emit with an addEventListener listener delivers an Event with matching data/detail", () => { + const { port1 } = new MessageChannel(); + try { + let msg, custom; + port1.addEventListener("message", ev => (msg = (ev as MessageEvent).data)); + port1.addEventListener("foo", ev => (custom = (ev as CustomEvent).detail)); + port1.emit("message", 7); + port1.emit("foo", 8); + expect(msg).toBe(7); + expect(custom).toBe(8); + } finally { + port1.close(); + } }); - test(".off removes a listener added via addEventListener", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1 } = new MessageChannel(); - const fn = () => {}; - port1.addEventListener("message", fn); - port1.off("message", fn); - process.stdout.write(String(port1.listenerCount("message"))); - port1.close(); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("0"); - expect(exitCode).toBe(0); + test(".off removes a listener added via addEventListener", () => { + const { port1 } = new MessageChannel(); + try { + const fn = () => {}; + port1.addEventListener("message", fn); + port1.off("message", fn); + expect(port1.listenerCount("message")).toBe(0); + } finally { + port1.close(); + } }); test("removeAllListeners('message') re-buffers instead of dropping", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - require("node:worker_threads"); - const { port1, port2 } = new MessageChannel(); - port1.on("message", () => { throw new Error("should not fire") }); - port1.removeAllListeners("message"); - port2.postMessage("x"); - setImmediate(() => { - let got; - port1.on("message", v => { got = v }); - setImmediate(() => { - process.stdout.write(String(got)); - port1.close(); - }); - }); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("x"); - expect(exitCode).toBe(0); + const { port1, port2 } = new MessageChannel(); + try { + port1.on("message", () => { + throw new Error("should not fire"); + }); + port1.removeAllListeners("message"); + port2.postMessage("x"); + await new Promise(r => setImmediate(() => r())); + const got = await once(port1, "message"); + expect(got[0]).toBe("x"); + } finally { + port1.close(); + } }); - // Spawns a worker inside a debug subprocess; give it more than the default 5s. test("parentPort.listenerCount / eventNames / removeAllListeners work in a worker", async () => { await using proc = Bun.spawn({ cmd: [ From b88a5818ea68b6bf61dcc05bddda6a4b3376e4d8 Mon Sep 17 00:00:00 2001 From: "autofix-ci[bot]" <114827586+autofix-ci[bot]@users.noreply.github.com> Date: Sat, 25 Jul 2026 22:38:10 +0000 Subject: [PATCH 06/13] [autofix.ci] apply automated fixes --- .../worker_threads/message-port-node-event-target.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts index 1e7d96c507a0..0825a3f4cac8 100644 --- a/test/js/node/worker_threads/message-port-node-event-target.test.ts +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -1,6 +1,6 @@ -import { test, expect, describe } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe } from "harness"; -import { once, getEventListeners } from "node:events"; +import { getEventListeners, once } from "node:events"; // Load so that a build without the native NodeEventTarget surface falls back // to the JS shim; with it present the require is a no-op for the prototype. From d2869b4e38aa775038340391148977a60cc9f4a5 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 22:44:16 +0000 Subject: [PATCH 07/13] lint + review nits: Array.from over spread, once() purges byType, name error for once - oxlint: Array.from(byType.keys()) instead of spread. - fakeParentPort.once(): register a wrapper that deletes the listener from byType when it fires so listenerCount/eventNames/emit are accurate afterwards; off() looks up the wrapper. - messagePortNodeAddListener: name the function in the listener-arg TypeError ('once' vs 'on'). --- src/js/node/worker_threads.ts | 28 +++++++++++++--------- src/jsc/bindings/webcore/JSMessagePort.cpp | 2 +- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index dcaf8ce5daf3..fee637dd1915 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -638,23 +638,29 @@ function fakeParentPort() { // eventNames/removeAllListeners see only those (self may carry internal // listeners the user must not touch). const byType = new SafeMap(); + function track(type: string, listener: any, registered: any) { + let set = byType.get(type); + if (!set) byType.set(type, (set = new SafeMap())); + set.set(listener, registered); + } function on(this: any, type: string, listener: any) { self.addEventListener(type, listener, { $kIsNodeStyleListener: true } as AddEventListenerOptions); - let set = byType.get(type); - if (!set) byType.set(type, (set = new Set())); - set.add(listener); + track(type, listener, listener); return this; } function once(this: any, type: string, listener: any) { - self.addEventListener(type, listener, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); - let set = byType.get(type); - if (!set) byType.set(type, (set = new Set())); - set.add(listener); + const wrapper = (arg: any) => { + byType.get(type)?.delete(listener); + listener(arg); + }; + self.addEventListener(type, wrapper, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); + track(type, listener, wrapper); return this; } function off(this: any, type: string, listener: any) { - self.removeEventListener(type, listener); - byType.get(type)?.delete(listener); + const set = byType.get(type); + self.removeEventListener(type, set?.get(listener) ?? listener); + set?.delete(listener); return this; } function listenerCount(type: string) { @@ -669,11 +675,11 @@ function fakeParentPort() { const clear = (t: string) => { const set = byType.get(t); if (set) { - for (const fn of set) self.removeEventListener(t, fn); + for (const registered of set.values()) self.removeEventListener(t, registered); byType.delete(t); } }; - if (arguments.length === 0) for (const t of [...byType.keys()]) clear(t); + if (arguments.length === 0) for (const t of Array.from(byType.keys())) clear(t); else clear(type!); return this; } diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index 3262ad209be4..dd15a1fddf46 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -451,7 +451,7 @@ static inline JSC::EncodedJSValue messagePortNodeAddListener(JSC::JSGlobalObject auto type = convert>(*lexicalGlobalObject, argument0.value()); RETURN_IF_EXCEPTION(throwScope, {}); EnsureStillAliveScope argument1 = callFrame->argument(1); - auto listener = convert>>(*lexicalGlobalObject, argument1.value(), *castedThis, [](JSC::JSGlobalObject& lexicalGlobalObject, JSC::ThrowScope& scope) { throwArgumentMustBeObjectError(lexicalGlobalObject, scope, 1, "listener"_s, "MessagePort"_s, "on"_s); }); + auto listener = convert>>(*lexicalGlobalObject, argument1.value(), *castedThis, [once](JSC::JSGlobalObject& lexicalGlobalObject, JSC::ThrowScope& scope) { throwArgumentMustBeObjectError(lexicalGlobalObject, scope, 1, "listener"_s, "MessagePort"_s, once ? "once"_s : "on"_s); }); RETURN_IF_EXCEPTION(throwScope, {}); if (listener) { AddEventListenerOptions options; From bc2a883c3c973101d8b4ff714bdd3a96156fa5c4 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 23:27:16 +0000 Subject: [PATCH 08/13] fakeParentPort: first-registration-wins dedup in on()/once() track() now returns false when the listener is already tracked and the caller skips self.addEventListener, so a second once()/on() for the same function is a no-op (matching node) and off() can't orphan an earlier registration. --- src/js/node/worker_threads.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index fee637dd1915..4409819ed1b9 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -641,11 +641,15 @@ function fakeParentPort() { function track(type: string, listener: any, registered: any) { let set = byType.get(type); if (!set) byType.set(type, (set = new SafeMap())); + // First registration of a given listener wins (including its once-ness), + // matching node's NodeEventTarget dedup. + if (set.has(listener)) return false; set.set(listener, registered); + return true; } function on(this: any, type: string, listener: any) { - self.addEventListener(type, listener, { $kIsNodeStyleListener: true } as AddEventListenerOptions); - track(type, listener, listener); + if (track(type, listener, listener)) + self.addEventListener(type, listener, { $kIsNodeStyleListener: true } as AddEventListenerOptions); return this; } function once(this: any, type: string, listener: any) { @@ -653,8 +657,8 @@ function fakeParentPort() { byType.get(type)?.delete(listener); listener(arg); }; - self.addEventListener(type, wrapper, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); - track(type, listener, wrapper); + if (track(type, listener, wrapper)) + self.addEventListener(type, wrapper, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); return this; } function off(this: any, type: string, listener: any) { From d0ad3a96bcb5d18b0f3900a868b1dd6bb89e948f Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sat, 25 Jul 2026 23:55:19 +0000 Subject: [PATCH 09/13] test: spawn the worker in-process for the parentPort row Avoids the subprocess + worker double-spawn that sat at the 5s default timeout boundary under debug+ASAN. --- .../message-port-node-event-target.test.ts | 51 ++++++------------- 1 file changed, 16 insertions(+), 35 deletions(-) diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts index 0825a3f4cac8..4b54590251ea 100644 --- a/test/js/node/worker_threads/message-port-node-event-target.test.ts +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -1,10 +1,9 @@ import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe } from "harness"; import { getEventListeners, once } from "node:events"; - // Load so that a build without the native NodeEventTarget surface falls back // to the JS shim; with it present the require is a no-op for the prototype. -require("node:worker_threads"); +import { Worker } from "node:worker_threads"; // MessagePort is a NodeEventTarget in node: .on/.addListener/.once share the // same listener list as addEventListener, so listenerCount/eventNames/ @@ -237,37 +236,19 @@ describe("MessagePort NodeEventTarget", () => { }); test("parentPort.listenerCount / eventNames / removeAllListeners work in a worker", async () => { - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - ` - const { Worker } = require("node:worker_threads"); - const w = new Worker( - "const { parentPort } = require('node:worker_threads');" + - "const fn = () => {};" + - "parentPort.on('message', fn);" + - "const lc = parentPort.listenerCount('message');" + - "const en = parentPort.eventNames().includes('message');" + - "parentPort.removeAllListeners('message');" + - "const after = parentPort.listenerCount('message');" + - "const ml = parentPort.getMaxListeners();" + - "parentPort.postMessage([lc, en, after, ml]);", - { eval: true } - ); - w.on("message", m => { - process.stdout.write(JSON.stringify(m)); - w.terminate(); - }); - `, - ], - env: bunEnv, - stdout: "pipe", - stderr: "pipe", - }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - expect(stdout).toBe("[1,true,0,10]"); - expect(exitCode).toBe(0); - }, 30_000); + await using w = new Worker( + "const { parentPort } = require('node:worker_threads');" + + "const fn = () => {};" + + "parentPort.on('message', fn);" + + "const lc = parentPort.listenerCount('message');" + + "const en = parentPort.eventNames().includes('message');" + + "parentPort.removeAllListeners('message');" + + "const after = parentPort.listenerCount('message');" + + "const ml = parentPort.getMaxListeners();" + + "parentPort.postMessage([lc, en, after, ml]);", + { eval: true }, + ); + const [got] = await once(w, "message"); + expect(got).toEqual([1, true, 0, 10]); + }); }); From 2687354164749ef420548f373d1e313e4695b18a Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 00:29:03 +0000 Subject: [PATCH 10/13] review nits: propagate once() return, iterate keys() directly, drop stale comment - once() wrapper returns listener(arg) so a promise return reaches handleEvent's addCatch path. - removeAllListeners iterates byType.keys() directly; SafeMap's iterator tolerates deleting the just-yielded key so no snapshot is needed. - Drop the test-file comment that referenced the deleted JS shim. --- src/js/node/worker_threads.ts | 4 ++-- .../worker_threads/message-port-node-event-target.test.ts | 2 -- 2 files changed, 2 insertions(+), 4 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 4409819ed1b9..dd9628bf2ba3 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -655,7 +655,7 @@ function fakeParentPort() { function once(this: any, type: string, listener: any) { const wrapper = (arg: any) => { byType.get(type)?.delete(listener); - listener(arg); + return listener(arg); }; if (track(type, listener, wrapper)) self.addEventListener(type, wrapper, { once: true, $kIsNodeStyleListener: true } as AddEventListenerOptions); @@ -683,7 +683,7 @@ function fakeParentPort() { byType.delete(t); } }; - if (arguments.length === 0) for (const t of Array.from(byType.keys())) clear(t); + if (arguments.length === 0) for (const t of byType.keys()) clear(t); else clear(type!); return this; } diff --git a/test/js/node/worker_threads/message-port-node-event-target.test.ts b/test/js/node/worker_threads/message-port-node-event-target.test.ts index 4b54590251ea..c5f0fdcd9948 100644 --- a/test/js/node/worker_threads/message-port-node-event-target.test.ts +++ b/test/js/node/worker_threads/message-port-node-event-target.test.ts @@ -1,8 +1,6 @@ import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe } from "harness"; import { getEventListeners, once } from "node:events"; -// Load so that a build without the native NodeEventTarget surface falls back -// to the JS shim; with it present the require is a no-op for the prototype. import { Worker } from "node:worker_threads"; // MessagePort is a NodeEventTarget in node: .on/.addListener/.once share the From 46834adc1b122afd0c18c83dc9cdf9c49469d696 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 01:05:18 +0000 Subject: [PATCH 11/13] fakeParentPort: removeEventListener also purges byType So .on(fn) -> removeEventListener(fn) -> .on(fn) re-registers instead of being swallowed by track()'s first-wins check. --- src/js/node/worker_threads.ts | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index dd9628bf2ba3..4a739229ef58 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -628,16 +628,18 @@ function fakeParentPort() { value: self.addEventListener.bind(self), }); - Object.defineProperty(fake, "removeEventListener", { - value: self.removeEventListener.bind(self), - }); - // MessagePort.prototype.on/etc. require a real MessagePort receiver; shadow // them so calls on this stand-in forward to the global scope. A local map // tracks which listeners were added via parentPort so listenerCount/ // eventNames/removeAllListeners see only those (self may carry internal // listeners the user must not touch). const byType = new SafeMap(); + Object.defineProperty(fake, "removeEventListener", { + value(type: string, listener: any) { + self.removeEventListener(type, listener); + byType.get(type)?.delete(listener); + }, + }); function track(type: string, listener: any, registered: any) { let set = byType.get(type); if (!set) byType.set(type, (set = new SafeMap())); From d2147a7de96effc2dd8c50cd6c12082de42ceafa Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 01:37:42 +0000 Subject: [PATCH 12/13] removeAllListeners(undefined|null) clears all; fakeParentPort.removeEventListener forwards options and wrapper - removeAllListeners treats an undefined/null argument the same as no argument (node checks type !== undefined && type !== null). - fakeParentPort.removeEventListener looks up the once-wrapper via byType and forwards the options/useCapture argument, mirroring off(). --- src/js/node/worker_threads.ts | 7 ++++--- src/jsc/bindings/webcore/JSMessagePort.cpp | 4 ++-- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/src/js/node/worker_threads.ts b/src/js/node/worker_threads.ts index 4a739229ef58..367ef91a0c9b 100644 --- a/src/js/node/worker_threads.ts +++ b/src/js/node/worker_threads.ts @@ -635,9 +635,10 @@ function fakeParentPort() { // listeners the user must not touch). const byType = new SafeMap(); Object.defineProperty(fake, "removeEventListener", { - value(type: string, listener: any) { - self.removeEventListener(type, listener); - byType.get(type)?.delete(listener); + value(type: string, listener: any, options?: any) { + const set = byType.get(type); + self.removeEventListener(type, set?.get(listener) ?? listener, options); + set?.delete(listener); }, }); function track(type: string, listener: any, registered: any) { diff --git a/src/jsc/bindings/webcore/JSMessagePort.cpp b/src/jsc/bindings/webcore/JSMessagePort.cpp index dd15a1fddf46..94f16ef892ea 100644 --- a/src/jsc/bindings/webcore/JSMessagePort.cpp +++ b/src/jsc/bindings/webcore/JSMessagePort.cpp @@ -574,10 +574,10 @@ static inline JSC::EncodedJSValue jsMessagePortPrototypeFunction_removeAllListen auto& vm = JSC::getVM(lexicalGlobalObject); auto throwScope = DECLARE_THROW_SCOPE(vm); auto& impl = castedThis->wrapped(); - if (callFrame->argumentCount() == 0) { + EnsureStillAliveScope argument0 = callFrame->argument(0); + if (argument0.value().isUndefinedOrNull()) { impl.removeAllEventListeners(); } else { - EnsureStillAliveScope argument0 = callFrame->uncheckedArgument(0); auto type = convert>(*lexicalGlobalObject, argument0.value()); RETURN_IF_EXCEPTION(throwScope, {}); impl.removeAllEventListenersForType(type); From a3675201618c0ee01fece7e8af25bfeb5db4c852 Mon Sep 17 00:00:00 2001 From: robobun <117481402+robobun@users.noreply.github.com> Date: Sun, 26 Jul 2026 02:41:15 +0000 Subject: [PATCH 13/13] ci: retrigger