fix(node:events): fix abort behavior

3 files changed, 177 insertions, 170 deletions

diff --git a/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp b/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp
index c6f2b3895..dca0ee431 100644
--- a/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp
+++ b/src/bun.js/builtins/cpp/NodeEventsBuiltins.cpp
@@ -51,13 +51,13 @@ namespace WebCore {
 const JSC::ConstructAbility s_nodeEventsOnAsyncIteratorCodeConstructAbility = JSC::ConstructAbility::CannotConstruct;
 const JSC::ConstructorKind s_nodeEventsOnAsyncIteratorCodeConstructorKind = JSC::ConstructorKind::None;
 const JSC::ImplementationVisibility s_nodeEventsOnAsyncIteratorCodeImplementationVisibility = JSC::ImplementationVisibility::Public;
-const int s_nodeEventsOnAsyncIteratorCodeLength = 4473;
+const int s_nodeEventsOnAsyncIteratorCodeLength = 4455;
 static const JSC::Intrinsic s_nodeEventsOnAsyncIteratorCodeIntrinsic = JSC::NoIntrinsic;
 const char* const s_nodeEventsOnAsyncIteratorCode =
     "(function (emitter, event, options) {\n" \
     "  \"use strict\";\n" \
     "\n" \
-    "  var { AbortSignal, Number, Error } = globalThis;\n" \
+    "  var { AbortSignal, Symbol, Number, Error } = globalThis;\n" \
     "\n" \
     "  var AbortError = class AbortError extends Error {\n" \
     "    constructor(message = \"The operation was aborted\", options = void 0) {\n" \
@@ -74,14 +74,14 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "\n" \
     "  if (@isUndefinedOrNull(emitter)) @throwTypeError(\"emitter is required\");\n" \
     "  //\n" \
-    "  if (!(typeof emitter === \"object\" && @isCallable(emitter.emit) && @isCallable(emitter.on)))\n" \
+    "  if (!(@isObject(emitter) && @isCallable(emitter.emit) && @isCallable(emitter.on)))\n" \
     "    @throwTypeError(\"emitter must be an EventEmitter\");\n" \
     "\n" \
     "  if (@isUndefinedOrNull(options)) options = {};\n" \
     "\n" \
     "  //\n" \
     "  var signal = options.signal;\n" \
-    "  if (!@isUndefinedOrNull(signal) && !(signal instanceof AbortSignal))\n" \
+    "  if (signal !== undefined && (!@isObject(signal) || !(signal instanceof AbortSignal)))\n" \
     "    @throwTypeError(\"options.signal must be an AbortSignal\");\n" \
     "\n" \
     "  if (signal?.aborted) {\n" \
@@ -122,7 +122,7 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "  }\n" \
     "\n" \
     "  function closeHandler() {\n" \
-    "    removeAllListeners(listeners);\n" \
+    "    removeAllListeners();\n" \
     "    finished = true;\n" \
     "    while (!unconsumedPromises.isEmpty()) {\n" \
     "      const promise = unconsumedPromises.shift();\n" \
@@ -199,19 +199,15 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "    }\n" \
     "  }\n" \
     "\n" \
-    "  if (signal)\n" \
-    "    signal.once(\"abort\", abortListener);\n" \
+    "  if (signal) signal.addEventListener(\"abort\", abortListener, { once: true });\n" \
     "\n" \
     "  var iterator = createIterator();\n" \
-    "\n" \
     "  @Object.defineProperties(iterator, {\n" \
     "    return: {\n" \
-    "      value: function() {\n" \
-    "        return closeHandler();\n" \
-    "      },\n" \
+    "      value: () => closeHandler(),\n" \
     "    },\n" \
     "    throw: {\n" \
-    "      value: function(err) {\n" \
+    "      value: (err) => {\n" \
     "        if (!err || !(err instanceof Error)) {\n" \
     "          throw new TypeError(\"EventEmitter.AsyncIterator must be called with an error\");\n" \
     "        }\n" \
@@ -219,7 +215,7 @@ const char* const s_nodeEventsOnAsyncIteratorCode =
     "      },\n" \
     "    },\n" \
     "    [Symbol.asyncIterator]: {\n" \
-    "      value: function() { return this; }\n" \
+    "      value: () => iterator,\n" \
     "    },\n" \
     "  });\n" \
     "  return iterator;\n" \
diff --git a/src/bun.js/builtins/js/NodeEvents.js b/src/bun.js/builtins/js/NodeEvents.js
index 6c80f81c8..ded8d1af7 100644
--- a/src/bun.js/builtins/js/NodeEvents.js
+++ b/src/bun.js/builtins/js/NodeEvents.js
@@ -26,7 +26,7 @@
 function onAsyncIterator(emitter, event, options) {
   "use strict";
 
-  var { AbortSignal, Number, Error } = globalThis;
+  var { AbortSignal, Symbol, Number, Error } = globalThis;
 
   var AbortError = class AbortError extends Error {
     constructor(message = "The operation was aborted", options = void 0) {
@@ -41,14 +41,14 @@ function onAsyncIterator(emitter, event, options) {
 
   if (@isUndefinedOrNull(emitter)) @throwTypeError("emitter is required");
   // TODO: Do a more accurate check
-  if (!(typeof emitter === "object" && @isCallable(emitter.emit) && @isCallable(emitter.on)))
+  if (!(@isObject(emitter) && @isCallable(emitter.emit) && @isCallable(emitter.on)))
     @throwTypeError("emitter must be an EventEmitter");
 
   if (@isUndefinedOrNull(options)) options = {};
 
   // Parameters validation
   var signal = options.signal;
-  if (!@isUndefinedOrNull(signal) && !(signal instanceof AbortSignal))
+  if (signal !== undefined && (!@isObject(signal) || !(signal instanceof AbortSignal)))
     @throwTypeError("options.signal must be an AbortSignal");
 
   if (signal?.aborted) {
@@ -89,7 +89,7 @@ function onAsyncIterator(emitter, event, options) {
   }
 
   function closeHandler() {
-    removeAllListeners(listeners);
+    removeAllListeners();
     finished = true;
     while (!unconsumedPromises.isEmpty()) {
       const promise = unconsumedPromises.shift();
@@ -166,19 +166,15 @@ function onAsyncIterator(emitter, event, options) {
     }
   }
 
-  if (signal)
-    signal.once("abort", abortListener);
+  if (signal) signal.addEventListener("abort", abortListener, { once: true });
 
   var iterator = createIterator();
-
   @Object.defineProperties(iterator, {
     return: {
-      value: function() {
-        return closeHandler();
-      },
+      value: () => closeHandler(),
     },
     throw: {
-      value: function(err) {
+      value: (err) => {
         if (!err || !(err instanceof Error)) {
           throw new TypeError("EventEmitter.AsyncIterator must be called with an error");
         }
@@ -186,7 +182,7 @@ function onAsyncIterator(emitter, event, options) {
       },
     },
     [Symbol.asyncIterator]: {
-      value: function() { return this; }
+      value: () => iterator,
     },
   });
   return iterator;
diff --git a/test/js/node/events/node-events.node.test.ts b/test/js/node/events/node-events.node.test.ts
index 6a1fd5f2a..c086aa0b7 100644
--- a/test/js/node/events/node-events.node.test.ts
+++ b/test/js/node/events/node-events.node.test.ts
@@ -1,10 +1,10 @@
 import { EventEmitter, on } from "node:events";
 import { createTest } from "node-harness";
 
-const { expect, assert, describe, it, createCallCheckCtx, createDoneDotAll } = createTest(import.meta.path);
+const { beforeAll, expect, assert, describe, it, createCallCheckCtx, createDoneDotAll } = createTest(import.meta.path);
 // const NodeEventTarget = globalThis.EventTarget;
 
-describe("node:events.on (EE async iterator)", () => {
+describe("node:events.on (EventEmitter AsyncIterator)", () => {
   it("should return an async iterator", async () => {
     const ee = new EventEmitter();
     const iterable = on(ee, "foo");
@@ -192,106 +192,174 @@ describe("node:events.on (EE async iterator)", () => {
     done();
   });
 
-  it("should throw a `TypeError` when calling throw without args", async () => {
+  describe(".throw()", () => {
+    let ee: EventEmitter;
+    let iterable: AsyncIterableIterator<any>;
+
+    beforeAll(() => {
+      ee = new EventEmitter();
+      iterable = on(ee, "foo");
+    });
+
+    it("should throw a `TypeError` when calling without args", async () => {
+      expect(() => {
+        iterable.throw!();
+      }).toThrow(TypeError);
+
+      assert.strictEqual(ee.listenerCount("foo"), 1);
+      assert.strictEqual(ee.listenerCount("error"), 1);
+    });
+
+    it("should throw when called with an error", async () => {
+      const _err = new Error("kaboom");
+
+      ee.emit("foo", "bar");
+      ee.emit("foo", 42);
+      iterable.throw!(_err);
+
+      const expected = [["bar"], [42]];
+
+      let thrown = false;
+      let looped = false;
+
+      try {
+        for await (const event of iterable) {
+          assert.deepStrictEqual(event, expected.shift());
+          looped = true;
+        }
+      } catch (err) {
+        thrown = true;
+        assert.strictEqual(err, _err);
+      }
+
+      assert.strictEqual(looped, true);
+      assert.strictEqual(thrown, true);
+      assert.strictEqual(ee.listenerCount("foo"), 0);
+      assert.strictEqual(ee.listenerCount("error"), 0);
+    });
+  });
+
+  it("should add an error listener when the iterable is created", () => {
     const ee = new EventEmitter();
-    const iterable = on(ee, "foo");
+    on(ee, "foo");
+    assert.strictEqual(ee.listenerCount("error"), 1);
+  });
 
-    expect(() => {
-      // @ts-ignore
-      iterable.throw();
-    }).toThrow(TypeError);
+  it("should throw when called with an aborted signal", () => {
+    const ee = new EventEmitter();
+    const abortedSignal = AbortSignal.abort();
+    [1, {}, null, false, "hi"].forEach((signal: any) => {
+      assert.throws(() => on(ee, "foo", { signal }), Error);
+    });
+    assert.throws(() => on(ee, "foo", { signal: abortedSignal }), {
+      name: "AbortError",
+    });
   });
 
-  // async function iterableThrow() {
-  //   const ee = new EventEmitter();
-  //   const iterable = on(ee, "foo");
+  it("should NOT THROW an `AbortError` AFTER done iterating over events", async _done => {
+    let _doneCalled = false;
+    const done = (err?: Error) => {
+      if (_doneCalled) return;
+      _doneCalled = true;
+      _done(err);
+    };
 
-  //   process.nextTick(() => {
-  //     ee.emit("foo", "bar");
-  //     ee.emit("foo", 42); // lost in the queue
-  //     iterable.throw(_err);
-  //   });
+    const ee = new EventEmitter();
+    const ac = new AbortController();
+
+    const i = setInterval(() => ee.emit("foo", "foo"), 1);
+    let count = 0;
+
+    async function foo() {
+      for await (const f of on(ee, "foo", { signal: ac.signal })) {
+        assert.strictEqual(f[0], "foo");
+        if (++count === 5) break;
+      }
+      ac.abort(); // No error will occur
+    }
+
+    foo()
+      .catch(err => done(err))
+      .finally(() => {
+        clearInterval(i);
+        if (!_doneCalled) expect(true).toBe(true);
+        done();
+      });
+  });
+
+  it("should THROW an `AbortError` BEFORE done iterating over events", async _done => {
+    let _doneCalled = false;
+    const done = (err?: Error) => {
+      if (_doneCalled) return;
+      _doneCalled = true;
+      _done(err);
+    };
+
+    let count = 0;
+
+    const createDone = createDoneDotAll(done);
+    const { mustCall, closeTimers } = createCallCheckCtx(createDone());
+    const finalDone = createDone();
+
+    const ee = new EventEmitter();
+    const ac = new AbortController();
+
+    const i = setInterval(() => ee.emit("foo", "foo"), 10);
+
+    setTimeout(() => ac.abort(), 50);
+
+    async function foo() {
+      for await (const f of on(ee, "foo", { signal: ac.signal })) {
+        assert.deepStrictEqual(f, ["foo"]);
+      }
+    }
+
+    foo()
+      .then(() => done(new Error("Should not be called")))
+      .catch(
+        mustCall(error => {
+          assert.strictEqual(error.name, "AbortError");
+        }),
+      )
+      .finally(() => {
+        clearInterval(i);
+        closeTimers();
+        if (!_doneCalled) finalDone();
+      });
+  });
 
-  //   const _err = new Error("kaboom");
-  //   let thrown = false;
-
-  //   assert.throws(
-  //     () => {
-  //       // No argument
-  //       iterable.throw();
-  //     },
-  //     {
-  //       message: 'The "EventEmitter.AsyncIterator" property must be' + " an instance of Error. Received undefined",
-  //       name: "TypeError",
-  //     },
-  //   );
-
-  //   const expected = [["bar"], [42]];
-
-  //   try {
-  //     for await (const event of iterable) {
-  //       assert.deepStrictEqual(event, expected.shift());
+  // TODO: Uncomment tests for NodeEventTarget and Web EventTarget
+
+  // async function eventTarget() {
+  //   const et = new EventTarget();
+  //   const tick = () => et.dispatchEvent(new Event("tick"));
+  //   const interval = setInterval(tick, 0);
+  //   let count = 0;
+  //   for await (const [event] of on(et, "tick")) {
+  //     count++;
+  //     assert.strictEqual(event.type, "tick");
+  //     if (count >= 5) {
+  //       break;
   //     }
-  //   } catch (err) {
-  //     thrown = true;
-  //     assert.strictEqual(err, _err);
   //   }
-  //   assert.strictEqual(thrown, true);
-  //   assert.strictEqual(expected.length, 0);
-  //   assert.strictEqual(ee.listenerCount("foo"), 0);
-  //   assert.strictEqual(ee.listenerCount("error"), 0);
+  //   assert.strictEqual(count, 5);
+  //   clearInterval(interval);
   // }
 
-  // // async function eventTarget() {
-  // //   const et = new EventTarget();
-  // //   const tick = () => et.dispatchEvent(new Event("tick"));
-  // //   const interval = setInterval(tick, 0);
-  // //   let count = 0;
-  // //   for await (const [event] of on(et, "tick")) {
-  // //     count++;
-  // //     assert.strictEqual(event.type, "tick");
-  // //     if (count >= 5) {
-  // //       break;
-  // //     }
-  // //   }
-  // //   assert.strictEqual(count, 5);
-  // //   clearInterval(interval);
-  // // }
-
-  // async function errorListenerCount() {
-  //   const et = new EventEmitter();
-  //   on(et, "foo");
-  //   assert.strictEqual(et.listenerCount("error"), 1);
-  // }
-
-  // // async function nodeEventTarget() {
-  // //   const et = new NodeEventTarget();
-  // //   const tick = () => et.dispatchEvent(new Event("tick"));
-  // //   const interval = setInterval(tick, 0);
-  // //   let count = 0;
-  // //   for await (const [event] of on(et, "tick")) {
-  // //     count++;
-  // //     assert.strictEqual(event.type, "tick");
-  // //     if (count >= 5) {
-  // //       break;
-  // //     }
-  // //   }
-  // //   assert.strictEqual(count, 5);
-  // //   clearInterval(interval);
-  // // }
-
-  // async function abortableOnBefore() {
-  //   it("should ");
-  //   const ee = new EventEmitter();
-  //   const abortedSignal = AbortSignal.abort();
-  //   [1, {}, null, false, "hi"].forEach((signal: any) => {
-  //     assert.throws(() => on(ee, "foo", { signal }), {
-  //       code: "ERR_INVALID_ARG_TYPE",
-  //     });
-  //   });
-  //   assert.throws(() => on(ee, "foo", { signal: abortedSignal }), {
-  //     name: "AbortError",
-  //   });
+  // async function nodeEventTarget() {
+  //   const et = new NodeEventTarget();
+  //   const tick = () => et.dispatchEvent(new Event("tick"));
+  //   const interval = setInterval(tick, 0);
+  //   let count = 0;
+  //   for await (const [event] of on(et, "tick")) {
+  //     count++;
+  //     assert.strictEqual(event.type, "tick");
+  //     if (count >= 5) {
+  //       break;
+  //     }
+  //   }
+  //   assert.strictEqual(count, 5);
+  //   clearInterval(interval);
   // }
 
   // async function eventTargetAbortableOnBefore() {
@@ -307,59 +375,6 @@ describe("node:events.on (EE async iterator)", () => {
   //   });
   // }
 
-  // it("should NOT throw an `AbortError` when done iterating over events", async done => {
-  //   const ee = new EventEmitter();
-  //   const ac = new AbortController();
-
-  //   const i = setInterval(() => ee.emit("foo", "foo"), 1);
-  //   let count = 0;
-
-  //   async function foo() {
-  //     for await (const f of on(ee, "foo", { signal: ac.signal })) {
-  //       assert.strictEqual(f[0], "foo");
-  //       if (++count === 5) break;
-  //     }
-  //     ac.abort(); // No error will occur
-  //   }
-
-  //   foo().finally(() => {
-  //     clearInterval(i);
-  //     expect(true).toBe(true);
-  //     done();
-  //   });
-  // });
-
-  // it("should throw an `AbortError` when NOT done iterating over events", async done => {
-  //   const createDone = createDoneDotAll(done);
-  //   const { mustCall, closeTimers } = createCallCheckCtx(createDone());
-  //   const finalDone = createDone();
-
-  //   const ee = new EventEmitter();
-  //   const ac = new AbortController();
-
-  //   const i = setInterval(() => ee.emit("foo", "foo"), 10);
-
-  //   async function foo() {
-  //     for await (const f of on(ee, "foo", { signal: ac.signal })) {
-  //       assert.strictEqual(f, "foo");
-  //     }
-  //   }
-
-  //   foo()
-  //     .catch(
-  //       mustCall(error => {
-  //         assert.strictEqual(error.name, "AbortError");
-  //       }),
-  //     )
-  //     .finally(() => {
-  //       clearInterval(i);
-  //       finalDone();
-  //       closeTimers();
-  //     });
-
-  //   process.nextTick(() => ac.abort());
-  // });
-
   // async function eventTargetAbortableOnAfter() {
   //   const et = new EventTarget();
   //   const ac = new AbortController();