node/lib/internal/abort_controller.js

'use strict';

// Modeled very closely on the AbortController implementation
// in https://github.com/mysticatea/abort-controller (MIT license)

const {
  ArrayPrototypePush,
  FunctionPrototypeCall,
  ObjectAssign,
  ObjectDefineProperties,
  ObjectDefineProperty,
  ObjectFreeze,
  ObjectSetPrototypeOf,
  PromiseResolve,
  PromiseWithResolvers,
  SafeFinalizationRegistry,
  SafeSet,
  SafeWeakRef,
  Symbol,
  SymbolToStringTag,
} = primordials;

const {
  defineEventHandler,
  EventTarget,
  Event,
  kEvents,
  kTrustEvent,
  kNewListener,
  kRemoveListener,
  kResistStopPropagation,
  kWeakHandler,
} = require('internal/event_target');
const { kMaxEventTargetListeners } = require('events');
const {
  customInspectSymbol,
  kEnumerableProperty,
} = require('internal/util');
const { inspect } = require('internal/util/inspect');
const {
  codes: {
    ERR_ILLEGAL_CONSTRUCTOR,
    ERR_INVALID_ARG_TYPE,
    ERR_INVALID_THIS,
  },
  isErrorStackTraceLimitWritable,
} = require('internal/errors');
const {
  converters,
  createInterfaceConverter,
  createSequenceConverter,
} = require('internal/webidl');

const {
  validateObject,
  validateUint32,
  kValidateObjectAllowObjects,
} = require('internal/validators');

const {
  DOMException,
} = internalBinding('messaging');

const {
  clearTimeout,
  setTimeout,
} = require('timers');
const assert = require('internal/assert');

const {
  kDeserialize,
  kTransfer,
  kTransferList,
  markTransferMode,
} = require('internal/worker/js_transferable');

let _MessageChannel;

const kDontThrowSymbol = Symbol('kDontThrowSymbol');

const kTrustedEventInit = ObjectFreeze({ [kTrustEvent]: true });

// V8 doesn't optimize keyed super property access such as
// super[kNewListener](), so call the EventTarget implementations directly.
const {
  [kNewListener]: eventTargetNewListener,
  [kRemoveListener]: eventTargetRemoveListener,
} = EventTarget.prototype;

// Loading the MessageChannel and markTransferable have to be done lazily
// because otherwise we'll end up with a require cycle that ends up with
// an incomplete initialization of abort_controller.

function lazyMessageChannel() {
  _MessageChannel ??= require('internal/worker/io').MessageChannel;
  return new _MessageChannel();
}

const clearTimeoutRegistry = new SafeFinalizationRegistry((holder) => {
  if (holder.timeout !== undefined) {
    clearTimeout(holder.timeout);
  }
});

// Each followed composite signal is registered once. When it is collected,
// remove it from the dependant signals of its remaining sources.
const dependantSignalsCleanupRegistry = new SafeFinalizationRegistry(
  ({ sourceSignals, dependantSignalRef }) => {
    for (const sourceSignalRef of sourceSignals) {
      sourceSignalRef.deref()?.[kDependantSignals]?.delete(dependantSignalRef);
    }
  });

const gcPersistentSignals = new SafeSet();

// Each source signal is registered once, when it gets its first dependant.
// When it is collected, remove it from the source signals of its remaining
// dependants. Registering per source rather than per (source, dependant) pair
// avoids unregister tokens, which are expensive to register.
const sourceSignalsCleanupRegistry = new SafeFinalizationRegistry(({ sourceSignalRef, dependantSignals }) => {
  for (const composedSignalRef of dependantSignals) {
    const composedSignal = composedSignalRef.deref();
    if (composedSignal !== undefined) {
      composedSignal[kSourceSignals].delete(sourceSignalRef);

      if (composedSignal[kSourceSignals].size === 0) {
        // This signal will no longer abort. There's no need to keep it in the gcPersistentSignals set.
        gcPersistentSignals.delete(composedSignal);
      }
    }
  }
});

const kAborted = Symbol('kAborted');
const kReason = Symbol('kReason');
const kCloneData = Symbol('kCloneData');
const kTimeout = Symbol('kTimeout');
const kMakeTransferable = Symbol('kMakeTransferable');
const kComposite = Symbol('kComposite');
const kFollowing = Symbol('kFollowing');
const kWeakRef = Symbol('kWeakRef');
const kSourceSignals = Symbol('kSourceSignals');
const kDependantSignals = Symbol('kDependantSignals');

