events: remove the abort listener on iterator completion

nbbeeken · nbbeeken · commit 883b57537190 · 2024-01-27T08:42:17.000-05:00
Given an AbortSignal when passed into the events.on AsyncIterator API the attached abort listener should always be removed when the iterator completes. 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 addEventListener helper returned by the listenersController helper adds the abort listener to the array of listeners that will be cleaned up by removeAll in the closeHandler for the AsyncIterator. Fixes: #51010
diff --git a/lib/events.js b/lib/events.js
@@ -1169,7 +1169,7 @@ function on(emitter, event, options = kEmptyObject) {
   }
   if (signal) {
     kResistStopPropagation ??= require('internal/event_target').kResistStopPropagation;
-    eventTargetAgnosticAddListener(
+    addEventListener(
       signal,
       'abort',
       abortListener,
diff --git a/test/parallel/test-events-on-async-iterator.js b/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,28 @@ async function abortableOnAfterDone() {
   });
 }
 
+async function abortListenerRemovedAfterComplete() {
+  const ee = new EventEmitter();
+  const ac = new AbortController();
+
+  const i = setInterval(() => ee.emit('foo', 'foo'), 1);
+  try {
+    // Return case
+    const endedIterator = on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].size > 0);
+    endedIterator.return();
+    assert.strictEqual(ac.signal[kEvents].size, 0);
+
+    // Throw case
+    const throwIterator = on(ee, 'foo', { signal: ac.signal });
+    assert.ok(ac.signal[kEvents].size > 0);
+    throwIterator.throw(new Error());
+    assert.strictEqual(ac.signal[kEvents].size, 0);
+  } finally {
+    clearInterval(i);
+  }
+}
+
 async function run() {
   const funcs = [
     basic,
@@ -391,6 +414,7 @@ async function run() {
     eventTargetAbortableOnAfter,
     eventTargetAbortableOnAfter2,
     abortableOnAfterDone,
+    abortListenerRemovedAfterComplete,
   ];
 
   for (const fn of funcs) {

Original file line number	Diff line number	Diff line change
`@@ -1169,7 +1169,7 @@ function on(emitter, event, options = kEmptyObject) {`
`1169`	`1169`	`}`
`1170`	`1170`	`if (signal) {`
`1171`	`1171`	`kResistStopPropagation ??= require('internal/event_target').kResistStopPropagation;`
`1172`		`- eventTargetAgnosticAddListener(`
	`1172`	`+ addEventListener(`
`1173`	`1173`	`signal,`
`1174`	`1174`	`'abort',`
`1175`	`1175`	`abortListener,`