events: allow safely adding listener to abortSignal

PR-URL: #48596
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>
Reviewed-By: Moshe Atlow <moshe@atlow.co.il>
Reviewed-By: Benjamin Gruenbaum <benjamingr@gmail.com>
Reviewed-By: Robert Nagy <ronagy@icloud.com>

Author: atlowChemi

doc/api/events.md

@@ -1799,6 +1799,64 @@ const emitter = new EventEmitter();
 setMaxListeners(5, target, emitter);
 ```
 
+## `events.addAbortListener(signal, resource)`
+
+<!-- YAML
+added: REPLACEME
+-->
+
+> Stability: 1 - Experimental
+
+* `signal` {AbortSignal}
+* `listener` {Function|EventListener}
+* Returns: {Disposable} that removes the `abort` listener.
+
+Listens once to the `abort` event on the provided `signal`.
+
+Listening to the `abort` event on abort signals is unsafe and may
+lead to resource leaks since another third party with the signal can
+call [`e.stopImmediatePropagation()`][]. Unfortunately Node.js cannot change
+this since it would violate the web standard. Additionally, the original
+API makes it easy to forget to remove listeners.
+
+This API allows safely using `AbortSignal`s in Node.js APIs by solving these
+two issues by listening to the event such that `stopImmediatePropagation` does
+not prevent the listener from running.
+
+Returns a disposable so that it may be unsubscribed from more easily.
+
+```cjs
+const { addAbortListener } = require('node:events');
+
+function example(signal) {
+  let disposable;
+  try {
+    signal.addEventListener('abort', (e) => e.stopImmediatePropagation());
+    disposable = addAbortListener(signal, (e) => {
+      // Do something when signal is aborted.
+    });
+  } finally {
+    disposable?.[Symbol.dispose]();
+  }
+}
+```
+
+```mjs
+import { addAbortListener } from 'node:events';
+
+function example(signal) {
+  let disposable;
+  try {
+    signal.addEventListener('abort', (e) => e.stopImmediatePropagation());
+    disposable = addAbortListener(signal, (e) => {
+      // Do something when signal is aborted.
+    });
+  } finally {
+    disposable?.[Symbol.dispose]();
+  }
+}
+```
+
 ## Class: `events.EventEmitterAsyncResource extends EventEmitter`
 
 <!-- YAML
@@ -2520,6 +2578,7 @@ to the `EventTarget`.
 [`EventTarget` error handling]: #eventtarget-error-handling
 [`Event` Web API]: https://dom.spec.whatwg.org/#event
 [`domain`]: domain.md
+[`e.stopImmediatePropagation()`]: #eventstopimmediatepropagation
 [`emitter.listenerCount()`]: #emitterlistenercounteventname-listener
 [`emitter.removeListener()`]: #emitterremovelistenereventname-listener
 [`emitter.setMaxListeners(n)`]: #emittersetmaxlistenersn

lib/events.js

@@ -48,6 +48,7 @@ const {
   Symbol,
   SymbolFor,
   SymbolAsyncIterator,
+  SymbolDispose,
 } = primordials;
 const kRejection = SymbolFor('nodejs.rejection');
 
@@ -214,6 +215,7 @@ function EventEmitter(opts) {
   EventEmitter.init.call(this, opts);
 }
 module.exports = EventEmitter;
+module.exports.addAbortListener = addAbortListener;
 module.exports.once = once;
 module.exports.on = on;
 module.exports.getEventListeners = getEventListeners;
@@ -1160,3 +1162,32 @@ function on(emitter, event, options = kEmptyObject) {
   }
   return iterator;
 }
+
+let queueMicrotask;
+
+function addAbortListener(signal, listener) {
+  if (signal === undefined) {
+    throw new ERR_INVALID_ARG_TYPE('signal', 'AbortSignal', signal);
+  }
+  validateAbortSignal(signal, 'signal');
+  validateFunction(listener, 'listener');
+
+  let removeEventListener;
+  if (signal.aborted) {
+    queueMicrotask ??= require('internal/process/task_queues').queueMicrotask;
+    queueMicrotask(() => listener());
+  } else {
+    kResistStopPropagation ??= require('internal/event_target').kResistStopPropagation;
+    // TODO(atlowChemi) add { subscription: true } and return directly
+    signal.addEventListener('abort', listener, { __proto__: null, once: true, [kResistStopPropagation]: true });
+    removeEventListener = () => {
+      signal.removeEventListener('abort', listener);
+    };
+  }
+  return {
+    __proto__: null,
+    [SymbolDispose]() {
+      removeEventListener?.();
+    },
+  };
+}

test/parallel/test-events-add-abort-listener.mjs

@@ -0,0 +1,55 @@
+import * as common from '../common/index.mjs';
+import * as events from 'node:events';
+import * as assert from 'node:assert';
+import { describe, it } from 'node:test';
+
+describe('events.addAbortListener', () => {
+  it('should throw if signal not provided', () => {
+    assert.throws(() => events.addAbortListener(), { code: 'ERR_INVALID_ARG_TYPE' });
+  });
+
+  it('should throw if provided signal is invalid', () => {
+    assert.throws(() => events.addAbortListener(undefined), { code: 'ERR_INVALID_ARG_TYPE' });
+    assert.throws(() => events.addAbortListener(null), { code: 'ERR_INVALID_ARG_TYPE' });
+    assert.throws(() => events.addAbortListener({}), { code: 'ERR_INVALID_ARG_TYPE' });
+  });
+
+  it('should throw if listener is not a function', () => {
+    const { signal } = new AbortController();
+    assert.throws(() => events.addAbortListener(signal), { code: 'ERR_INVALID_ARG_TYPE' });
+    assert.throws(() => events.addAbortListener(signal, {}), { code: 'ERR_INVALID_ARG_TYPE' });
+    assert.throws(() => events.addAbortListener(signal, undefined), { code: 'ERR_INVALID_ARG_TYPE' });
+  });
+
+  it('should return a Disposable', () => {
+    const { signal } = new AbortController();
+    const disposable = events.addAbortListener(signal, common.mustNotCall());
+
+    assert.strictEqual(typeof disposable[Symbol.dispose], 'function');
+  });
+
+  it('should execute the listener immediately for aborted runners', () => {
+    const disposable = events.addAbortListener(AbortSignal.abort(), common.mustCall());
+    assert.strictEqual(typeof disposable[Symbol.dispose], 'function');
+  });
+
+  it('should execute the listener even when event propagation stopped', () => {
+    const controller = new AbortController();
+    const { signal } = controller;
+
+    signal.addEventListener('abort', (e) => e.stopImmediatePropagation());
+    events.addAbortListener(
+      signal,
+      common.mustCall((e) => assert.strictEqual(e.target, signal)),
+    );
+
+    controller.abort();
+  });
+
+  it('should remove event listeners when disposed', () => {
+    const controller = new AbortController();
+    const disposable = events.addAbortListener(controller.signal, common.mustNotCall());
+    disposable[Symbol.dispose]();
+    controller.abort();
+  });
+});

tools/doc/type-parser.mjs

@@ -267,6 +267,7 @@ const customTypesMap = {
   'Headers': 'https://developer.mozilla.org/en-US/docs/Web/API/Headers',
   'Response': 'https://developer.mozilla.org/en-US/docs/Web/API/Response',
   'Request': 'https://developer.mozilla.org/en-US/docs/Web/API/Request',
+  'Disposable': 'https://tc39.es/proposal-explicit-resource-management/#sec-disposable-interface',
 };
 
 const arrayPart = /(?:\[])+$/;