// A signal's own WeakRef, shared by every composite that follows it so that
// a long-lived source doesn't allocate a new WeakRef per AbortSignal.any().
function getWeakRef(signal) {
  return signal[kWeakRef] ??= new SafeWeakRef(signal);
}

function customInspect(self, obj, depth, options) {
  if (depth < 0)
    return self;

  const opts = ObjectAssign({}, options, {
    depth: options.depth === null ? null : options.depth - 1,
  });

  return `${self.constructor.name} ${inspect(obj, opts)}`;
}

function validateThisAbortSignal(obj) {
  if (obj?.[kAborted] === undefined)
    throw new ERR_INVALID_THIS('AbortSignal');
}

function refreshCompositeSignal(signal) {
  if (!signal[kComposite] || signal[kAborted] || !signal[kSourceSignals]?.size) {
    return;
  }

  for (const sourceSignalWeakRef of signal[kSourceSignals]) {
    const sourceSignal = sourceSignalWeakRef.deref();
    if (sourceSignal === undefined) {
      signal[kSourceSignals].delete(sourceSignalWeakRef);
      continue;
    }

    if (sourceSignal[kAborted]) {
      abortSignal(signal, sourceSignal[kReason]);
      return;
    }
  }
}

function followCompositeSignal(signal) {
  if (signal[kFollowing] || signal[kAborted] || !signal[kSourceSignals]?.size) {
    return;
  }

  const resultSignalWeakRef = getWeakRef(signal);
  const sourceSignals = signal[kSourceSignals];

  dependantSignalsCleanupRegistry.register(signal, {
    sourceSignals,
    dependantSignalRef: resultSignalWeakRef,
  });

  for (const sourceSignalWeakRef of sourceSignals) {
    const sourceSignal = sourceSignalWeakRef.deref();
    if (sourceSignal === undefined) {
      sourceSignals.delete(sourceSignalWeakRef);
      continue;
    }

    if (sourceSignal[kAborted]) {
      abortSignal(signal, sourceSignal[kReason]);
      return;
    }

    let dependantSignals = sourceSignal[kDependantSignals];
    if (dependantSignals === undefined) {
      dependantSignals = sourceSignal[kDependantSignals] = new SafeSet();
      sourceSignalsCleanupRegistry.register(sourceSignal, {
        sourceSignalRef: sourceSignalWeakRef,
        dependantSignals,
      });
    }
    dependantSignals.add(resultSignalWeakRef);
  }

  signal[kFollowing] = true;
}

// Because the AbortSignal timeout cannot be canceled, we don't want the
// presence of the timer alone to keep the AbortSignal from being garbage
// collected if it otherwise no longer accessible. We also don't want the
// timer to keep the Node.js process open on it's own. Therefore, we wrap
// the AbortSignal in a WeakRef and have the setTimeout callback close
// over the WeakRef rather than directly over the AbortSignal, and we unref
// the created timer object. Separately, we add the signal to a
// FinalizerRegistry that will clear the timeout when the signal is gc'd.
// The registry holds the timeout through a holder that is emptied once the
// timer fires, which releases the fired timer without having to unregister
// the signal. Registering with an unregister token is several times more
// expensive than registering without one.
function setWeakAbortSignalTimeout(weakRef, delay) {
  const holder = { timeout: undefined };
  holder.timeout = setTimeout(() => {
    holder.timeout = undefined;
    const signal = weakRef.deref();
    if (signal !== undefined) {
      gcPersistentSignals.delete(signal);
      abortSignal(
        signal,
        new DOMException(
          'The operation was aborted due to timeout',
          'TimeoutError'));
    }
  }, delay);
  holder.timeout.unref();
  return holder;
}

// The default reason captures no stack. A captured stack keeps the receivers
// and functions of the aborting caller's frames alive until it is formatted,
// so a retained signal would keep the aborted work alive, and formatting it
// eagerly costs several microseconds per abort. Callers that want the stack
// can pass their own reason.
// Refs: https://github.com/nodejs/node/issues/66192
function createDefaultAbortReason() {
  if (!isErrorStackTraceLimitWritable()) {
    return new DOMException('This operation was aborted', 'AbortError');
  }
  const limit = Error.stackTraceLimit;
  Error.stackTraceLimit = 0;
  const reason = new DOMException('This operation was aborted', 'AbortError');
  Error.stackTraceLimit = limit;
  return reason;
}

class AbortSignal extends EventTarget {
  #brand;

  static {
    converters.AbortSignal = createInterfaceConverter(
      'AbortSignal',
      (value) => typeof value === 'object' && value !== null && #brand in value,
    );
  }

