From ef358f76c061cbcdf32cb5e4680b8e92d59137fa Mon Sep 17 00:00:00 2001 From: Nicki Peternel Date: Mon, 24 Aug 2026 09:05:19 -0700 Subject: [PATCH] @W-23735204: Add circuit breaker around data store reads Wrap DataStore.getEntry in a per-warm-container in-memory circuit breaker (closed -> open -> half-open) that sheds load when DynamoDB is failing or throttling: after a run of failures (throttles weighted heavier) the breaker opens and fails fast for a cooldown without calling DynamoDB, letting the client's application-level API fallback serve reads, then admits a bounded probe burst to recover. - Misses (DataStoreNotFoundError) count as healthy, never trip the breaker. - Telemetry only on state transitions: open via logMRTError, recovery via a new info-level logMRTEvent so recovery doesn't trip error alerting. - Kill switch: MRT_DATA_STORE_CIRCUIT_BREAKER_DISABLED. Threshold/cooldown/ probe count are internal constants. - CircuitBreaker kept internal (not exported from the package barrel). --- .changeset/data-store-circuit-breaker.md | 5 + .../src/data-store/circuit-breaker.ts | 200 +++++++++++++++ .../src/data-store/production.ts | 109 ++++++++- packages/mrt-utilities/src/utils/utils.ts | 23 ++ .../test/circuit-breaker.test.ts | 189 +++++++++++++++ .../mrt-utilities/test/data-store.test.ts | 228 ++++++++++++++++++ 6 files changed, 753 insertions(+), 1 deletion(-) create mode 100644 .changeset/data-store-circuit-breaker.md create mode 100644 packages/mrt-utilities/src/data-store/circuit-breaker.ts create mode 100644 packages/mrt-utilities/test/circuit-breaker.test.ts diff --git a/.changeset/data-store-circuit-breaker.md b/.changeset/data-store-circuit-breaker.md new file mode 100644 index 000000000..74a6af8eb --- /dev/null +++ b/.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. diff --git a/packages/mrt-utilities/src/data-store/circuit-breaker.ts b/packages/mrt-utilities/src/data-store/circuit-breaker.ts new file mode 100644 index 000000000..35c2a2f3e --- /dev/null +++ b/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}); + } +} diff --git a/packages/mrt-utilities/src/data-store/production.ts b/packages/mrt-utilities/src/data-store/production.ts index 620c2cc25..9b4abaf92 100644 --- a/packages/mrt-utilities/src/data-store/production.ts +++ b/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) => void) | null = null; + /** @internal Test hook: inject logMRTEvent for unit tests */ + static _testLogMRTEvent: ((namespace: string, message: string, context?: Record) => 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.`); } diff --git a/packages/mrt-utilities/src/utils/utils.ts b/packages/mrt-utilities/src/utils/utils.ts index b9ed8cf6f..d46d495cd 100644 --- a/packages/mrt-utilities/src/utils/utils.ts +++ b/packages/mrt-utilities/src/utils/utils.ts @@ -45,3 +45,26 @@ export const logMRTError = (namespace: string, err: unknown, context?: Record` 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) => { + console.info( + JSON.stringify({ + [`__MRT__${namespace}`]: 'event', + type: 'MRT_internal', + message, + ...context, + }), + ); +}; diff --git a/packages/mrt-utilities/test/circuit-breaker.test.ts b/packages/mrt-utilities/test/circuit-breaker.test.ts new file mode 100644 index 000000000..3885acfa5 --- /dev/null +++ b/packages/mrt-utilities/test/circuit-breaker.test.ts @@ -0,0 +1,189 @@ +/* + * 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 + */ + +import {expect} from 'chai'; +import sinon from 'sinon'; +// Imported via a relative path, not the package barrel: CircuitBreaker is an internal +// implementation detail of the data store, not part of the published API surface. +import {CircuitBreaker} from '../src/data-store/circuit-breaker.js'; + +describe('CircuitBreaker', () => { + let clock: number; + const now = () => clock; + + const makeBreaker = (overrides: Partial[0]> = {}) => + new CircuitBreaker({ + failureThreshold: 3, + throttleWeight: 2, + cooldownMs: 1_000, + halfOpenProbes: 1, + now, + ...overrides, + }); + + beforeEach(() => { + clock = 0; + }); + + it('starts closed and admits requests', () => { + const breaker = makeBreaker(); + expect(breaker.state).to.equal('closed'); + expect(breaker.canRequest()).to.equal(true); + }); + + it('opens once accumulated plain failures reach the threshold', () => { + const breaker = makeBreaker({failureThreshold: 3}); + breaker.recordFailure(false); + breaker.recordFailure(false); + expect(breaker.state).to.equal('closed'); + breaker.recordFailure(false); + expect(breaker.state).to.equal('open'); + expect(breaker.canRequest()).to.equal(false); + }); + + it('weights throttling failures heavier when tripping', () => { + const breaker = makeBreaker({failureThreshold: 4, throttleWeight: 2}); + breaker.recordFailure(true); // 2 + breaker.recordFailure(true); // 4 => trip + expect(breaker.state).to.equal('open'); + }); + + it('decays the failure score on a success while closed', () => { + const breaker = makeBreaker({failureThreshold: 2}); + breaker.recordFailure(false); // 1 + breaker.recordSuccess(); // reset to 0 + breaker.recordFailure(false); // 1, not 2 + expect(breaker.state).to.equal('closed'); + }); + + it('stays open until the cooldown elapses, then admits a half-open probe', () => { + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000}); + breaker.recordFailure(false); + expect(breaker.state).to.equal('open'); + + clock = 999; + expect(breaker.canRequest()).to.equal(false); + expect(breaker.state).to.equal('open'); + + clock = 1_000; + expect(breaker.canRequest()).to.equal(true); + expect(breaker.state).to.equal('half-open'); + }); + + it('closes after the required number of successful probes', () => { + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, halfOpenProbes: 2}); + breaker.recordFailure(false); + clock = 1_000; + breaker.canRequest(); // -> half-open + breaker.recordSuccess(); // 1 of 2 + expect(breaker.state).to.equal('half-open'); + breaker.recordSuccess(); // 2 of 2 + expect(breaker.state).to.equal('closed'); + }); + + it('re-opens on any failure while half-open', () => { + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000}); + breaker.recordFailure(false); + clock = 1_000; + breaker.canRequest(); // -> half-open + breaker.recordFailure(false); + expect(breaker.state).to.equal('open'); + }); + + it('admits only up to halfOpenProbes concurrent probes; further callers fail fast', () => { + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, halfOpenProbes: 1}); + breaker.recordFailure(false); + clock = 1_000; + + // First caller is admitted as the probe... + expect(breaker.canRequest()).to.equal(true); + expect(breaker.state).to.equal('half-open'); + // ...concurrent callers, before the probe resolves, are turned away. + expect(breaker.canRequest()).to.equal(false); + expect(breaker.canRequest()).to.equal(false); + }); + + it('admits a fresh probe after an in-flight probe resolves without closing', () => { + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, halfOpenProbes: 2}); + breaker.recordFailure(false); + clock = 1_000; + + expect(breaker.canRequest()).to.equal(true); // probe 1 in flight + expect(breaker.canRequest()).to.equal(true); // probe 2 in flight (budget 2) + expect(breaker.canRequest()).to.equal(false); // budget exhausted + breaker.recordSuccess(); // probe 1 resolves; 1 of 2 successes, still half-open + expect(breaker.state).to.equal('half-open'); + expect(breaker.canRequest()).to.equal(true); // slot freed, admit another + }); + + it('does not open under a mixed failure/success stream (successes reset the score)', () => { + const breaker = makeBreaker({failureThreshold: 3}); + // Alternating fail/success never accumulates to the threshold. + for (let i = 0; i < 20; i++) { + breaker.recordFailure(i % 2 === 0); // vary throttled vs not + breaker.recordSuccess(); + } + expect(breaker.state).to.equal('closed'); + }); + + it('resets the cooldown window when it re-opens from half-open', () => { + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000}); + breaker.recordFailure(false); // opened at t=0 + clock = 1_000; + breaker.canRequest(); // -> half-open + breaker.recordFailure(false); // re-opened at t=1000 + + clock = 1_500; // only 500ms since re-open + expect(breaker.canRequest()).to.equal(false); + clock = 2_000; // full cooldown since re-open + expect(breaker.canRequest()).to.equal(true); + }); + + it('ignores a late failure that arrives while already open', () => { + const breaker = makeBreaker({failureThreshold: 1}); + breaker.recordFailure(false); + expect(breaker.state).to.equal('open'); + // A call that started before opening lands late — must not extend/alter state. + breaker.recordFailure(false); + expect(breaker.state).to.equal('open'); + }); + + it('notifies onTransition for every state change with from/to/reason', () => { + const onTransition = sinon.stub(); + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, onTransition}); + + breaker.recordFailure(true); // closed -> open + clock = 1_000; + breaker.canRequest(); // open -> half-open + breaker.recordSuccess(); // half-open -> closed + + expect(onTransition.callCount).to.equal(3); + const transitions = onTransition.getCalls().map((c) => ({from: c.args[0].from, to: c.args[0].to})); + expect(transitions).to.deep.equal([ + {from: 'closed', to: 'open'}, + {from: 'open', to: 'half-open'}, + {from: 'half-open', to: 'closed'}, + ]); + for (const call of onTransition.getCalls()) { + expect(call.args[0].reason).to.be.a('string').and.not.empty; + } + }); + + it('defaults the clock to Date.now when none is injected', () => { + // Just exercises the default-now branch; behavior is unchanged when never opened. + const breaker = new CircuitBreaker({ + failureThreshold: 1, + throttleWeight: 2, + cooldownMs: 0, + halfOpenProbes: 1, + }); + expect(breaker.canRequest()).to.equal(true); + breaker.recordFailure(false); + // cooldownMs 0 => immediately eligible for a probe. + expect(breaker.canRequest()).to.equal(true); + expect(breaker.state).to.equal('half-open'); + }); +}); diff --git a/packages/mrt-utilities/test/data-store.test.ts b/packages/mrt-utilities/test/data-store.test.ts index 6a55f3480..bee7de659 100644 --- a/packages/mrt-utilities/test/data-store.test.ts +++ b/packages/mrt-utilities/test/data-store.test.ts @@ -14,6 +14,8 @@ import { DataStoreServiceError, DataStoreUnavailableError, } from '@salesforce/mrt-utilities'; +// Internal detail, imported directly rather than through the package barrel. +import {CircuitBreaker} from '../src/data-store/circuit-breaker.js'; describe('DataStore', () => { let mockSend: sinon.SinonStub; @@ -25,7 +27,9 @@ describe('DataStore', () => { (DataStore as unknown as {_instance: DataStore | null})._instance = null; DataStore._testDocumentClient = null; DataStore._testLogMRTError = null; + DataStore._testLogMRTEvent = null; DataStore._testRandom = null; + DataStore._testBreaker = null; mockSend = sinon.stub(); mockDocumentClient = {send: mockSend} as unknown as DynamoDBDocumentClient; @@ -41,7 +45,9 @@ describe('DataStore', () => { (DataStore as unknown as {_instance: DataStore | null})._instance = null; DataStore._testDocumentClient = null; DataStore._testLogMRTError = null; + DataStore._testLogMRTEvent = null; DataStore._testRandom = null; + DataStore._testBreaker = null; sinon.restore(); }); @@ -352,6 +358,228 @@ describe('DataStore', () => { }); } }); + + describe('circuit breaker', () => { + beforeEach(() => { + // Silence the request-failure logs these tests deliberately provoke; tests that + // assert on logging install their own stubs. + DataStore._testLogMRTError = sinon.stub(); + DataStore._testLogMRTEvent = sinon.stub(); + }); + + // A breaker that trips after a single failure and never leaves open on its own, so + // integration tests can drive open/closed deterministically without a clock. + const makeBreaker = (overrides: Partial[0]> = {}) => + new CircuitBreaker({ + failureThreshold: 1, + throttleWeight: 2, + cooldownMs: 1_000, + halfOpenProbes: 1, + now: () => 0, + ...overrides, + }); + + const expectRejects = async (promise: Promise, ErrorType: new (...args: never[]) => Error) => { + try { + await promise; + expect.fail('should have thrown'); + } catch (e) { + expect(e).to.be.an.instanceOf(ErrorType); + } + }; + + it('fails fast without calling DynamoDB when the breaker is open', async () => { + const breaker = makeBreaker(); + breaker.recordFailure(false); // trip it + expect(breaker.state).to.equal('open'); + DataStore._testBreaker = breaker; + + await expectRejects(DataStore.getDataStore().getEntry('my-key'), DataStoreServiceError); + + expect(mockSend.callCount).to.equal(0); + }); + + it('accumulates service failures toward tripping and opens the breaker', async () => { + const breaker = makeBreaker({failureThreshold: 2}); + DataStore._testBreaker = breaker; + mockSend.rejects(new Error('boom')); + + const store = DataStore.getDataStore(); + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('closed'); // 1 point < threshold 2 + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('open'); // 2 points >= threshold + + // Now open: the next call short-circuits without reaching DynamoDB. + expect(mockSend.callCount).to.equal(2); + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(mockSend.callCount).to.equal(2); + }); + + it('weights a throttling failure heavier than a plain failure', async () => { + const breaker = makeBreaker({failureThreshold: 2, throttleWeight: 2}); + DataStore._testBreaker = breaker; + const throttle = new Error('slow down'); + throttle.name = 'ThrottlingException'; + mockSend.rejects(throttle); + + // A single throttle contributes 2 points, meeting the threshold on its own. + await expectRejects(DataStore.getDataStore().getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('open'); + }); + + it('does not trip on a miss (a miss is a healthy response, not a failure)', async () => { + const breaker = makeBreaker({failureThreshold: 1}); + DataStore._testBreaker = breaker; + mockSend.resolves({}); // miss + + const store = DataStore.getDataStore(); + for (let i = 0; i < 5; i++) { + await expectRejects(store.getEntry('my-key'), DataStoreNotFoundError); + } + expect(breaker.state).to.equal('closed'); + expect(mockSend.callCount).to.equal(5); + }); + + it('recovers: a successful probe after cooldown closes the breaker', async () => { + let clock = 0; + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, now: () => clock}); + DataStore._testBreaker = breaker; + + const store = DataStore.getDataStore(); + mockSend.rejects(new Error('boom')); + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('open'); + + // Still open before cooldown elapses: fails fast, no send. + clock = 999; + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(mockSend.callCount).to.equal(1); + + // After cooldown, a probe is admitted; a success closes the breaker. + clock = 1_000; + mockSend.resolves({Item: {value: {theme: 'dark'}}}); + const result = await store.getEntry('my-key'); + expect(result).to.deep.equal({key: 'my-key', value: {theme: 'dark'}}); + expect(breaker.state).to.equal('closed'); + expect(mockSend.callCount).to.equal(2); + }); + + it('admits only one probe when concurrent reads straddle the half-open transition', async () => { + let clock = 0; + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, halfOpenProbes: 1, now: () => clock}); + DataStore._testBreaker = breaker; + + const store = DataStore.getDataStore(); + mockSend.rejects(new Error('boom')); + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('open'); + expect(mockSend.callCount).to.equal(1); + + // Cooldown elapsed: fire several reads concurrently. Each getEntry runs synchronously + // up to its `await ddb.send`, in call order — so the first is admitted as the probe + // and increments the in-flight count before the others check, and the rest fail fast. + // This proves a burst can't stampede a backend that may still be saturated. + clock = 1_000; + mockSend.onCall(1).resolves({Item: {value: {theme: 'dark'}}}); + + const results = await Promise.allSettled([ + store.getEntry('my-key'), + store.getEntry('my-key'), + store.getEntry('my-key'), + ]); + + // Exactly one call was admitted to DynamoDB as the probe; the other two rejected. + expect(mockSend.callCount).to.equal(2); + const fulfilled = results.filter((r) => r.status === 'fulfilled'); + const rejected = results.filter((r) => r.status === 'rejected'); + expect(fulfilled).to.have.lengthOf(1); + expect(rejected).to.have.lengthOf(2); + for (const r of rejected) { + expect((r as PromiseRejectedResult).reason).to.be.an.instanceOf(DataStoreServiceError); + } + // The admitted probe succeeded, closing the breaker. + expect(breaker.state).to.equal('closed'); + }); + + it('re-opens if the half-open probe fails', async () => { + let clock = 0; + const breaker = makeBreaker({failureThreshold: 1, cooldownMs: 1_000, now: () => clock}); + DataStore._testBreaker = breaker; + mockSend.rejects(new Error('boom')); + + const store = DataStore.getDataStore(); + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('open'); + + clock = 1_000; // probe admitted, but the backend is still failing + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + expect(breaker.state).to.equal('open'); + expect(mockSend.callCount).to.equal(2); + }); + + it('emits an error log when opening and an event log on recovery', async () => { + let clock = 0; + const errorLog = sinon.stub(); + const eventLog = sinon.stub(); + DataStore._testLogMRTError = errorLog; + DataStore._testLogMRTEvent = eventLog; + + const breaker = new CircuitBreaker({ + failureThreshold: 1, + throttleWeight: 2, + cooldownMs: 1_000, + halfOpenProbes: 1, + now: () => clock, + onTransition: ({from, to, reason}) => { + const context = {from, to, reason}; + if (to === 'open') { + errorLog('data_store', new Error(`Circuit breaker opened: ${reason}`), context); + } else { + eventLog('data_store', 'circuit breaker state change', context); + } + }, + }); + DataStore._testBreaker = breaker; + + const store = DataStore.getDataStore(); + mockSend.rejects(new Error('boom')); + await expectRejects(store.getEntry('my-key'), DataStoreServiceError); + // The opening call logs twice at error level: the failed request itself, and the + // breaker-opened transition. Find the transition one by its context shape. + const openTransitionCall = errorLog + .getCalls() + .find((c) => (c.args[2] as {to?: string} | undefined)?.to === 'open'); + expect(openTransitionCall, 'expected an error log for the open transition').to.exist; + expect(openTransitionCall!.args[0]).to.equal('data_store'); + expect(openTransitionCall!.args[2]).to.deep.include({from: 'closed', to: 'open'}); + const errorLogsAfterOpen = errorLog.callCount; + + clock = 1_000; + mockSend.resolves({Item: {value: {theme: 'dark'}}}); + await store.getEntry('my-key'); + // half-open then closed => two event logs; recovery adds no further error logs. + expect(errorLog.callCount).to.equal(errorLogsAfterOpen); + expect(eventLog.callCount).to.equal(2); + expect(eventLog.getCalls().map((c) => c.args[2].to)).to.deep.equal(['half-open', 'closed']); + }); + + for (const disabledValue of ['true', '1']) { + it(`bypasses the breaker entirely when disabled via the kill switch (${disabledValue})`, async () => { + process.env.MRT_DATA_STORE_CIRCUIT_BREAKER_DISABLED = disabledValue; + const breaker = makeBreaker(); + breaker.recordFailure(false); // would be open + expect(breaker.state).to.equal('open'); + DataStore._testBreaker = breaker; + mockSend.resolves({Item: {value: {theme: 'dark'}}}); + + // Breaker is open but disabled, so the read still reaches DynamoDB. + const result = await DataStore.getDataStore().getEntry('my-key'); + expect(result).to.deep.equal({key: 'my-key', value: {theme: 'dark'}}); + expect(mockSend.callCount).to.equal(1); + }); + } + }); }); });