@W-23735204 Add circuit breaker around data store reads

.changeset/data-store-circuit-breaker.md

@@ -0,0 +1,5 @@
+---
+'@salesforce/mrt-utilities': minor
+---
+
+Add an in-memory circuit breaker around data store reads. When DynamoDB is failing or throttling, each warm container trips the breaker and fails fast for a short cooldown instead of piling load onto a saturated table — letting the client's application-level fallback serve reads — then probes to recover. State is per-container and resets on cold start. Set `MRT_DATA_STORE_CIRCUIT_BREAKER_DISABLED=true` to disable it.

packages/mrt-utilities/src/data-store/circuit-breaker.ts

@@ -0,0 +1,200 @@
+/*
+ * Copyright (c) 2025, Salesforce, Inc.
+ * SPDX-License-Identifier: Apache-2
+ * For full license text, see the license.txt file in the repo root or http://www.apache.org/licenses/LICENSE-2.0
+ */
+
+/**
+ * Circuit-breaker states.
+ *
+ * - `closed`: requests flow through; failures are counted toward the trip threshold.
+ * - `open`: requests fail fast without touching the backend, for a cooldown window.
+ * - `half-open`: a limited number of probe requests are allowed; success closes the
+ *   breaker, any failure re-opens it.
+ */
+export type CircuitBreakerState = 'closed' | 'open' | 'half-open';
+
+/**
+ * A state transition, emitted for observability.
+ */
+export interface CircuitBreakerTransition {
+  from: CircuitBreakerState;
+  to: CircuitBreakerState;
+  /** The reason the transition happened, for the telemetry message. */
+  reason: string;
+}
+
+/**
+ * Tuning for the circuit breaker. All fields are engineering-tuned defaults; they are
+ * internal constants at the call site (only an on/off kill switch is env-configurable), but
+ * kept injectable so tests can drive transitions deterministically.
+ */
+export interface CircuitBreakerOptions {
+  /**
+   * Failure weight (in "points") needed to trip the breaker from closed to open. A plain
+   * failure contributes 1 point; a throttling failure contributes {@link throttleWeight}.
+   */
+  failureThreshold: number;
+  /**
+   * Points contributed by a throttling failure. Weighted heavier than a plain failure
+   * because throttles are the signal this breaker exists to shed load for.
+   */
+  throttleWeight: number;
+  /** How long (ms) the breaker stays open before allowing a half-open probe. */
+  cooldownMs: number;
+  /** Number of consecutive successful probes in half-open required to close the breaker. */
+  halfOpenProbes: number;
+  /** Injectable clock (ms epoch) for deterministic tests. Defaults to `Date.now`. */
+  now?: () => number;
+  /** Callback invoked on every state transition, for telemetry. */
+  onTransition?: (transition: CircuitBreakerTransition) => void;
+}
+
+/**
+ * A per-instance, in-memory circuit breaker.
+ *
+ * Load-shedding for a single warm execution environment — state is NOT shared across the
+ * fleet and does not survive a cold start (which begins `closed`). This is intentional: it
+ * is a local guard that stops a container from hammering an already-saturated backend, not a
+ * coordinated fleet-wide switch.
+ *
+ * The breaker is policy-only: it decides whether a call may proceed ({@link canRequest}) and
+ * records outcomes ({@link recordSuccess} / {@link recordFailure}). It never calls the
+ * backend itself, so it is trivially testable and reusable.
+ */
+export class CircuitBreaker {
+  private _state: CircuitBreakerState = 'closed';
+  /** Accumulated failure points while closed. */
+  private _failureScore = 0;
+  /** Successful probes recorded while half-open. */
+  private _probeSuccesses = 0;
+  /** Probes admitted but not yet resolved while half-open (concurrency budget). */
+  private _probesInFlight = 0;
+  /** Epoch ms when the breaker opened; used to time the cooldown. */
+  private _openedAt = 0;
+
+  private readonly _failureThreshold: number;
+  private readonly _throttleWeight: number;
+  private readonly _cooldownMs: number;
+  private readonly _halfOpenProbes: number;
+  private readonly _now: () => number;
+  private readonly _onTransition?: (transition: CircuitBreakerTransition) => void;
+
+  constructor(options: CircuitBreakerOptions) {
+    this._failureThreshold = options.failureThreshold;
+    this._throttleWeight = options.throttleWeight;
+    this._cooldownMs = options.cooldownMs;
+    this._halfOpenProbes = options.halfOpenProbes;
+    this._now = options.now ?? Date.now;
+    this._onTransition = options.onTransition;
+  }
+
+  /**
+   * The current state. This is a pure read and never advances the machine — the open→half-open
+   * transition happens only when a caller asks to proceed via {@link canRequest}.
+   */
+  get state(): CircuitBreakerState {
+    return this._state;
+  }
+
+  /**
+   * Whether a request may proceed to the backend right now.
+   *
+   * - `closed`: always true.
+   * - `open`: false until the cooldown elapses, then transitions to half-open and returns
+   *   true to admit the first probe.
+   * - `half-open`: true only while fewer than {@link CircuitBreakerOptions.halfOpenProbes}
+   *   probes are in flight — the caller must report each admitted probe's outcome via
+   *   {@link recordSuccess} / {@link recordFailure}. This caps the probe burst so concurrent
+   *   callers on one warm container don't stampede a backend that may still be saturated.
+   *
+   * @returns true if the request should be attempted, false if it should fail fast
+   */
+  canRequest(): boolean {
+    if (this._state === 'open') {
+      if (this._now() - this._openedAt >= this._cooldownMs) {
+        this.transition('half-open', 'cooldown elapsed; probing');
+        this._probesInFlight += 1;
+        return true;
+      }
+      return false;
+    }
+    if (this._state === 'half-open') {
+      // Admit only up to the probe budget; further concurrent callers fail fast until an
+      // in-flight probe resolves (closing the breaker on success, re-opening on failure).
+      if (this._probesInFlight < this._halfOpenProbes) {
+        this._probesInFlight += 1;
+        return true;
+      }
+      return false;
+    }
+    return true;
+  }
+
+  /**
+   * Record a successful backend call. Closes the breaker once enough probes succeed in
+   * half-open; resets the failure score in closed.
+   */
+  recordSuccess(): void {
+    if (this._state === 'half-open') {
+      this._probesInFlight = Math.max(0, this._probesInFlight - 1);
+      this._probeSuccesses += 1;
+      if (this._probeSuccesses >= this._halfOpenProbes) {
+        this.transition('closed', 'probe succeeded; recovered');
+      }
+      return;
+    }
+    // A success in the closed state is the healthy path — reset the failure score so only a
+    // sustained, near-uninterrupted run of failures (not an intermittent trickle interleaved
+    // with successes) can open the breaker. This is what keeps normal miss/hit traffic from
+    // ever tripping it.
+    this._failureScore = 0;
+  }
+
+  /**
+   * Record a failed backend call.
+   *
+   * @param throttled Whether the failure was a throttling response (weighted heavier).
+   */
+  recordFailure(throttled: boolean): void {
+    if (this._state === 'half-open') {
+      // Any failure during probing means the backend has not recovered.
+      this._probesInFlight = Math.max(0, this._probesInFlight - 1);
+      this.transition('open', 'probe failed; backend still unhealthy');
+      return;
+    }
+    if (this._state === 'open') {
+      // Already open (e.g. a call that started before opening lands late) — nothing to do.
+      return;
+    }
+    this._failureScore += throttled ? this._throttleWeight : 1;
+    if (this._failureScore >= this._failureThreshold) {
+      this.transition('open', throttled ? 'failure threshold reached (throttling)' : 'failure threshold reached');
+    }
+  }
+
+  private transition(to: CircuitBreakerState, reason: string): void {
+    const from = this._state;
+    if (from === to) {
+      return;
+    }
+    this._state = to;
+    if (to === 'open') {
+      this._openedAt = this._now();
+      this._failureScore = 0;
+      this._probeSuccesses = 0;
+      this._probesInFlight = 0;
+    } else if (to === 'half-open') {
+      this._probeSuccesses = 0;
+      // Note: the admitting caller in canRequest() increments _probesInFlight after this
+      // transition runs, so it must start from zero here.
+      this._probesInFlight = 0;
+    } else {
+      // closed
+      this._failureScore = 0;
+      this._probeSuccesses = 0;
+      this._probesInFlight = 0;
+    }
+    this._onTransition?.({from, to, reason});
+  }
+}