  /**
   * Takes positional arguments rather than an options object: destructuring
   * the default kEmptyObject, which V8 keeps in dictionary mode, cost more
   * than the rest of the constructor.
   * @param {symbol | undefined} dontThrowSymbol
   * @param {boolean} [aborted]
   * @param {any} [reason]
   * @param {boolean} [transferable]
   * @param {boolean} [composite]
   * @private
   */
  constructor(dontThrowSymbol = undefined, aborted = false, reason = undefined,
              transferable = false, composite = false) {
    if (dontThrowSymbol !== kDontThrowSymbol) {
      throw new ERR_ILLEGAL_CONSTRUCTOR();
    }
    super();

    this[kMaxEventTargetListeners] = 0;
    this[kAborted] = aborted;
    this[kReason] = reason;
    this[kComposite] = composite;
    // Signals that become sources of AbortSignal.any() get these slots later.
    // Initialize them up front so that adding them doesn't change the shape
    // of live signals and make the call sites that use them polymorphic.
    this[kWeakRef] = undefined;
    this[kDependantSignals] = undefined;
    if (composite) {
      this[kTimeout] = false;
      this[kFollowing] = false;
      this[kSourceSignals] = undefined;
    }
    if (transferable) {
      markTransferMode(this, false, true);
    }
  }

  /**
   * @type {boolean}
   */
  get aborted() {
    validateThisAbortSignal(this);
    refreshCompositeSignal(this);
    return !!this[kAborted];
  }

  /**
   * @type {any}
   */
  get reason() {
    validateThisAbortSignal(this);
    refreshCompositeSignal(this);
    return this[kReason];
  }

  throwIfAborted() {
    validateThisAbortSignal(this);
    refreshCompositeSignal(this);
    if (this[kAborted]) {
      throw this[kReason];
    }
  }

  [customInspectSymbol](depth, options) {
    return customInspect(this, {
      aborted: this.aborted,
    }, depth, options);
  }

  /**
   * @param {any} [reason]
   * @returns {AbortSignal}
   */
  static abort(reason = createDefaultAbortReason()) {
    return new AbortSignal(kDontThrowSymbol, true, reason);
  }

  /**
   * @param {number} delay
   * @returns {AbortSignal}
   */
  static timeout(delay) {
    validateUint32(delay, 'delay', false);
    const signal = new AbortSignal(kDontThrowSymbol);
    signal[kTimeout] = true;
    clearTimeoutRegistry.register(
      signal,
      setWeakAbortSignalTimeout(getWeakRef(signal), delay));
    return signal;
  }

  /**
   * @param {AbortSignal[]} signals
   * @returns {AbortSignal}
   */
  static any(signals) {
    const signalsArray = converters['sequence<AbortSignal>'](
      signals,
      { __proto__: null, context: 'signals' },
    );

    const resultSignal = new AbortSignal(kDontThrowSymbol, false, undefined, false, true);
    if (!signalsArray.length) {
      return resultSignal;
    }

    resultSignal[kSourceSignals] = new SafeSet();

    // Track if we have any timeout signals
    let hasTimeoutSignals = false;

    for (let i = 0; i < signalsArray.length; i++) {
      const signal = signalsArray[i];

      // Check if this is a timeout signal
      if (signal[kTimeout]) {
        hasTimeoutSignals = true;

        // Add the timeout signal to gcPersistentSignals to keep it alive
        // This is what the kNewListener method would do when adding abort listeners
        gcPersistentSignals.add(signal);
      }

      refreshCompositeSignal(signal);
      if (signal[kAborted]) {
        abortSignal(resultSignal, signal[kReason]);
        return resultSignal;
      }

      if (!signal[kComposite]) {
        resultSignal[kSourceSignals].add(getWeakRef(signal));
      } else if (!signal[kSourceSignals]) {
        continue;
      } else {
        for (const sourceSignalWeakRef of signal[kSourceSignals]) {
          const sourceSignal = sourceSignalWeakRef.deref();
          if (!sourceSignal) {
            continue;
          }
          assert(!sourceSignal[kComposite]);

          if (sourceSignal[kAborted]) {
            abortSignal(resultSignal, sourceSignal[kReason]);
            return resultSignal;
          }

          if (resultSignal[kSourceSignals].has(sourceSignalWeakRef)) {
            continue;
          }
          resultSignal[kSourceSignals].add(sourceSignalWeakRef);
        }
      }
    }

    if (hasTimeoutSignals && resultSignal[kSourceSignals].size > 0) {
      resultSignal[kTimeout] = true;
    }

    // Preserve abort state even if sources are collected before this signal is
    // observed. Following uses weak references so unused signals can be collected.
    followCompositeSignal(resultSignal);

    return resultSignal;
  }

