All files / async / unstable_circuit_breaker.ts

// Copyright 2018-2026 the Deno authors. MIT license.
// This module is browser compatible.

/**
 * Circuit breaker states following the standard pattern.
 * - `"closed"`: Normal operation, requests pass through
 * - `"open"`: Failing, all requests rejected immediately
 * - `"half_open"`: Testing recovery, limited requests allowed
 */
export type CircuitState = "closed" | "open" | "half_open";

/** Options for {@linkcode CircuitBreaker}. */
export interface CircuitBreakerOptions<T> {
  /**
   * Number of failures before opening the circuit.
   *
   * Note: For high-volume services, a low absolute threshold may cause
   * premature circuit opening during normal transient errors. Consider
   * a higher value proportional to your request volume, or combine with
   * a shorter {@linkcode failureWindowMs} to implement rate-based detection.
   *
   * @default {5}
   */
  failureThreshold?: number;

  /**
   * Duration in milliseconds the circuit stays open before entering half-open.
   *
   * @default {30000}
   */
  cooldownMs?: number;

  /**
   * Number of consecutive successes needed to close from half-open state.
   *
   * @default {2}
   */
  successThreshold?: number;

  /**
   * Maximum concurrent requests allowed in half-open state.
   *
   * @default {3}
   */
  halfOpenMaxConcurrent?: number;

  /**
   * Time window in milliseconds for counting failures.
   * Failures older than this are forgotten (sliding window).
   * Set to `0` to disable decay.
   *
   * @default {60000}
   */
  failureWindowMs?: number;

  /**
   * Custom predicate to determine if an error should count as a circuit failure.
   * By default, any thrown error is a failure.
   *
   * @param error The error that was thrown.
   * @returns `true` if the error should count as a circuit failure.
   */
  isFailure?: (error: unknown) => boolean;

  /**
   * Custom predicate to determine if a successful result should count as failure.
   * Useful for HTTP responses where status codes indicate logical failures.
   *
   * @param result The successful result.
   * @returns `true` if the result should count as a circuit failure.
   */
  isResultFailure?: (result: T) => boolean;

  /**
   * Callback invoked when circuit state changes.
   * Must not throw.
   *
   * @param from Previous state.
   * @param to New state.
   */
  onStateChange?: (from: CircuitState, to: CircuitState) => void;

  /**
   * Callback invoked when a failure is recorded.
   * Must not throw.
   *
   * @param error The error that caused the failure.
   * @param failureCount Current number of failures in the window.
   */
  onFailure?: (error: unknown, failureCount: number) => void;

  /**
   * Callback invoked when circuit opens.
   * Must not throw.
   *
   * @param failureCount Number of failures that triggered the open.
   */
  onOpen?: (failureCount: number) => void;

  /**
   * Callback invoked when circuit enters half-open state (testing recovery).
   * Must not throw.
   */
  onHalfOpen?: () => void;

  /**
   * Callback invoked when circuit closes (recovery complete).
   * Must not throw.
   */
  onClose?: () => void;
}

/** Options for {@linkcode CircuitBreaker.execute}. */
export interface CircuitBreakerExecuteOptions {
  /**
   * An optional abort signal that can be used to cancel the operation
   * before it starts. If the signal is already aborted when `execute` is
   * called, the operation will fail immediately without executing the function.
   *
   * Note: This only checks the abort status before execution. It does not
   * interrupt an in-progress operation — pass the signal to your async
   * function for that behavior.
   */
  signal?: AbortSignal;
}

/**
 * Error thrown when {@linkcode CircuitBreaker} is open and rejects a request.
 *
 * @experimental **UNSTABLE**: New API, yet to be vetted.
 *
 * @example Usage
 * ```ts
 * import { CircuitBreaker, CircuitBreakerOpenError } from "@std/async/unstable-circuit-breaker";
 * import { assert } from "@std/assert";
 *
 * const breaker = new CircuitBreaker({ failureThreshold: 1 });
 *
 * // Trigger a failure to open the circuit
 * try {
 *   await breaker.execute(() => Promise.reject(new Error("fail")));
 * } catch (_) {
 *   // Expected to fail
 * }
 *
 * // Now the circuit is open
 * try {
 *   await breaker.execute(() => Promise.resolve("ok"));
 * } catch (error) {
 *   assert(error instanceof CircuitBreakerOpenError);
 *   assert(error.remainingCooldownMs >= 0);
 * }
 * ```
 */