packages/mrt-utilities/src/data-store/production.ts

@@ -7,8 +7,9 @@
 import {DynamoDBClient} from '@aws-sdk/client-dynamodb';
 import {DynamoDBDocumentClient, GetCommand, type GetCommandOutput} from '@aws-sdk/lib-dynamodb';
 
+import {CircuitBreaker} from './circuit-breaker.js';
 import {DataStoreNotFoundError, DataStoreServiceError, DataStoreUnavailableError} from './errors.js';
-import {logMRTError} from '../utils/utils.js';
+import {logMRTError, logMRTEvent} from '../utils/utils.js';
 
 export {DataStoreNotFoundError, DataStoreServiceError, DataStoreUnavailableError} from './errors.js';
 
@@ -70,6 +71,55 @@ const THROTTLING_ERROR_NAMES = new Set([
   'Throttling',
 ]);
 
+/**
+ * Circuit-breaker failure weight needed to trip from closed to open.
+ *
+ * A plain service error contributes 1 point; a throttling error contributes
+ * {@link DAL_BREAKER_THROTTLE_WEIGHT}. Sized so normal miss/hit traffic (misses are not
+ * failures) never opens the breaker, but a short run of real service failures does.
+ */
+const DAL_BREAKER_FAILURE_THRESHOLD = 5;
+
+/**
+ * Points a throttling failure contributes toward the trip threshold.
+ *
+ * Weighted heavier than a plain failure because sustained throttling is exactly the load
+ * signal this breaker exists to shed — a handful of throttles should open it well before an
+ * equal number of unrelated transient errors would.
+ */
+const DAL_BREAKER_THROTTLE_WEIGHT = 2;
+
+/**
+ * How long (ms) the breaker stays open before admitting a half-open probe.
+ *
+ * A short window: long enough to give a saturated table room to recover, short enough that a
+ * false trip only briefly diverts reads to the application-level API fallback.
+ */
+const DAL_BREAKER_COOLDOWN_MS = 5_000;
+
+/**
+ * Consecutive successful probes required in half-open to close the breaker.
+ */
+const DAL_BREAKER_HALF_OPEN_PROBES = 1;
+
+/**
+ * Env var kill switch for the circuit breaker.
+ *
+ * Set to a truthy value (`'1'` / `'true'`) to disable breaking entirely — reads always flow
+ * through to DynamoDB as if the breaker were permanently closed. Unset or falsy leaves the
+ * breaker active. The only operational dial; thresholds/cooldown are engineering-tuned
+ * constants, not incident-time knobs.
+ */
+const DAL_BREAKER_DISABLED_ENV = 'MRT_DATA_STORE_CIRCUIT_BREAKER_DISABLED';
+
+/**
+ * Whether the circuit breaker is disabled via {@link DAL_BREAKER_DISABLED_ENV}.
+ */
+function isBreakerDisabled(): boolean {
+  const value = process.env[DAL_BREAKER_DISABLED_ENV];
+  return value === '1' || value?.toLowerCase() === 'true';
+}
+
 /**
  * Whether an error represents a throttling response.
  *
@@ -127,14 +177,20 @@ export function createDalDynamoDBClient(): DynamoDBClient {
 export class DataStore {
   private _tableName: string = '';
   private _ddb: DynamoDBDocumentClient | null = null;
+  private _breaker: CircuitBreaker | null = null;
   private static _instance: DataStore | null = null;
 
   /** @internal Test hook: inject a document client for unit tests */
   static _testDocumentClient: DynamoDBDocumentClient | null = null;
   /** @internal Test hook: inject logMRTError for unit tests */
   static _testLogMRTError: ((namespace: string, err: unknown, context?: Record<string, unknown>) => void) | null = null;
