@studnicky/resilience
Circuit breaker, token bucket rate limiter, and bounded dead-letter queue. Each primitive is independently usable and composable.
Install
bash
pnpm add @studnicky/resilienceRequires @studnicky:registry=https://npm.pkg.github.com in .npmrc.
Usage
CircuitBreaker
Tracks failures and opens the circuit after a threshold, then probes with limited calls after a timeout.
ts
import { CircuitBreaker, CircuitBreakerOpenError } from '../src/index.js';
// Deterministic clock so tests are instant with no real waits.
let now = 0;
class Clock {
static now(): number { const result = now; return result; }
}
const breaker = CircuitBreaker.create({
'clock': Clock.now,
'failureThreshold': 3,
'name': 'test-service',
'resetTimeoutMs': 1_000,
'successThreshold': 2
});
// --- CLOSED: successful calls pass through ---
await breaker.execute(() => { const result = Promise.resolve('ok'); return result; });
console.log('State after success:', breaker.state);
// --- Trip the breaker: 3 consecutive failures ---
class Fail {
static boom(): Promise<never> { throw new Error('boom'); }
}
for (let i = 0; i < 3; i++) {
await breaker.execute(Fail.boom).catch(() => { /* expected */ });
}
console.log('State after 3 failures:', breaker.state);
// --- OPEN: next call is fast-rejected with CircuitBreakerOpenError ---
await breaker.execute(() => { const result = Promise.resolve('should not run'); return result; }).catch((err: unknown) => {
console.log('Open-circuit rejection:', err instanceof CircuitBreakerOpenError ? 'CircuitBreakerOpenError' : 'other');
});
// --- Advance past resetTimeoutMs → halfOpen on next call ---
now = 1_001;
await breaker.execute(() => { const result = Promise.resolve('probe 1'); return result; });
console.log('State after probe 1:', breaker.state);
// --- 2 successes in halfOpen close the circuit (successThreshold: 2) ---
await breaker.execute(() => { const result = Promise.resolve('probe 2'); return result; });
console.log('State after probe 2:', breaker.state);
// --- forceOpen / forceClosed ---
breaker.forceOpen();
console.log('State after forceOpen:', breaker.state);
breaker.reset();
console.log('State after reset:', breaker.state);TokenBucket
Leaky-bucket rate limiter; consume throws immediately when exhausted, waitForToken blocks until tokens refill.
ts
import { TokenBucket, TokenBucketExhaustedError } from '../src/index.js';
// Deterministic clock: start at t=0, advance manually.
let now = 0;
class Clock {
static now(): number { const result = now; return result; }
}
// 2 tokens/s, burst capacity 3 → starts full (3 tokens).
const bucket = TokenBucket.create({ 'burstSize': 3, 'clock': Clock.now, 'requestsPerSecond': 2 });
console.log('Initial available tokens:', bucket.available);
// --- consume() drains tokens ---
bucket.consume();
bucket.consume();
bucket.consume();
console.log('Available after 3 consumes:', bucket.available);
// --- Advance 500 ms → 1 new token (2 tokens/s × 0.5 s = 1) ---
now = 500;
console.log('Available at t=500ms:', bucket.available);
bucket.consume();
// --- Advance to 1500 ms → 2 tokens refilled, capped at burstSize=3 ---
now = 1_500;
console.log('Available at t=1500ms:', bucket.available);
// --- waitForToken with abort signal cancels correctly ---
// Drain remaining tokens so waitForToken has to block, then abort concurrently.
bucket.consume();
bucket.consume();
const controller = new AbortController();
const abortError = new Error('cancelled');
// Abort after a microtask so waitForToken is already suspended in the Promise.
const waitPromise = bucket.waitForToken({ 'signal': controller.signal, 'tokens': 1 });
queueMicrotask(() => { controller.abort(abortError); });
const abortRejected = await waitPromise.then(() => { const result = false; return result; }).catch(() => { const result = true; return result; });
console.log('waitForToken aborted:', abortRejected);DeadLetterQueue
Bounded FIFO queue for items that failed processing. Drain via async generator.
DeadLetterQueueRetryGenerator: timed re-delivery
ts
import {
DeadLetterQueue,
DeadLetterQueueRetryGenerator,
DlqClosedError,
DlqFullError
} from '../src/index.js';
await (async function runDeadLetterQueueExample(): Promise<void> {
// --- Basic enqueue and drain ---
const dlq = DeadLetterQueue.create<string>({ 'capacity': 5 });
dlq.enqueue('job-1', 'timeout');
dlq.enqueue('job-2', 'network error', new Error('ECONNREFUSED'));
console.log('Queue size after 2 enqueues:', dlq.size);
// Close before draining so the generator terminates instead of waiting.
dlq.close();
const collected: string[] = [];
for await (const entry of dlq.drain()) {
collected.push(entry.item);
}
console.log('Drained items:', collected);
console.log('Queue size after drain:', dlq.size);
// --- Capacity enforcement ---
const bounded = DeadLetterQueue.create<number>({ 'capacity': 2 });
bounded.enqueue(1, 'err');
bounded.enqueue(2, 'err');
console.log('Bounded queue size:', bounded.size);
// --- DeadLetterQueueRetryGenerator re-yields entries with a pause ---
const retryDlq = DeadLetterQueue.create<string>();
retryDlq.enqueue('retry-job-1', 'failed');
retryDlq.enqueue('retry-job-2', 'failed');
retryDlq.close();
const gen = DeadLetterQueueRetryGenerator.create({ 'dlq': retryDlq, 'intervalMs': 0 });
const retried: string[] = [];
for await (const entry of gen.generate()) {
retried.push(entry.item);
}
console.log('Retried items:', retried);
// --- AbortSignal aborts the queue on construction ---
const controller = new AbortController();
controller.abort();
const abortedDlq = DeadLetterQueue.create<string>({ 'signal': controller.signal });
const abortedEntries: string[] = [];
for await (const entry of abortedDlq.drain()) {
abortedEntries.push(entry.item);
}
console.log('Aborted drain count:', abortedEntries.length);
assert.deepEqual(collected, ['job-1', 'job-2']);
assert.equal(dlq.size, 0);
assert.throws(() => { dlq.enqueue('job-3', 'late'); }, DlqClosedError);
assert.throws(() => { bounded.enqueue(3, 'overflow'); }, DlqFullError);
assert.deepEqual(retried, ['retry-job-1', 'retry-job-2']);
assert.equal(abortedEntries.length, 0);
console.log('dead-letter-queue: all assertions passed');
})();Observability hooks
Subclass any primitive and override protected hooks to add logging, metrics, or tracing without coupling the core to any observability library.
CircuitBreaker hooks
| Hook | When it fires | Args |
|---|---|---|
onSuccess() | After fn() resolves in any state | — |
onFailure(error) | After fn() throws in any state | error: unknown |
onTrip() | When failure threshold is reached and state transitions closed → open | — |
onOpen() | Every time state becomes open (threshold trip or halfOpen → open on failure) | — |
onHalfOpen() | When state transitions open → halfOpen after resetTimeoutMs | — |
onClose() | When state becomes closed (success threshold reached in halfOpen or manual reset) | — |
onReject() | When a call is short-circuited because the circuit is open | — |
TokenBucket hooks
| Hook | When it fires | Args |
|---|---|---|
onTokenAcquired(count) | After consume() or waitForToken() successfully deducts tokens | count: number |
onTokenDepleted() | When consume() finds insufficient tokens (before throwing) | — |
onRefill(added) | When the internal refill adds tokens due to elapsed time | added: number |
DeadLetterQueue hooks
| Hook | When it fires | Args |
|---|---|---|
onEnqueue(item) | After an item is added to the queue | item: T |
onDequeue(item) | After an item is shifted from the queue during drain | item: T |
onOverflow() | When enqueue() is called on a full queue (before throwing) | — |
onClose() | At the end of close() | — |
onAbort() | At the end of abort() | — |
DeadLetterQueueRetryGenerator hooks
| Hook | When it fires | Args |
|---|---|---|
onYield(entry) | Immediately before each entry is yielded from generate() | entry: DlqEntryType<T> |
onWait(intervalMs) | Before each inter-entry delay | intervalMs: number |
onDone() | When the generator finishes (DLQ closed or aborted) | — |
ts
import type { CircuitBreakerOptionsInterface } from '../src/interfaces/CircuitBreakerOptionsInterface.js';
import type { DeadLetterQueueOptionsInterface } from '../src/interfaces/DeadLetterQueueOptionsInterface.js';
import { CircuitBreaker } from '../src/CircuitBreaker.js';
import { CircuitBreakerOpenError } from '../src/CircuitBreakerOpenError.js';
import { DeadLetterQueue } from '../src/DeadLetterQueue.js';
// --- Observed CircuitBreaker ---
class TracedBreaker extends CircuitBreaker {
constructor(options: CircuitBreakerOptionsInterface) { super(options); }
protected override onSuccess(): void { console.log('[resilience:cb] onSuccess — circuit closed and call succeeded'); }
protected override onFailure(error: unknown): void { console.log(`[resilience:cb] onFailure — error=${String(error)}`); }
protected override onTrip(): void { console.log('[resilience:cb] onTrip — failure threshold reached, circuit OPEN'); }
protected override onOpen(): void { console.log('[resilience:cb] onOpen — circuit is now open'); }
protected override onHalfOpen(): void { console.log('[resilience:cb] onHalfOpen — probing after timeout'); }
protected override onClose(): void { console.log('[resilience:cb] onClose — circuit CLOSED, service healthy'); }
protected override onReject(): void { console.log('[resilience:cb] onReject — call short-circuited (circuit open)'); }
}
// --- Observed DeadLetterQueue ---
class TracedDlq<T> extends DeadLetterQueue<T> {
constructor(options?: DeadLetterQueueOptionsInterface) { super(options); }
protected override onEnqueue(item: T): void { console.log(`[resilience:dlq] onEnqueue — item=${String(item)}`); }
protected override onDequeue(item: T): void { console.log(`[resilience:dlq] onDequeue — item=${String(item)}`); }
protected override onOverflow(): void { console.log('[resilience:dlq] onOverflow — queue full, item dropped'); }
protected override onClose(): void { console.log('[resilience:dlq] onClose — queue sealed'); }
protected override onAbort(): void { console.log('[resilience:dlq] onAbort — queue aborted'); }
}
const events: string[] = [];
// Scenario: breaker trips open → half-open → closes; rejected calls go to DLQ
let time = 0;
class Clock {
static now(): number { const result = time; return result; }
}
const cb = new TracedBreaker({ 'clock': Clock.now, 'failureThreshold': 2, 'resetTimeoutMs': 100, 'successThreshold': 1 });
const dlq = new TracedDlq<string>({ 'capacity': 10 });
// Trip the breaker open with 2 failures
console.log('\n--- Tripping open ---');
try { await cb.execute(() => { throw new Error('service down'); }); } catch { dlq.enqueue('msg-1', 'service-failure'); events.push('failure-1'); }
try { await cb.execute(() => { throw new Error('service down'); }); } catch { dlq.enqueue('msg-2', 'service-failure'); events.push('failure-2'); }
// Call while open — should be rejected and sent to DLQ
console.log('\n--- Rejecting while open ---');
try {
await cb.execute(() => { const result = Promise.resolve('ok'); return result; });
} catch (e) {
if (e instanceof CircuitBreakerOpenError) {
dlq.enqueue('msg-3', 'circuit-open');
events.push('rejected');
}
}
// Advance clock past resetTimeoutMs → half-open
console.log('\n--- Half-open probe ---');
time = 100;
await cb.execute(() => { const result = Promise.resolve('ok'); return result; }); // half-open → closed
events.push('recovered');
// Drain the DLQ
console.log('\n--- Draining DLQ ---');
dlq.close();
for await (const entry of dlq.drain()) {
events.push(`drain:${entry.item}`);
}The base class never calls any logger or metrics library. All hooks are no-ops by default.
Try it
The builder demo constructs a CircuitBreaker via CircuitBreaker.builder().withName().withFailureThreshold().withResetTimeoutMs().withSuccessThreshold().withClock().build() and walks the breaker through all three states: closed, open (fast-rejection), and half-open probe back to closed. No real timers are used — the clock is a deterministic function.
CircuitBreaker builder
/** builderResilience — constructs a CircuitBreaker via CircuitBreaker.builder()...build() and exercises all three states. Run: npx tsx examples/builderResilience.ts */
import assert from 'node:assert/strict';
// #region usage
import { CircuitBreaker, CircuitBreakerOpenError } from '../src/index.js';
// Deterministic clock so the demo is instant with no real waits
let now = 0;
class Clock {
static now(): number { const result = now;
return result; }
}
// Build a CircuitBreaker with a fluent builder chain
const breaker = CircuitBreaker.builder()
.withName('payment-service')
.withFailureThreshold(2)
.withResetTimeoutMs(500)
.withSuccessThreshold(1)
.withClock(Clock.now)
.build();
console.log('CircuitBreaker built. State:', breaker.state);
// CLOSED — successful call passes through
await breaker.execute(() => { const result = Promise.resolve('ok');
return result; });
console.log('After success. State:', breaker.state);
// Trip the breaker with 2 consecutive failures
class Fail {
static downstream(): Promise<never> { throw new Error('downstream error'); }
}
await breaker.execute(Fail.downstream).catch(() => { /* expected */ });
await breaker.execute(Fail.downstream).catch(() => { /* expected */ });
console.log('After 2 failures. State:', breaker.state);
// OPEN — fast rejection
const rejectionError = await breaker
.execute(() => { const result = Promise.resolve('should not run');
return result; })
.catch((err: unknown) => { const result = err;
return result; });
console.log('Open-circuit rejection type:', rejectionError instanceof CircuitBreakerOpenError ? 'CircuitBreakerOpenError' : 'other');
// Advance past resetTimeoutMs → halfOpen probe succeeds → closed
now = 501;
await breaker.execute(() => { const result = Promise.resolve('probe');
return result; });
console.log('After probe. State:', breaker.state);
// #endregion usage
assert.equal(breaker.state, 'closed', 'Circuit re-closed after successful probe');
assert.ok(rejectionError instanceof CircuitBreakerOpenError, 'Open circuit rejected with CircuitBreakerOpenError');
console.log('builderResilience: all assertions passed');
Output
Press Execute to run this example against the real library.The hooks demo subclasses both CircuitBreaker and DeadLetterQueue and overrides their lifecycle hooks. Watch the full scenario: two failures trigger onFailure, onTrip, and onOpen; a rejected call triggers onReject; advancing the virtual clock into half-open triggers onHalfOpen, onSuccess, and onClose; and DLQ drain emits onDequeue for every item recovered from the queue.
Resilience lifecycle hooks
/** observedResilience — trace every lifecycle hook across CircuitBreaker, TokenBucket, and DeadLetterQueue. Run: npx tsx examples/observedResilience.ts */
import assert from 'node:assert/strict';
// #region usage
import type { CircuitBreakerOptionsInterface } from '../src/interfaces/CircuitBreakerOptionsInterface.js';
import type { DeadLetterQueueOptionsInterface } from '../src/interfaces/DeadLetterQueueOptionsInterface.js';
import { CircuitBreaker } from '../src/CircuitBreaker.js';
import { CircuitBreakerOpenError } from '../src/CircuitBreakerOpenError.js';
import { DeadLetterQueue } from '../src/DeadLetterQueue.js';
// --- Observed CircuitBreaker ---
class TracedBreaker extends CircuitBreaker {
constructor(options: CircuitBreakerOptionsInterface) { super(options); }
protected override onSuccess(): void { console.log('[resilience:cb] onSuccess — circuit closed and call succeeded'); }
protected override onFailure(error: unknown): void { console.log(`[resilience:cb] onFailure — error=${String(error)}`); }
protected override onTrip(): void { console.log('[resilience:cb] onTrip — failure threshold reached, circuit OPEN'); }
protected override onOpen(): void { console.log('[resilience:cb] onOpen — circuit is now open'); }
protected override onHalfOpen(): void { console.log('[resilience:cb] onHalfOpen — probing after timeout'); }
protected override onClose(): void { console.log('[resilience:cb] onClose — circuit CLOSED, service healthy'); }
protected override onReject(): void { console.log('[resilience:cb] onReject — call short-circuited (circuit open)'); }
}
// --- Observed DeadLetterQueue ---
class TracedDlq<T> extends DeadLetterQueue<T> {
constructor(options?: DeadLetterQueueOptionsInterface) { super(options); }
protected override onEnqueue(item: T): void { console.log(`[resilience:dlq] onEnqueue — item=${String(item)}`); }
protected override onDequeue(item: T): void { console.log(`[resilience:dlq] onDequeue — item=${String(item)}`); }
protected override onOverflow(): void { console.log('[resilience:dlq] onOverflow — queue full, item dropped'); }
protected override onClose(): void { console.log('[resilience:dlq] onClose — queue sealed'); }
protected override onAbort(): void { console.log('[resilience:dlq] onAbort — queue aborted'); }
}
const events: string[] = [];
// Scenario: breaker trips open → half-open → closes; rejected calls go to DLQ
let time = 0;
class Clock {
static now(): number { const result = time; return result; }
}
const cb = new TracedBreaker({ 'clock': Clock.now, 'failureThreshold': 2, 'resetTimeoutMs': 100, 'successThreshold': 1 });
const dlq = new TracedDlq<string>({ 'capacity': 10 });
// Trip the breaker open with 2 failures
console.log('\n--- Tripping open ---');
try { await cb.execute(() => { throw new Error('service down'); }); } catch { dlq.enqueue('msg-1', 'service-failure'); events.push('failure-1'); }
try { await cb.execute(() => { throw new Error('service down'); }); } catch { dlq.enqueue('msg-2', 'service-failure'); events.push('failure-2'); }
// Call while open — should be rejected and sent to DLQ
console.log('\n--- Rejecting while open ---');
try {
await cb.execute(() => { const result = Promise.resolve('ok'); return result; });
} catch (e) {
if (e instanceof CircuitBreakerOpenError) {
dlq.enqueue('msg-3', 'circuit-open');
events.push('rejected');
}
}
// Advance clock past resetTimeoutMs → half-open
console.log('\n--- Half-open probe ---');
time = 100;
await cb.execute(() => { const result = Promise.resolve('ok'); return result; }); // half-open → closed
events.push('recovered');
// Drain the DLQ
console.log('\n--- Draining DLQ ---');
dlq.close();
for await (const entry of dlq.drain()) {
events.push(`drain:${entry.item}`);
}
// #endregion usage
assert.ok(events.includes('failure-1'), 'failure-1 recorded');
assert.ok(events.includes('failure-2'), 'failure-2 recorded');
assert.ok(events.includes('rejected'), 'rejected recorded');
assert.ok(events.includes('recovered'), 'recovered');
assert.ok(events.includes('drain:msg-1'), 'DLQ drained msg-1');
assert.ok(events.includes('drain:msg-2'), 'DLQ drained msg-2');
assert.ok(events.includes('drain:msg-3'), 'DLQ drained msg-3');
console.log('\nobservedResilience: all assertions passed');
Output
Press Execute to run this example against the real library.API
| Export | Type | Description |
|---|---|---|
CircuitBreaker | class | Three-state async circuit breaker |
CircuitBreakerOpenError | class | Thrown when the circuit is open |
CircuitBreakerOptionsType | type | { failureThreshold, resetTimeoutMs, successThreshold?, name?, clock? } |
CircuitStateType | type | 'closed' | 'open' | 'halfOpen' |
TokenBucket | class | Token bucket rate limiter |
TokenBucketExhaustedError | class | Thrown by consume() when no tokens remain |
TokenBucketOptionsType | type | { requestsPerSecond, burstSize, clock? } |
DeadLetterQueue<T> | class | Bounded FIFO DLQ with async-generator drain |
DlqEntryType<T> | type | { id, item, reason, error?, enqueuedAtMs } |
DeadLetterQueueOptionsType | type | { capacity?, clock?, signal? } |
DeadLetterQueueRetryGenerator<T> | class | Re-yields DLQ entries with a configurable pause |
DeadLetterQueueRetryGeneratorOptionsType | type | { intervalMs } |
DlqFullError | class | Thrown by enqueue when at capacity |
DlqClosedError | class | Thrown by enqueue after close() |
DlqAbortedError | class | Thrown by enqueue after signal abort |
CircuitBreaker
| Member | Signature | Description |
|---|---|---|
execute | <T>(fn: () => Promise<T>) => Promise<T> | Runs fn; throws CircuitBreakerOpenError when open |
state | CircuitStateType | Current circuit state |
reset | () => void | Forces circuit closed |
forceOpen | () => void | Forces circuit open |
forceClosed | () => void | Alias for reset |
TokenBucket
| Member | Signature | Description |
|---|---|---|
consume | (tokens?: number) => void | Consumes tokens; throws TokenBucketExhaustedError if insufficient |
waitForToken | (tokens?, signal?) => Promise<void> | Waits until tokens are available, then consumes |
available | number | Current token count (triggers a refill calculation) |
DeadLetterQueue<T>
| Member | Signature | Description |
|---|---|---|
enqueue | (item, reason, error?) => void | Adds item; throws on full, closed, or aborted |
drain | () => AsyncGenerator<DlqEntryType<T>> | Yields all entries; suspends when queue is empty |
close | () => void | Signals drain to stop after the current entries |
abort | () => void | Immediately stops drain |
size | number | Current entry count |