export class CircuitBreakerOpenError extends Error {
  /**
   * Milliseconds until the circuit breaker cooldown expires.
   *
   * @example Usage
   * ```ts
   * import { CircuitBreakerOpenError } from "@std/async/unstable-circuit-breaker";
   * import { assertEquals } from "@std/assert";
   *
   * const error = new CircuitBreakerOpenError(5000);
   * assertEquals(error.remainingCooldownMs, 5000);
   * ```
   */
  readonly remainingCooldownMs: number;

  /**
   * Constructs a new {@linkcode CircuitBreakerOpenError} instance.
   *
   * @param remainingCooldownMs Milliseconds until cooldown expires.
   */
  constructor(remainingCooldownMs: number) {
    super(
      `Circuit breaker is open. Retry after ${remainingCooldownMs}ms.`,
    );
    this.name = "CircuitBreakerOpenError";
    this.remainingCooldownMs = remainingCooldownMs;
  }
}

/** Base properties shared by all circuit breaker states. */
interface CircuitBreakerStateBase {
  /** Failure timestamps in milliseconds since epoch. */
  readonly failureTimestamps: readonly number[];
  readonly consecutiveSuccesses: number;
  readonly halfOpenInFlight: number;
}

/** Internal state managed by the circuit breaker */
type CircuitBreakerState =
  | (CircuitBreakerStateBase & {
    readonly state: "closed";
    /** Timestamp when circuit opened, in milliseconds since epoch. */
    readonly openedAt: null;
  })
  | (CircuitBreakerStateBase & {
    readonly state: "open";
    /** Timestamp when circuit opened, in milliseconds since epoch. */
    readonly openedAt: number;
  })
  | (CircuitBreakerStateBase & {
    readonly state: "half_open";
    /** Timestamp when circuit opened, in milliseconds since epoch. */
    readonly openedAt: number;
  });

/** Creates initial circuit breaker state. */
function createInitialState(): CircuitBreakerState & { state: "closed" } {
  return {
    state: "closed",
    failureTimestamps: [],
    consecutiveSuccesses: 0,
    openedAt: null,
    halfOpenInFlight: 0,
  };
}

/**
 * Removes failure timestamps outside the decay window.
 *
 * @param timestamps Readonly array of failure timestamps in ms.
 * @param windowMs Duration window in milliseconds.
 * @param nowMs Current time in milliseconds.
 * @returns Readonly filtered array of timestamps within the window.
 */
function pruneOldFailures(
  timestamps: readonly number[],
  windowMs: number,
  nowMs: number,
): readonly number[] {
  if (windowMs === 0) return timestamps;
  const cutoff = nowMs - windowMs;
  return timestamps.filter((ts) => ts > cutoff);
}

/** Validates a numeric option for the circuit breaker constructor. */
function validateOption(name: string, value: number, min: number): void {
  if (!Number.isFinite(value) || value < min) {
    const constraint = min === 0
      ? "a finite non-negative number"
      : `a finite number >= ${min}`;
    throw new RangeError(
      `Cannot create circuit breaker as '${name}' must be ${constraint}: received ${value}`,
    );
  }
}

