diff --git a/lib/web/fetch/index.js b/lib/web/fetch/index.js index 1959f27c04a..9745e167d7d 100644 --- a/lib/web/fetch/index.js +++ b/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,6 +261,7 @@ function fetch (input, init = undefined) { // deserializedError. abortFetch(p, request, responseObject, controller.serializedAbortReason, controller.controller) + cleanupAbortListeners() return } @@ -259,6 +269,7 @@ function fetch (input, init = undefined) { // 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 diff --git a/lib/web/fetch/request.js b/lib/web/fetch/request.js index 6ef40f99920..1fb6b8a45e5 100644 --- a/lib/web/fetch/request.js +++ b/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 } diff --git a/test/fetch/issue-5285.js b/test/fetch/issue-5285.js new file mode 100644 index 00000000000..77a5cb2bbff --- /dev/null +++ b/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) +})