+  /** @internal Test hook: inject logMRTEvent for unit tests */
+  static _testLogMRTEvent: ((namespace: string, message: string, context?: Record<string, unknown>) => void) | null =
+    null;
   /** @internal Test hook: inject a deterministic random source (returns [0, 1)) for unit tests */
   static _testRandom: (() => number) | null = null;
+  /** @internal Test hook: inject a circuit breaker (e.g. with a fake clock) for unit tests */
+  static _testBreaker: CircuitBreaker | null = null;
 
   private constructor() {
     // Private constructor for singleton; use DataStore.getDataStore() instead.
@@ -192,6 +248,43 @@ export class DataStore {
     return shard === 0 ? base : `${base} ${shard}`;
   }
 
+  /**
+   * Get or create this instance's circuit breaker.
+   *
+   * The breaker is memoized per DataStore instance so its state rides the same warm-container
+   * reuse as the singleton and the memoized DynamoDB client — a cold start begins closed. It
+   * emits state transitions via the MRT internal log constructs: opening is logged as an
+   * error (the backend is failing), recovery/probing as an event (info level) so recovery
+   * doesn't trip error-based alerting.
+   *
+   * @private
+   * @returns The circuit breaker guarding data store reads
+   */
+  private getBreaker(): CircuitBreaker {
+    if (DataStore._testBreaker) {
+      return DataStore._testBreaker;
+    }
+    if (!this._breaker) {
+      this._breaker = new CircuitBreaker({
+        failureThreshold: DAL_BREAKER_FAILURE_THRESHOLD,
+        throttleWeight: DAL_BREAKER_THROTTLE_WEIGHT,
+        cooldownMs: DAL_BREAKER_COOLDOWN_MS,
+        halfOpenProbes: DAL_BREAKER_HALF_OPEN_PROBES,
+        onTransition: ({from, to, reason}) => {
+          const context = {from, to, reason};
+          if (to === 'open') {
+            const logFn = DataStore._testLogMRTError ?? logMRTError;
+            logFn('data_store', new Error(`Circuit breaker opened: ${reason}`), context);
+          } else {
+            const logFn = DataStore._testLogMRTEvent ?? logMRTEvent;
+            logFn('data_store', 'circuit breaker state change', context);
+          }
+        },
+      });
+    }
+    return this._breaker;
+  }
+
   /**
    * Get or create the singleton DataStore instance.
    *
@@ -229,6 +322,15 @@ export class DataStore {
 
     const ddb = this.getClient();
     const projectEnvironment = this.resolveShardPartitionKey();
+
+    // Circuit breaker: shed load when the table is failing. When open, fail fast without
+    // calling DynamoDB — the client's application-level API fallback then serves correct
+    // data. Skipped entirely when disabled via the kill switch.
+    const breaker = isBreakerDisabled() ? null : this.getBreaker();
+    if (breaker && !breaker.canRequest()) {
+      throw new DataStoreServiceError('Data store request failed.');
+    }
+
     let response: GetCommandOutput;
     try {
       response = await ddb.send(
@@ -243,11 +345,16 @@ export class DataStore {
     } catch (error) {
       const errorName = error instanceof Error ? error.name : undefined;
       const throttled = isThrottlingError(error);
+      breaker?.recordFailure(throttled);
       const logFn = DataStore._testLogMRTError ?? logMRTError;
       logFn('data_store', error, {key, projectEnvironment, tableName: this._tableName, errorName, throttled});
       throw new DataStoreServiceError('Data store request failed.');
     }
 
+    // The send succeeded (the table answered) — record success even on a miss, since a miss
+    // is a healthy response, not a backend failure.
+    breaker?.recordSuccess();
+
     if (!response.Item?.value) {
       throw new DataStoreNotFoundError(`Data store entry '${key}' not found.`);
     }

packages/mrt-utilities/src/utils/utils.ts

@@ -45,3 +45,26 @@ export const logMRTError = (namespace: string, err: unknown, context?: Record<st
     }),
   );
 };
+
+/**
+ * Log an internal MRT event (non-error).
+ *
+ * Companion to {@link logMRTError} for structured operational events that are NOT errors —
+ * e.g. circuit-breaker state transitions, including recovery. Uses the same searchable
+ * `__MRT__<namespace>` key but emits at info level so recovery events don't trip
+ * error-based alerting.
+ *
+ * @param namespace Namespace for the event (e.g. data_store) to facilitate searching
+ * @param message Short, low-cardinality event message
+ * @param context Optional context to include in the log
+ */
+export const logMRTEvent = (namespace: string, message: string, context?: Record<string, unknown>) => {
+  console.info(
+    JSON.stringify({
+      [`__MRT__${namespace}`]: 'event',
+      type: 'MRT_internal',
+      message,
+      ...context,
+    }),
+  );
+};