fix(fetch): remove abort listener when request settles

lib/web/fetch/index.js

@@ -11,7 +11,7 @@ const {
   getResponseState
 } = require('./response')
 const { HeadersList } = require('./headers')
-const { Request, cloneRequest, getRequestDispatcher, getRequestState } = require('./request')
+const { Request, cloneRequest, getRequestDispatcher, getRequestState, removeRequestAbortListener } = require('./request')
 const zlib = require('node:zlib')
 const {
   makePolicyContainer,
@@ -208,7 +208,7 @@ function fetch (input, init = undefined) {
   let controller = null
 
   // 11. Add the following abort steps to requestObject’s signal:
-  addAbortListener(
+  const removeAbortListener = addAbortListener(
     requestObject.signal,
     () => {
       // 1. Set locallyAborted to true.
@@ -228,6 +228,15 @@ function fetch (input, init = undefined) {
     }
   )
 
+  // Remove the `abort` listeners registered above and in the Request
+  // constructor once the fetch has settled. Without this, reusing a single
+  // signal across many requests leaks listeners and Node.js emits a
+  // MaxListenersExceededWarning. See https://github.com/nodejs/undici/issues/5285
+  const cleanupAbortListeners = () => {
+    removeAbortListener()
+    removeRequestAbortListener(requestObject)
+  }
+
   // 12. Let handleFetchDone given response response be to finalize and
   // report timing with response, globalObject, and "fetch".
   // see function handleFetchDone
@@ -252,13 +261,15 @@ function fetch (input, init = undefined) {
       //    deserializedError.
 
       abortFetch(p, request, responseObject, controller.serializedAbortReason, controller.controller)
+      cleanupAbortListeners()
       return
     }
 
     // 3. If response is a network error, then reject p with a TypeError
     // and terminate these substeps.
     if (response.type === 'error') {
       p.reject(new TypeError('fetch failed', { cause: response.error }))
+      cleanupAbortListeners()
       return
     }
 
@@ -273,7 +284,10 @@ function fetch (input, init = undefined) {
 
   controller = fetching({
     request,
-    processResponseEndOfBody: handleFetchDone,
+    processResponseEndOfBody: (response) => {
+      handleFetchDone(response)
+      cleanupAbortListeners()
+    },
     processResponse,
     dispatcher: getRequestDispatcher(requestObject), // undici
     // Keep requestObject alive to prevent its AbortController from being GC'd

lib/web/fetch/request.js

@@ -97,6 +97,13 @@ class Request {
 
   #state
 
+  /**
+   * Removes the `abort` listener that makes this request's signal follow the
+   * passed signal. `null` when no such listener was registered.
+   * @type {(() => void) | null}
+   */
+  #abortCleanup = null
+
   // https://fetch.spec.whatwg.org/#dom-request
   constructor (input, init = undefined) {
     webidl.util.markAsUncloneable(this)
@@ -436,12 +443,23 @@ class Request {
           setMaxListeners(1500, signal)
         }
 
-        util.addAbortListener(signal, abort)
+        const removeAbortListener = util.addAbortListener(signal, abort)
         // The third argument must be a registry key to be unregistered.
         // Without it, you cannot unregister.
         // https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/FinalizationRegistry
         // abort is used as the unregister key. (because it is unique)
         requestFinalizer.register(ac, { signal, abort }, abort)
+
+        // Allow the listener to be removed deterministically once the fetch
+        // that owns this request has settled, instead of relying solely on the
+        // FinalizationRegistry (i.e. garbage collection). Reusing a single
+        // signal across many requests would otherwise leak listeners.
+        // See https://github.com/nodejs/undici/issues/5285
+        this.#abortCleanup = () => {
+          requestFinalizer.unregister(abort)
+          removeAbortListener()
+          this.#abortCleanup = null
+        }
       }
     }
 
@@ -868,15 +886,25 @@ class Request {
   static setRequestState (request, newState) {
     request.#state = newState
   }
+
+  /**
+   * Removes the `abort` listener that makes this request's signal follow the
+   * signal passed to its constructor, if any. Idempotent.
+   * @param {Request} request
+   */
+  static removeRequestAbortListener (request) {
+    request.#abortCleanup?.()
+  }
 }
 
-const { setRequestSignal, getRequestDispatcher, setRequestDispatcher, setRequestHeaders, getRequestState, setRequestState } = Request
+const { setRequestSignal, getRequestDispatcher, setRequestDispatcher, setRequestHeaders, getRequestState, setRequestState, removeRequestAbortListener } = Request
 Reflect.deleteProperty(Request, 'setRequestSignal')
 Reflect.deleteProperty(Request, 'getRequestDispatcher')
 Reflect.deleteProperty(Request, 'setRequestDispatcher')
 Reflect.deleteProperty(Request, 'setRequestHeaders')
 Reflect.deleteProperty(Request, 'getRequestState')
 Reflect.deleteProperty(Request, 'setRequestState')
+Reflect.deleteProperty(Request, 'removeRequestAbortListener')
 
 mixinBody(Request, getRequestState)
 
@@ -1111,5 +1139,6 @@ module.exports = {
   fromInnerRequest,
   cloneRequest,
   getRequestDispatcher,
-  getRequestState
+  getRequestState,
+  removeRequestAbortListener
 }

test/fetch/issue-5285.js

@@ -0,0 +1,51 @@
+'use strict'
+
+const { test } = require('node:test')
+const assert = require('node:assert')
+const http = require('node:http')
+const { once, getEventListeners } = require('node:events')
+const { fetch } = require('../..')
+const { closeServerAsPromise } = require('../utils/node-http')
+
+// https://github.com/nodejs/undici/issues/5285
+// Reusing a single AbortSignal across many fetch() calls must not leak
+// `abort` listeners on the signal, which previously caused Node.js to emit
+// a MaxListenersExceededWarning.
+test('fetch removes the abort listener once the request settles', async (t) => {
+  const server = http.createServer({ joinDuplicateHeaders: true }, (req, res) => {
+    res.writeHead(200, { 'content-type': 'text/plain' })
+    res.end('hello')
+  })
+
+  t.after(closeServerAsPromise(server))
+  await once(server.listen(0), 'listening')
+
+  let warning = null
+  function onWarning (value) {
+    warning = value
+  }
+  process.on('warning', onWarning)
+  t.after(() => process.off('warning', onWarning))
+
+  const controller = new AbortController()
+  const { signal } = controller
+
+  const url = `http://localhost:${server.address().port}`
+
+  // Issue many more requests than the default max listeners (10) while
+  // sharing the same signal. Each settled request must remove its listener,
+  // otherwise a MaxListenersExceededWarning is emitted and the listeners leak.
+  for (let i = 0; i < 100; i++) {
+    const res = await fetch(url, { signal })
+    await res.text()
+  }
+
+  // Allow the trailing end-of-body cleanup of the final request, which is
+  // scheduled in a microtask, to run before asserting.
+  await new Promise((resolve) => setTimeout(resolve, 100))
+
+  // No `abort` listeners should remain registered on the signal once every
+  // request has settled.
+  assert.strictEqual(getEventListeners(signal, 'abort').length, 0)
+  assert.strictEqual(warning, null)
+})