/**
 * A circuit breaker that wraps async operations to prevent cascading failures.
 *
 * The circuit breaker monitors for failures and "trips" (opens) when a threshold
 * is reached, rejecting subsequent requests immediately without executing them.
 * After a cooldown period, it enters a "half-open" state to test if the service
 * has recovered.
 *
 * @experimental **UNSTABLE**: New API, yet to be vetted.
 *
 * @example Usage
 * ```ts
 * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
 * import { assertEquals } from "@std/assert";
 *
 * const breaker = new CircuitBreaker({
 *   failureThreshold: 5,
 *   cooldownMs: 30000,
 * });
 *
 * assertEquals(breaker.state, "closed");
 *
 * const result = await breaker.execute(() => Promise.resolve("success"));
 * assertEquals(result, "success");
 * ```
 *
 * @example Handling open circuit
 * ```ts
 * import { CircuitBreaker, CircuitBreakerOpenError } from "@std/async/unstable-circuit-breaker";
 *
 * const breaker = new CircuitBreaker({ failureThreshold: 5 });
 *
 * try {
 *   const result = await breaker.execute(() => fetch("https://api.example.com"));
 * } catch (error) {
 *   if (error instanceof CircuitBreakerOpenError) {
 *     console.log(`Service unavailable, retry in ${error.remainingCooldownMs}ms`);
 *   }
 * }
 * ```
 *
 * @example With custom failure detection
 * ```ts no-assert
 * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
 *
 * const breaker = new CircuitBreaker<Response>({
 *   failureThreshold: 3,
 *   // Only count server errors as circuit failures
 *   isResultFailure: (response) => response.status >= 500,
 *   onStateChange: (from, to) => console.log(`Circuit: ${from} → ${to}`),
 * });
 * ```
 *
 * @example Composing with retry and AbortSignal
 * ```ts ignore
 * import { retry } from "@std/async/retry";
 * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
 *
 * const breaker = new CircuitBreaker({ failureThreshold: 5 });
 *
 * // Timeout applies to the entire operation (circuit breaker + retries)
 * const signal = AbortSignal.timeout(5000);
 *
 * const result = await breaker.execute(
 *   () => retry(() => fetch("https://api.example.com"), { signal }),
 *   { signal },
 * );
 * ```
 *
 * @example Waiting out the cooldown with delay
 * ```ts ignore
 * import { delay } from "@std/async/delay";
 * import { CircuitBreaker, CircuitBreakerOpenError } from "@std/async/unstable-circuit-breaker";
 *
 * const breaker = new CircuitBreaker({ failureThreshold: 5 });
 *
 * try {
 *   return await breaker.execute(() => fetch("https://api.example.com"));
 * } catch (error) {
 *   if (error instanceof CircuitBreakerOpenError) {
 *     // Wait for the circuit to transition to half-open, then retry
 *     await delay(error.remainingCooldownMs);
 *     return await breaker.execute(() => fetch("https://api.example.com"));
 *   }
 *   throw error;
 * }
 * ```
 *
 * @typeParam T The type of value returned by the executed function.
 */
export class CircuitBreaker<T = unknown> {
  #failureThreshold: number;
  #cooldownMs: number;
  #successThreshold: number;
  #halfOpenMaxConcurrent: number;
  #failureWindowMs: number;
  #isFailure: (error: unknown) => boolean;
  #isResultFailure: (result: T) => boolean;
  #onStateChange: ((from: CircuitState, to: CircuitState) => void) | undefined;
  #onFailure: ((error: unknown, failureCount: number) => void) | undefined;
  #onOpen: ((failureCount: number) => void) | undefined;
  #onHalfOpen: (() => void) | undefined;
  #onClose: (() => void) | undefined;
  #state: CircuitBreakerState;

  /**
   * Constructs a new {@linkcode CircuitBreaker} instance.
   *
   * @param options Configuration options for the circuit breaker.
   * @throws {RangeError} If any numeric option is not a finite number within its valid range.
   */
  constructor(options: CircuitBreakerOptions<T> = {}) {
    const {
      failureThreshold = 5,
      cooldownMs = 30_000,
      successThreshold = 2,
      halfOpenMaxConcurrent = 3,
      failureWindowMs = 60_000,
      isFailure = () => true,
      isResultFailure = () => false,
      onStateChange,
      onFailure,
      onOpen,
      onHalfOpen,
      onClose,
    } = options;

    validateOption("failureThreshold", failureThreshold, 1);
    validateOption("cooldownMs", cooldownMs, 0);
    validateOption("successThreshold", successThreshold, 1);
    validateOption("halfOpenMaxConcurrent", halfOpenMaxConcurrent, 1);
    validateOption("failureWindowMs", failureWindowMs, 0);

    this.#failureThreshold = failureThreshold;
    this.#cooldownMs = cooldownMs;
    this.#successThreshold = successThreshold;
    this.#halfOpenMaxConcurrent = halfOpenMaxConcurrent;
    this.#failureWindowMs = failureWindowMs;
    this.#isFailure = isFailure;
    this.#isResultFailure = isResultFailure;
    this.#onStateChange = onStateChange;
    this.#onFailure = onFailure;
    this.#onOpen = onOpen;
    this.#onHalfOpen = onHalfOpen;
    this.#onClose = onClose;
    this.#state = createInitialState();
  }

