events: remove abort listener from signal in on

the `abortHandler` function is declared within the scope of
the `events.on` function so cannot be removed by the caller
which can lead to a memory leak
adding the abort listener using the `addAbortListener` helper
returns a disposable that can be used to clean up the listener
when the iterator is exited

Fixes: nodejs#51010
PR-URL: nodejs#51091
Reviewed-By: Chemi Atlow <[email protected]>
Reviewed-By: Benjamin Gruenbaum <[email protected]>

lib/events.js

@@ -1167,14 +1167,8 @@ function on(emitter, event, options = kEmptyObject) {
       addEventListener(emitter, closeEvents[i], closeHandler);
     }
   }
-  if (signal) {
-    kResistStopPropagation ??= require('internal/event_target').kResistStopPropagation;
-    eventTargetAgnosticAddListener(
-      signal,
-      'abort',
-      abortListener,
-      { __proto__: null, once: true, [kResistStopPropagation]: true });
-  }
+
+  const abortListenerDisposable = signal ? addAbortListener(signal, abortListener) : null;
 
   return iterator;
 
@@ -1201,6 +1195,7 @@ function on(emitter, event, options = kEmptyObject) {
   }
 
   function closeHandler() {
+    abortListenerDisposable?.[SymbolDispose]();
     removeAll();
     finished = true;
     const doneResult = createIterResult(undefined, true);

test/parallel/test-events-on-async-iterator.js

@@ -6,6 +6,7 @@ const assert = require('assert');
 const { on, EventEmitter } = require('events');
 const {
   NodeEventTarget,
+  kEvents
 } = require('internal/event_target');
 
 async function basic() {
@@ -372,6 +373,36 @@ async function abortableOnAfterDone() {
   });
 }
 
+async function abortListenerRemovedAfterComplete() {
+  const ee = new EventEmitter();
+  const ac = new AbortController();
+
+  const i = setInterval(() => ee.emit('foo', 'foo'), 1);
+  try {
+    // Below: either the kEvents map is empty or the 'abort' listener list is empty
+
+    // Return case
+    const endedIterator = on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].get('abort').size > 0);
+    endedIterator.return();
+    assert.strictEqual(ac.signal[kEvents].get('abort')?.size ?? ac.signal[kEvents].size, 0);
+
+    // Throw case
+    const throwIterator = on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].get('abort').size > 0);
+    throwIterator.throw(new Error());
+    assert.strictEqual(ac.signal[kEvents].get('abort')?.size ?? ac.signal[kEvents].size, 0);
+
+    // Abort case
+    on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].get('abort').size > 0);
+    ac.abort(new Error());
+    assert.strictEqual(ac.signal[kEvents].get('abort')?.size ?? ac.signal[kEvents].size, 0);
+  } finally {
+    clearInterval(i);
+  }
+}
+
 async function run() {
   const funcs = [
     basic,
@@ -391,6 +422,7 @@ async function run() {
     eventTargetAbortableOnAfter,
     eventTargetAbortableOnAfter2,
     abortableOnAfterDone,
+    abortListenerRemovedAfterComplete,
   ];
 
   for (const fn of funcs) {