From c9991178c66aef932bfc5716d7f69ee5c492a546 Mon Sep 17 00:00:00 2001 From: Arman-Luthra <129011616+Arman-Luthra@users.noreply.github.com> Date: Fri, 10 Jul 2026 16:48:32 -0700 Subject: [PATCH] feat(interceptors): add signal interceptor Add a new interceptor that aborts the dispatched request when the signal dispatch option is aborted, per the direction in #3276. Supports both AbortSignal and EventEmitter-style signals, aborts with signal.reason when available (RequestAbortedError otherwise), and removes the abort listener once the request completes, errors or is upgraded. A signal that is already aborted fails the dispatch before the request reaches the server. --- docs/docs/api/Dispatcher.md | 4 +- docs/docs/api/Interceptors.md | 45 +++++- index.js | 3 +- lib/interceptor/signal.js | 93 +++++++++++ test/interceptors/signal.js | 287 ++++++++++++++++++++++++++++++++++ test/types/index.test-d.ts | 1 + types/interceptors.d.ts | 1 + 7 files changed, 430 insertions(+), 4 deletions(-) create mode 100644 lib/interceptor/signal.js create mode 100644 test/interceptors/signal.js diff --git a/docs/docs/api/Dispatcher.md b/docs/docs/api/Dispatcher.md index 1e03811f94c..1355ade70a8 100644 --- a/docs/docs/api/Dispatcher.md +++ b/docs/docs/api/Dispatcher.md @@ -786,8 +786,8 @@ can make progress. ## Pre-built interceptors For the full reference of built-in interceptors (`dump`, `retry`, `redirect`, -`decompress`, `responseError`, `dns`, `cache`, `deduplicate`) and their options, -see [Interceptors](Interceptors.md). +`decompress`, `responseError`, `dns`, `cache`, `deduplicate`, `signal`) and +their options, see [Interceptors](Interceptors.md). [Fastify]: https://fastify.dev [GOAWAY frame]: https://webconcepts.info/concepts/http2-frame-type/0x7 diff --git a/docs/docs/api/Interceptors.md b/docs/docs/api/Interceptors.md index c6a02a14b9c..9c6a07cb80d 100644 --- a/docs/docs/api/Interceptors.md +++ b/docs/docs/api/Interceptors.md @@ -9,7 +9,7 @@ retries, response decompression, redirect following, DNS caching, and more. ```js import { Agent, interceptors } from 'undici' -const { retry, redirect, decompress, dump, responseError, dns, cache, deduplicate } = interceptors +const { retry, redirect, decompress, dump, responseError, dns, cache, deduplicate, signal } = interceptors const agent = new Agent().compose([ retry({ maxRetries: 3 }), @@ -329,6 +329,49 @@ const agent = new Agent().compose( --- +## `interceptors.signal()` + +Aborts the request when the `signal` dispatch option is aborted. Accepts both +[`AbortSignal`](https://developer.mozilla.org/en-US/docs/Web/API/AbortSignal) +and `EventEmitter`-style signals (any object emitting an `abort` event). The +request is aborted with `signal.reason` when available, or a +`RequestAbortedError` otherwise. If the signal is already aborted when the +request is dispatched, the request fails immediately without reaching the +server. + +The abort listener is removed as soon as the request completes, errors or is +upgraded, so long-lived signals do not accumulate listeners across requests. + +This is mainly useful when calling `dispatch()` directly with a custom +handler; the higher-level API methods (`request`, `stream`, `pipeline`, etc.) +already handle the `signal` option themselves. + +**Parameters** + +* None. + +**Returns:** {Dispatcher.DispatcherComposeInterceptor} + +**Example** + +```js +import { Client, interceptors } from 'undici' + +const client = new Client('https://example.com').compose( + interceptors.signal() +) + +const ac = new AbortController() +queueMicrotask(() => ac.abort()) + +client.dispatch( + { method: 'GET', path: '/', signal: ac.signal }, + myCustomHandler +) +``` + +--- + ## Composing multiple interceptors Interceptors are applied in the order they appear in the `compose()` call. diff --git a/index.js b/index.js index 7ece983f1db..c250dec32fc 100644 --- a/index.js +++ b/index.js @@ -54,7 +54,8 @@ module.exports.interceptors = { dns: require('./lib/interceptor/dns'), cache: require('./lib/interceptor/cache'), decompress: require('./lib/interceptor/decompress'), - deduplicate: require('./lib/interceptor/deduplicate') + deduplicate: require('./lib/interceptor/deduplicate'), + signal: require('./lib/interceptor/signal') } module.exports.cacheStores = { diff --git a/lib/interceptor/signal.js b/lib/interceptor/signal.js new file mode 100644 index 00000000000..771540828c3 --- /dev/null +++ b/lib/interceptor/signal.js @@ -0,0 +1,93 @@ +'use strict' + +const { InvalidArgumentError, RequestAbortedError } = require('../core/errors') +const { addAbortListener } = require('../core/util') +const DecoratorHandler = require('../handler/decorator-handler') + +class SignalHandler extends DecoratorHandler { + /** @type {AbortSignal|import('node:events').EventEmitter} */ + #signal + /** @type {import('../core/request').RequestController|null} */ + #controller = null + /** @type {(() => void)|null} */ + #removeAbortListener = null + /** @type {boolean} */ + #aborted = false + + constructor (signal, handler) { + super(handler) + this.#signal = signal + } + + #abort () { + this.#aborted = true + this.#removeListener() + this.#controller.abort(this.#signal.reason ?? new RequestAbortedError()) + } + + #removeListener () { + if (this.#removeAbortListener !== null) { + this.#removeAbortListener() + this.#removeAbortListener = null + } + } + + onRequestStart (controller, context) { + // A fresh controller is passed for every dispatch of the same request + // (e.g. when composed with the retry interceptor), so keep the reference + // up to date while registering the abort listener only once. + this.#controller = controller + + if (this.#aborted || this.#signal.aborted) { + this.#abort() + return + } + + if (this.#removeAbortListener === null) { + this.#removeAbortListener = addAbortListener(this.#signal, () => this.#abort()) + } + + super.onRequestStart(controller, context) + } + + onRequestUpgrade (controller, statusCode, headers, socket) { + this.#removeListener() + super.onRequestUpgrade(controller, statusCode, headers, socket) + } + + onResponseEnd (controller, trailers) { + this.#removeListener() + super.onResponseEnd(controller, trailers) + } + + onResponseError (controller, err) { + this.#removeListener() + super.onResponseError(controller, err) + } +} + +/** + * Interceptor that aborts the request when the `signal` dispatch option is + * aborted. Supports both `AbortSignal` and `EventEmitter`-style signals. + * The abort listener is removed once the request completes, errors or is + * upgraded. + * + * @returns {import('../../types/dispatcher').default.DispatcherComposeInterceptor} + */ +module.exports = () => { + return (dispatch) => { + return function signalInterceptor (opts, handler) { + const { signal } = opts + + if (!signal) { + return dispatch(opts, handler) + } + + if (typeof signal.on !== 'function' && typeof signal.addEventListener !== 'function') { + throw new InvalidArgumentError('signal must be an EventEmitter or EventTarget') + } + + return dispatch(opts, new SignalHandler(signal, handler)) + } + } +} diff --git a/test/interceptors/signal.js b/test/interceptors/signal.js new file mode 100644 index 00000000000..7a1d6feae8b --- /dev/null +++ b/test/interceptors/signal.js @@ -0,0 +1,287 @@ +'use strict' + +const assert = require('node:assert') +const { once, EventEmitter, getEventListeners } = require('node:events') +const { createServer } = require('node:http') +const { test, after } = require('node:test') +const { interceptors, errors, Client } = require('../..') +const { signal } = interceptors + +function startServer (handler) { + const server = createServer({ joinDuplicateHeaders: true }, handler) + + server.listen(0) + + return once(server, 'listening').then(() => server) +} + +function createHandler (events) { + let resolve + const completed = new Promise((_resolve) => { + resolve = _resolve + }) + + return { + completed, + error: null, + onRequestStart (controller, context) { + events.push('onRequestStart') + }, + onResponseStart (controller, statusCode, headers, statusMessage) { + events.push('onResponseStart') + }, + onResponseData (controller, chunk) { + events.push('onResponseData') + }, + onResponseEnd (controller, trailers) { + events.push('onResponseEnd') + resolve() + }, + onResponseError (controller, err) { + events.push('onResponseError') + this.error = err + resolve() + } + } +} + +test('aborts the request when the signal is already aborted', async () => { + const server = await startServer((req, res) => { + res.end('hello') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.close() + server.close() + + await once(server, 'close') + }) + + const reason = new Error('pre-aborted') + const ac = new AbortController() + ac.abort(reason) + + const events = [] + const handler = createHandler(events) + + client.dispatch({ method: 'GET', path: '/', signal: ac.signal }, handler) + await handler.completed + + assert.deepStrictEqual(events, ['onResponseError']) + assert.strictEqual(handler.error, reason) +}) + +test('aborts an in-flight request when the signal is aborted', async () => { + const server = await startServer((req, res) => { + res.writeHead(200, { 'content-type': 'text/plain' }) + res.write('partial') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.destroy() + server.close() + + await once(server, 'close') + }) + + const reason = new Error('custom abort reason') + const ac = new AbortController() + + const events = [] + const handler = createHandler(events) + handler.onResponseData = (controller, chunk) => { + events.push('onResponseData') + ac.abort(reason) + } + + client.dispatch({ method: 'GET', path: '/', signal: ac.signal }, handler) + await handler.completed + + assert.deepStrictEqual(events, [ + 'onRequestStart', + 'onResponseStart', + 'onResponseData', + 'onResponseError' + ]) + assert.strictEqual(handler.error, reason) + assert.strictEqual(getEventListeners(ac.signal, 'abort').length, 0) +}) + +test('supports EventEmitter-style signals', async () => { + const server = await startServer((req, res) => { + res.writeHead(200, { 'content-type': 'text/plain' }) + res.write('partial') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.destroy() + server.close() + + await once(server, 'close') + }) + + const ee = new EventEmitter() + + const events = [] + const handler = createHandler(events) + handler.onResponseData = (controller, chunk) => { + events.push('onResponseData') + ee.emit('abort') + } + + client.dispatch({ method: 'GET', path: '/', signal: ee }, handler) + await handler.completed + + assert.ok(handler.error instanceof errors.RequestAbortedError) + assert.strictEqual(ee.listenerCount('abort'), 0) +}) + +test('removes the abort listener once the response completes', async () => { + const server = await startServer((req, res) => { + res.end('hello') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.close() + server.close() + + await once(server, 'close') + }) + + const ac = new AbortController() + + const events = [] + const handler = createHandler(events) + + client.dispatch({ method: 'GET', path: '/', signal: ac.signal }, handler) + await handler.completed + + assert.deepStrictEqual(events, [ + 'onRequestStart', + 'onResponseStart', + 'onResponseData', + 'onResponseEnd' + ]) + assert.strictEqual(handler.error, null) + assert.strictEqual(getEventListeners(ac.signal, 'abort').length, 0) + + // Aborting after completion must not affect the finished request. + ac.abort() +}) + +test('removes the abort listener when the request errors', async () => { + const server = await startServer((req, res) => { + res.destroy() + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.close() + server.close() + + await once(server, 'close') + }) + + const ac = new AbortController() + + const events = [] + const handler = createHandler(events) + + client.dispatch({ method: 'GET', path: '/', signal: ac.signal }, handler) + await handler.completed + + assert.ok(handler.error) + assert.strictEqual(getEventListeners(ac.signal, 'abort').length, 0) +}) + +test('does not interfere when no signal is provided', async () => { + const server = await startServer((req, res) => { + res.end('hello') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.close() + server.close() + + await once(server, 'close') + }) + + const response = await client.request({ method: 'GET', path: '/' }) + + assert.strictEqual(response.statusCode, 200) + assert.strictEqual(await response.body.text(), 'hello') +}) + +test('throws when the signal is not an EventEmitter or EventTarget', async () => { + const server = await startServer((req, res) => { + res.end('hello') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.close() + server.close() + + await once(server, 'close') + }) + + assert.throws( + () => client.dispatch( + { method: 'GET', path: '/', signal: 'not-a-signal' }, + createHandler([]) + ), + errors.InvalidArgumentError + ) +}) + +test('works with client.request', async () => { + const server = await startServer((req, res) => { + res.writeHead(200, { 'content-type': 'text/plain' }) + res.write('partial') + }) + + const client = new Client( + `http://localhost:${server.address().port}` + ).compose(signal()) + + after(async () => { + await client.destroy() + server.close() + + await once(server, 'close') + }) + + const reason = new Error('custom abort reason') + const ac = new AbortController() + + const response = await client.request({ method: 'GET', path: '/', signal: ac.signal }) + ac.abort(reason) + + await assert.rejects(response.body.text(), reason) +}) diff --git a/test/types/index.test-d.ts b/test/types/index.test-d.ts index 2d5c425b35e..88f9b966f4d 100644 --- a/test/types/index.test-d.ts +++ b/test/types/index.test-d.ts @@ -32,6 +32,7 @@ expectAssignable(Undici.interceptors.re expectAssignable(Undici.interceptors.retry()) expectAssignable(Undici.interceptors.decompress()) expectAssignable(Undici.interceptors.cache()) +expectAssignable(Undici.interceptors.signal()) expectAssignable(new Undici.cacheStores.MemoryCacheStore()) expectAssignable(new Undici.cacheStores.SqliteCacheStore()) expectAssignable(new cacheStores.MemoryCacheStore()) diff --git a/types/interceptors.d.ts b/types/interceptors.d.ts index d21d717cec5..e1d5a8fed78 100644 --- a/types/interceptors.d.ts +++ b/types/interceptors.d.ts @@ -77,4 +77,5 @@ declare namespace Interceptors { export function dns (opts?: DNSInterceptorOpts): Dispatcher.DispatcherComposeInterceptor export function cache (opts?: CacheInterceptorOpts): Dispatcher.DispatcherComposeInterceptor export function deduplicate (opts?: DeduplicateInterceptorOpts): Dispatcher.DispatcherComposeInterceptor + export function signal (): Dispatcher.DispatcherComposeInterceptor }