  [kNewListener](size, type, listener, once, capture, passive, weak) {
    FunctionPrototypeCall(eventTargetNewListener, this, size, type, listener, once, capture, passive, weak);

    if (this[kComposite] && type === 'abort' && !this.aborted && size === 1) {
      followCompositeSignal(this);
    }

    const isTimeoutOrNonEmptyCompositeSignal = this[kTimeout] || (this[kComposite] && this[kSourceSignals]?.size);
    if (isTimeoutOrNonEmptyCompositeSignal &&
        type === 'abort' &&
        !this.aborted &&
        !weak &&
        size === 1) {
      // If this is a timeout signal, or a non-empty composite signal, and we're adding a non-weak abort
      // listener, then we don't want it to be gc'd while the listener
      // is attached and the timer still hasn't fired. So, we retain a
      // strong ref that is held for as long as the listener is registered.
      gcPersistentSignals.add(this);
    }
  }

  [kRemoveListener](size, type, listener, capture) {
    FunctionPrototypeCall(eventTargetRemoveListener, this, size, type, listener, capture);
    const isTimeoutOrNonEmptyCompositeSignal = this[kTimeout] || (this[kComposite] && this[kSourceSignals]?.size);
    if (isTimeoutOrNonEmptyCompositeSignal && type === 'abort' && size === 0) {
      gcPersistentSignals.delete(this);
    }
  }

  [kTransfer]() {
    validateThisAbortSignal(this);
    const aborted = this.aborted;
    if (aborted) {
      const reason = this.reason;
      return {
        data: { aborted, reason },
        deserializeInfo: 'internal/abort_controller:ClonedAbortSignal',
      };
    }

    const { port1, port2 } = this[kCloneData];
    this[kCloneData] = undefined;

    this.addEventListener('abort', () => {
      port1.postMessage(this.reason);
      port1.close();
    }, { once: true });

    return {
      data: { port: port2 },
      deserializeInfo: 'internal/abort_controller:ClonedAbortSignal',
    };
  }

  [kTransferList]() {
    if (!this.aborted) {
      const { port1, port2 } = lazyMessageChannel();
      port1.unref();
      port2.unref();
      this[kCloneData] = {
        port1,
        port2,
      };
      return [port2];
    }
    return [];
  }

  [kDeserialize]({ aborted, reason, port }) {
    if (aborted) {
      this[kAborted] = aborted;
      this[kReason] = reason;
      return;
    }

    port.onmessage = ({ data }) => {
      abortSignal(this, data);
      port.close();
      port.onmessage = undefined;
    };
    // The receiving port, by itself, should never keep the event loop open.
    // The unref() has to be called *after* setting the onmessage handler.
    port.unref();
  }
}

converters['sequence<AbortSignal>'] = createSequenceConverter(converters.AbortSignal);

function ClonedAbortSignal() {
  return new AbortSignal(kDontThrowSymbol, false, undefined, true);
}
ClonedAbortSignal.prototype[kDeserialize] = () => {};

ObjectDefineProperties(AbortSignal.prototype, {
  aborted: kEnumerableProperty,
});

ObjectDefineProperty(AbortSignal.prototype, SymbolToStringTag, {
  __proto__: null,
  writable: false,
  enumerable: false,
  configurable: true,
  value: 'AbortSignal',
});

defineEventHandler(AbortSignal.prototype, 'abort');

