util: fix debounce/throttle abort listener leak

lib/internal/util/debounce.js

@@ -18,12 +18,14 @@ const {
   validateInteger,
   validateObject,
 } = require('internal/validators');
-const { addAbortListener } = require('internal/events/abort_listener');
+const { kResistStopPropagation } = require('internal/event_target');
 const { kEmptyObject } = require('internal/util');
 const { TIMEOUT_MAX } = require('internal/timers');
 const { clearTimeout, setTimeout } = require('timers');
 const { markPromiseAsHandled } = internalBinding('util');
 
+const kAbortListenerOptions = { __proto__: null, once: true, [kResistStopPropagation]: true };
+
 /**
  * @typedef {object} DebounceOptions
  * @property {AbortSignal} [signal] An AbortSignal that cancels pending and future calls.
@@ -60,6 +62,7 @@ function debounce(fn, wait, options = kEmptyObject) {
 
   let args;
   let abortError;
+  let listeningForAbort = false;
   let pendingCalls = [];
   let refed = true;
   let timeout;
@@ -76,6 +79,7 @@ function debounce(fn, wait, options = kEmptyObject) {
     if (timeout === undefined) return;
     clearTimeout(timeout);
     timeout = undefined;
+    stopListeningForAbort();
     args = undefined;
     rejectPending(error);
   }
@@ -97,6 +101,20 @@ function debounce(fn, wait, options = kEmptyObject) {
     cancelWithError(abortError);
   }
 
+  function listenForAbort() {
+    if (signal !== undefined && !listeningForAbort) {
+      signal.addEventListener('abort', abort, kAbortListenerOptions);
+      listeningForAbort = true;
+    }
+  }
+
+  function stopListeningForAbort() {
+    if (listeningForAbort) {
+      signal.removeEventListener('abort', abort);
+      listeningForAbort = false;
+    }
+  }
+
   function ref() {
     refed = true;
     timeout?.ref();
@@ -143,6 +161,7 @@ function debounce(fn, wait, options = kEmptyObject) {
 
   function onTimeout() {
     timeout = undefined;
+    stopListeningForAbort();
     invokePending();
   }
 
@@ -151,6 +170,7 @@ function debounce(fn, wait, options = kEmptyObject) {
 
     clearTimeout(timeout);
     timeout = undefined;
+    stopListeningForAbort();
     invokePending();
   }
 
@@ -170,6 +190,7 @@ function debounce(fn, wait, options = kEmptyObject) {
     } else {
       timeout = setTimeout(onTimeout, wait);
       if (!refed) timeout.unref();
+      listenForAbort();
     }
 
     if (signal?.aborted) {
@@ -227,10 +248,6 @@ function debounce(fn, wait, options = kEmptyObject) {
     },
   });
 
-  if (signal !== undefined) {
-    addAbortListener(signal, abort);
-  }
-
   return debounced;
 }
 

lib/internal/util/throttle.js

@@ -25,14 +25,16 @@ const {
   validateObject,
   validateOneOf,
 } = require('internal/validators');
-const { addAbortListener } = require('internal/events/abort_listener');
+const { kResistStopPropagation } = require('internal/event_target');
 const { kEmptyObject } = require('internal/util');
 const { TIMEOUT_MAX } = require('internal/timers');
 const { clearTimeout, setTimeout } = require('timers');
 
 const timersBinding = internalBinding('timers');
 const { markPromiseAsHandled } = internalBinding('util');
 
+const kAbortListenerOptions = { __proto__: null, once: true, [kResistStopPropagation]: true };
+
 /**
  * @typedef {object} ThrottleOptions
  * @property {number} [concurrency] The maximum number of unsettled invocations.
@@ -79,6 +81,7 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
   }
 
   let abortError;
+  let listeningForAbort = false;
   let activeCount = 0;
   let pendingCalls = [];
   let pendingIndex = 0;
@@ -100,6 +103,7 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
     if (pendingIndex === pendingCalls.length) {
       pendingCalls = [];
       pendingIndex = 0;
+      stopListeningForAbort();
     } else if (pendingIndex > 1024 && pendingIndex * 2 >= pendingCalls.length) {
       pendingCalls = ArrayPrototypeSlice(pendingCalls, pendingIndex);
       pendingIndex = 0;
@@ -113,6 +117,7 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
     const start = pendingIndex;
     pendingCalls = [];
     pendingIndex = 0;
+    stopListeningForAbort();
     for (let i = start; i < calls.length; i++) {
       calls[i].reject(error);
     }
@@ -150,6 +155,20 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
     cancelWithError(abortError);
   }
 
+  function listenForAbort() {
+    if (signal !== undefined && !listeningForAbort) {
+      signal.addEventListener('abort', abort, kAbortListenerOptions);
+      listeningForAbort = true;
+    }
+  }
+
+  function stopListeningForAbort() {
+    if (listeningForAbort) {
+      signal.removeEventListener('abort', abort);
+      listeningForAbort = false;
+    }
+  }
+
   function ref() {
     refed = true;
     timeout?.ref();
@@ -306,6 +325,7 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
       invoke(call);
     } else {
       ArrayPrototypePush(pendingCalls, call);
+      listenForAbort();
       schedule();
     }
 
@@ -360,10 +380,6 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
     },
   });
 
-  if (signal !== undefined) {
-    addAbortListener(signal, abort);
-  }
-
   return throttled;
 }
 

test/parallel/test-util-debounce.js

@@ -4,10 +4,15 @@
 const common = require('../common');
 const assert = require('node:assert');
 const { createHook } = require('node:async_hooks');
+const { getEventListeners } = require('node:events');
 const { setTimeout } = require('node:timers/promises');
 const { debounce } = require('node:util');
 const { TIMEOUT_MAX } = require('internal/timers');
 
+function abortListenerCount(signal) {
+  return getEventListeners(signal, 'abort').length;
+}
+
 for (const value of [undefined, null, true, 0, 'fn', {}, [], Symbol()]) {
   assert.throws(() => debounce(value, 1), {
     code: 'ERR_INVALID_ARG_TYPE',
@@ -320,6 +325,41 @@ assert.throws(() => debounce(() => {}, 1, { signal: {} }), {
     assert.strictEqual(await result, 1);
   }
 
+  {
+    const { signal } = new AbortController();
+    const debounced = debounce(common.mustCall((value) => value, 2), 1, {
+      signal,
+    });
+    assert.strictEqual(abortListenerCount(signal), 0);
+    const first = debounced(1);
+    assert.strictEqual(abortListenerCount(signal), 1);
+    assert.strictEqual(await first, 1);
+    assert.strictEqual(abortListenerCount(signal), 0);
+
+    const second = debounced(2);
+    debounced.flush();
+    assert.strictEqual(abortListenerCount(signal), 0);
+    assert.strictEqual(await second, 2);
+
+    const third = debounced(3);
+    debounced.cancel();
+    assert.strictEqual(abortListenerCount(signal), 0);
+    await assert.rejects(third, { code: 'ABORT_ERR' });
+  }
+
+  {
+    const { signal } = new AbortController();
+    const debounced = debounce(common.mustCall((value) => value), 10, {
+      leading: true,
+      signal,
+    });
+    assert.strictEqual(await debounced(1), 1);
+    // The leading call keeps the listener until the wait is over.
+    assert.strictEqual(abortListenerCount(signal), 1);
+    await setTimeout(20);
+    assert.strictEqual(abortListenerCount(signal), 0);
+  }
+
   {
     const debounced = debounce(common.mustCall((value) => value), 1);
     assert.strictEqual(await debounced(42), 42);

test/parallel/test-util-throttle.js

@@ -4,10 +4,15 @@
 const common = require('../common');
 const assert = require('node:assert');
 const { createHook } = require('node:async_hooks');
+const { getEventListeners } = require('node:events');
 const { setImmediate } = require('node:timers/promises');
 const { throttle } = require('node:util');
 const { TIMEOUT_MAX } = require('internal/timers');
 
+function abortListenerCount(signal) {
+  return getEventListeners(signal, 'abort').length;
+}
+
 process.on('unhandledRejection', common.mustNotCall());
 
 for (const value of [undefined, null, true, 0, 'fn', {}, [], Symbol()]) {
@@ -257,6 +262,47 @@ throttle(() => {}, 1, 1, {
     );
   }
 
+  {
+    const { signal } = new AbortController();
+    const throttled = throttle(common.mustCall((value) => value, 2), 1, 10, {
+      signal,
+    });
+    assert.strictEqual(abortListenerCount(signal), 0);
+    const first = throttled(1);
+    assert.strictEqual(abortListenerCount(signal), 0);
+    const second = throttled(2);
+    assert.strictEqual(abortListenerCount(signal), 1);
+    assert.deepStrictEqual(await Promise.all([first, second]), [1, 2]);
+    assert.strictEqual(abortListenerCount(signal), 0);
+  }
+
+  {
+    const { signal } = new AbortController();
+    const throttled = throttle(common.mustCall(), 1, TIMEOUT_MAX, { signal });
+    throttled();
+    const queued = throttled();
+    assert.strictEqual(abortListenerCount(signal), 1);
+    throttled.cancel();
+    assert.strictEqual(abortListenerCount(signal), 0);
+    await assert.rejects(queued, { code: 'ABORT_ERR' });
+  }
+
+  {
+    // Calls waiting on concurrency are queued too, so they keep the listener.
+    const { signal } = new AbortController();
+    const { promise, resolve } = Promise.withResolvers();
+    const throttled = throttle(common.mustCall(() => promise, 2), 2, TIMEOUT_MAX, {
+      concurrency: 1,
+      signal,
+    });
+    const first = throttled();
+    const second = throttled();
+    assert.strictEqual(abortListenerCount(signal), 1);
+    resolve();
+    await Promise.all([first, second]);
+    assert.strictEqual(abortListenerCount(signal), 0);
+  }
+
   {
     const expected = new Error('failure');
     const throttled = throttle(common.mustCall((value) => {