  /**
   * Current stored state of the circuit breaker.
   *
   * Note: This returns the stored state without resolving time-based
   * transitions. After a cooldown expires, this may still show `"open"`
   * until the next {@linkcode execute} call triggers the transition to
   * `"half_open"`.
   *
   * @example Usage
   * ```ts
   * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
   * import { assertEquals } from "@std/assert";
   *
   * const breaker = new CircuitBreaker();
   * assertEquals(breaker.state, "closed");
   * ```
   *
   * @returns The stored {@linkcode CircuitState}.
   */
  get state(): CircuitState {
    return this.#state.state;
  }

  /**
   * Executes a function through the circuit breaker.
   *
   * The function can be synchronous or asynchronous. The result is always
   * returned as a promise.
   *
   * @example Usage with async function
   * ```ts ignore
   * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
   *
   * const breaker = new CircuitBreaker({ failureThreshold: 5 });
   * const response = await breaker.execute(() => fetch("https://api.example.com"));
   * ```
   *
   * @example With timeout
   * ```ts ignore
   * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
   *
   * const breaker = new CircuitBreaker({ failureThreshold: 5 });
   *
   * // Abort if operation takes longer than 5 seconds
   * const response = await breaker.execute(
   *   () => fetch("https://api.example.com"),
   *   { signal: AbortSignal.timeout(5000) },
   * );
   * ```
   *
   * @typeParam R The return type of the function, must extend T.
   * @param fn The function to execute (sync or async).
   * @param options Optional execution options including an abort signal.
   * @returns A promise that resolves to the result of the operation.
   * @throws {CircuitBreakerOpenError} If circuit is open.
   * @throws {DOMException} If the abort signal is already aborted.
   */
  async execute<R extends T>(
    fn: () => R | Promise<R>,
    options?: CircuitBreakerExecuteOptions,
  ): Promise<R> {
    options?.signal?.throwIfAborted();

    const currentTime = Date.now();
    const currentState = this.#resolveCurrentState(currentTime);

    // Check if we should reject
    if (currentState.state === "open") {
      const cooldownEnd = currentState.openedAt + this.#cooldownMs;
      const remainingMs = Math.max(0, cooldownEnd - currentTime);
      throw new CircuitBreakerOpenError(Math.round(remainingMs));
    }

    // In half-open, check concurrent limit
    if (currentState.state === "half_open") {
      if (currentState.halfOpenInFlight >= this.#halfOpenMaxConcurrent) {
        throw new CircuitBreakerOpenError(0);
      }
      this.#state = {
        ...this.#state,
        halfOpenInFlight: this.#state.halfOpenInFlight + 1,
      };
    }

    let result: R;
    try {
      result = await fn();
    } catch (error) {
      this.#handleFailure(error, currentState.state, currentTime);
      throw error;
    } finally {
      // Decrement half-open in-flight counter
      if (currentState.state === "half_open") {
        this.#state = {
          ...this.#state,
          halfOpenInFlight: Math.max(0, this.#state.halfOpenInFlight - 1),
        };
      }
    }

    // Check if successful result should count as failure
    if (this.#isResultFailure(result)) {
      const syntheticError = new Error("Result classified as failure");
      this.#handleFailure(syntheticError, currentState.state, currentTime);
      return result; // Still return the result, but record the failure
    }