// https://dom.spec.whatwg.org/#dom-abortsignal-abort
function abortSignal(signal, reason) {
  // 1. If signal is aborted, then return.
  if (signal[kAborted]) return;

  // 2. Set signal's abort reason to reason if it is given;
  //    otherwise to a new "AbortError" DOMException.
  signal[kAborted] = true;
  signal[kReason] = reason;

  // 3. Let dependentSignalsToAbort be a new list.
  // The list is only allocated when there are dependent signals.
  let dependentSignalsToAbort;

  // 4. For each dependentSignal of signal's dependent signals:
  const dependantSignals = signal[kDependantSignals];
  if (dependantSignals !== undefined && dependantSignals.size > 0) {
    dependentSignalsToAbort = ObjectSetPrototypeOf([], null);
    dependantSignals.forEach((s) => {
      const dependentSignal = s.deref();
      // 1. If dependentSignal is not aborted, then:
      if (dependentSignal && !dependentSignal[kAborted]) {
        // 1. Set dependentSignal's abort reason to signal's abort reason.
        dependentSignal[kReason] = reason;
        dependentSignal[kAborted] = true;
        // 2. Append dependentSignal to dependentSignalsToAbort.
        ArrayPrototypePush(dependentSignalsToAbort, dependentSignal);
      }
    });
  }

  // 5. Run the abort steps for signal
  runAbort(signal);

  // 6. For each dependentSignal of dependentSignalsToAbort,
  //    run the abort steps for dependentSignal.
  if (dependentSignalsToAbort !== undefined) {
    for (let i = 0; i < dependentSignalsToAbort.length; i++) {
      const dependentSignal = dependentSignalsToAbort[i];
      runAbort(dependentSignal);
      // A transitively-aborted dependent signal can never abort again, so there
      // is no longer any reason to keep it alive. Dropping it from
      // gcPersistentSignals lets it be collected, which in turn prunes its
      // WeakRef from its sources' kDependantSignals sets. Otherwise, an observed
      // composite that follows a long-lived source stays retained forever and
      // its entry accumulates in that source's kDependantSignals.
      gcPersistentSignals.delete(dependentSignal);
    }
  }

  // Clean up the signal from gcPersistentSignals
  gcPersistentSignals.delete(signal);

  // If this is a composite signal, also remove all of its source signals from gcPersistentSignals
  // when they get dereferenced from the signal's kSourceSignals set
  if (signal[kComposite] && signal[kSourceSignals]) {
    signal[kSourceSignals].forEach((sourceWeakRef) => {
      const sourceSignal = sourceWeakRef.deref();
      if (sourceSignal) {
        gcPersistentSignals.delete(sourceSignal);
      }
    });
  }
}

// To run the abort steps for an AbortSignal signal
function runAbort(signal) {
  // Firing an event at a signal without abort listeners has no observable
  // effect, so skip allocating and dispatching it.
  const root = signal[kEvents]?.get('abort');
  if (root === undefined || root.next === undefined) {
    return;
  }
  const event = new Event('abort', kTrustedEventInit);
  signal.dispatchEvent(event);
}

class AbortController {
  #signal;

  /**
   * @type {AbortSignal}
   */
  get signal() {
    this.#signal ??= new AbortSignal(kDontThrowSymbol);

    return this.#signal;
  }

  /**
   * @param {any} [reason]
   */
  abort(reason = createDefaultAbortReason()) {
    abortSignal(this.#signal ??= new AbortSignal(kDontThrowSymbol), reason);
  }

  [customInspectSymbol](depth, options) {
    return customInspect(this, {
      signal: this.signal,
    }, depth, options);
  }

  static [kMakeTransferable]() {
    const controller = new AbortController();
    controller.#signal = new AbortSignal(kDontThrowSymbol, false, undefined, true);
    return controller;
  }
}

/**
 * Enables the AbortSignal to be transferable using structuredClone/postMessage.
 * @param {AbortSignal} signal
 * @returns {AbortSignal}
 */
function transferableAbortSignal(signal) {
  if (signal?.[kAborted] === undefined)
    throw new ERR_INVALID_ARG_TYPE('signal', 'AbortSignal', signal);
  markTransferMode(signal, false, true);
  return signal;
}

/**
 * Creates an AbortController with a transferable AbortSignal
 * @returns {AbortController}
 */
function transferableAbortController() {
  return AbortController[kMakeTransferable]();
}

/**
 * @param {AbortSignal} signal
 * @param {any} resource
 * @returns {Promise<void>}
 */
async function aborted(signal, resource) {
  converters.AbortSignal(signal, { __proto__: null, context: 'signal' });
  validateObject(resource, 'resource', kValidateObjectAllowObjects);
  if (signal.aborted)
    return PromiseResolve();
  const abortPromise = PromiseWithResolvers();
  const opts = { __proto__: null, [kWeakHandler]: resource, once: true, [kResistStopPropagation]: true };
  signal.addEventListener('abort', abortPromise.resolve, opts);
  return abortPromise.promise;
}

ObjectDefineProperties(AbortController.prototype, {
  signal: kEnumerableProperty,
  abort: kEnumerableProperty,
});

ObjectDefineProperty(AbortController.prototype, SymbolToStringTag, {
  __proto__: null,
  writable: false,
  enumerable: false,
  configurable: true,
  value: 'AbortController',
});

module.exports = {
  AbortController,
  AbortSignal,
  ClonedAbortSignal,
  abortSignal,
  aborted,
  transferableAbortSignal,
  transferableAbortController,
};