util: fix debounce/throttle abort listener leak

debounce() and throttle() added an abort listener to the signal when
they were created and never removed it, so a long-lived signal kept
every debounced or throttled function, and its data, in memory.

Only listen for abort while calls are pending. Calls made after the
signal is aborted still reject, since each call checks signal.aborted.

Assisted-by: a closed-source coding agent
Signed-off-by: vedchaudhari <vedc2853@gmail.com>

Author: vedchaudhari

lib/internal/util/debounce.js

@@ -6,6 +6,7 @@ const {
   PromiseReject,
   PromiseWithResolvers,
   ReflectApply,
+  SymbolDispose,
 } = primordials;
 
 const {
@@ -60,6 +61,7 @@ function debounce(fn, wait, options = kEmptyObject) {
 
   let args;
   let abortError;
+  let abortListener;
   let pendingCalls = [];
   let refed = true;
   let timeout;
@@ -76,6 +78,7 @@ function debounce(fn, wait, options = kEmptyObject) {
     if (timeout === undefined) return;
     clearTimeout(timeout);
     timeout = undefined;
+    stopListeningForAbort();
     args = undefined;
     rejectPending(error);
   }
@@ -97,6 +100,19 @@ function debounce(fn, wait, options = kEmptyObject) {
     cancelWithError(abortError);
   }
 
+  function listenForAbort() {
+    if (signal !== undefined && abortListener === undefined) {
+      abortListener = addAbortListener(signal, abort);
+    }
+  }
+
+  function stopListeningForAbort() {
+    if (abortListener !== undefined) {
+      abortListener[SymbolDispose]();
+      abortListener = undefined;
+    }
+  }
+
   function ref() {
     refed = true;
     timeout?.ref();
@@ -143,6 +159,7 @@ function debounce(fn, wait, options = kEmptyObject) {
 
   function onTimeout() {
     timeout = undefined;
+    stopListeningForAbort();
     invokePending();
   }
 
@@ -151,6 +168,7 @@ function debounce(fn, wait, options = kEmptyObject) {
 
     clearTimeout(timeout);
     timeout = undefined;
+    stopListeningForAbort();
     invokePending();
   }
 
@@ -170,6 +188,7 @@ function debounce(fn, wait, options = kEmptyObject) {
     } else {
       timeout = setTimeout(onTimeout, wait);
       if (!refed) timeout.unref();
+      listenForAbort();
     }
 
     if (signal?.aborted) {
@@ -227,10 +246,6 @@ function debounce(fn, wait, options = kEmptyObject) {
     },
   });
 
-  if (signal !== undefined) {
-    addAbortListener(signal, abort);
-  }
-
   return debounced;
 }
 

lib/internal/util/throttle.js

@@ -9,6 +9,7 @@ const {
   PromiseResolve,
   PromiseWithResolvers,
   ReflectApply,
+  SymbolDispose,
 } = primordials;
 
 const {
@@ -79,6 +80,7 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
   }
 
   let abortError;
+  let abortListener;
   let activeCount = 0;
   let pendingCalls = [];
   let pendingIndex = 0;
@@ -100,6 +102,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 +116,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 +154,19 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
     cancelWithError(abortError);
   }
 
+  function listenForAbort() {
+    if (signal !== undefined && abortListener === undefined) {
+      abortListener = addAbortListener(signal, abort);
+    }
+  }
+
+  function stopListeningForAbort() {
+    if (abortListener !== undefined) {
+      abortListener[SymbolDispose]();
+      abortListener = undefined;
+    }
+  }
+
   function ref() {
     refed = true;
     timeout?.ref();
@@ -306,6 +323,7 @@ function throttle(fn, limit, interval, options = kEmptyObject) {
       invoke(call);
     } else {
       ArrayPrototypePush(pendingCalls, call);
+      listenForAbort();
       schedule();
     }
 
@@ -360,10 +378,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) => {