    // Success path
    this.#handleSuccess(currentState.state);
    return result;
  }

  /**
   * Forces the circuit breaker to open state.
   * Useful for maintenance or known outages.
   *
   * @example Usage
   * ```ts
   * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
   * import { assertEquals } from "@std/assert";
   *
   * const breaker = new CircuitBreaker();
   * breaker.forceOpen();
   * assertEquals(breaker.state, "open");
   * ```
   */
  forceOpen(): void {
    const previous = this.#state.state;
    const failureTimestamps = this.#state.failureTimestamps;
    const now = Date.now();
    this.#state = {
      ...this.#state,
      state: "open",
      openedAt: now,
      consecutiveSuccesses: 0,
    };
    if (previous !== "open") {
      this.#onStateChange?.(previous, "open");
      const failureCount = pruneOldFailures(
        failureTimestamps,
        this.#failureWindowMs,
        now,
      ).length;
      this.#onOpen?.(failureCount);
    }
  }

  /**
   * Forces the circuit breaker to closed state and notifies observers.
   *
   * This is an operational transition that fires {@linkcode onStateChange}
   * and {@linkcode onClose} callbacks. Use this when the protected service
   * has recovered and you want observers to be notified.
   *
   * For silent resets (e.g., in tests), use {@linkcode reset} instead.
   *
   * @example Usage
   * ```ts
   * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
   * import { assertEquals } from "@std/assert";
   *
   * const breaker = new CircuitBreaker();
   * breaker.forceOpen();
   * breaker.forceClose();
   * assertEquals(breaker.state, "closed");
   * ```
   */
  forceClose(): void {
    const previous = this.#state.state;
    this.#state = createInitialState();
    if (previous !== "closed") {
      this.#onStateChange?.(previous, "closed");
      this.#onClose?.();
    }
  }

  /**
   * Silently resets the circuit breaker to initial state.
   *
   * Unlike {@linkcode forceClose}, this does not fire any callbacks
   * (`onStateChange`, `onClose`). Use this for testing or administrative
   * resets where observers should not be notified.
   *
   * @example Usage
   * ```ts
   * import { CircuitBreaker } from "@std/async/unstable-circuit-breaker";
   * import { assertEquals } from "@std/assert";
   *
   * const breaker = new CircuitBreaker();
   * breaker.reset();
   * assertEquals(breaker.state, "closed");
   * ```
   */
  reset(): void {
    this.#state = createInitialState();
  }

  /**
   * Resolves the current state, handling automatic transitions.
   * OPEN → HALF_OPEN after cooldown expires.
   *
   * @param now Current timestamp in milliseconds.
   */
  #resolveCurrentState(now: number): CircuitBreakerState {
    if (this.#state.state !== "open") {
      return this.#state;
    }

    const cooldownEnd = this.#state.openedAt + this.#cooldownMs;
    if (now < cooldownEnd) {
      return this.#state;
    }

    // Auto-transition from OPEN to HALF_OPEN after cooldown
    this.#state = {
      ...this.#state,
      state: "half_open",
      consecutiveSuccesses: 0,
      halfOpenInFlight: 0,
    };
    this.#onStateChange?.("open", "half_open");
    this.#onHalfOpen?.();
    return this.#state;
  }

  /**
   * Records a failure and potentially opens the circuit.
   *
   * @param error The error that occurred.
   * @param previousState The state before this failure.
   * @param now Current timestamp in milliseconds.
   */
  #handleFailure(
    error: unknown,
    previousState: CircuitState,
    now: number,
  ): void {
    if (!this.#isFailure(error)) return;

    const newFailures = [
      ...pruneOldFailures(
        this.#state.failureTimestamps,
        this.#failureWindowMs,
        now,
      ),
      now,
    ];

    const shouldOpen = previousState === "half_open" ||
      newFailures.length >= this.#failureThreshold;

    // State mutations first (before any callbacks)
    if (shouldOpen) {
      this.#state = {
        ...this.#state,
        state: "open",
        failureTimestamps: newFailures,
        openedAt: now,
        consecutiveSuccesses: 0,
      };
    } else {
      this.#state = {
        ...this.#state,
        failureTimestamps: newFailures,
        consecutiveSuccesses: 0,
      };
    }

    // Callbacks last (state is consistent even if these throw)
    this.#onFailure?.(error, newFailures.length);
    if (shouldOpen) {
      this.#onStateChange?.(previousState, "open");
      this.#onOpen?.(newFailures.length);
    }
  }

  /** Records a success and potentially closes the circuit from half-open. */
  #handleSuccess(previousState: CircuitState): void {
    if (previousState === "closed") return;

    const newSuccessCount = this.#state.consecutiveSuccesses + 1;
    if (newSuccessCount >= this.#successThreshold) {
      this.#state = createInitialState();
      this.#onStateChange?.("half_open", "closed");
      this.#onClose?.();
      return;
    }

    this.#state = { ...this.#state, consecutiveSuccesses: newSuccessCount };
  }
}