From a32be8d594fdbf7ddf6686792b54731662515f95 Mon Sep 17 00:00:00 2001 From: Renato Alcara Date: Mon, 27 Apr 2026 01:06:25 -0300 Subject: [PATCH] Feat/replace circuit breaker with bounded retry (#393) Feat/replace circuit breaker with bounded retry (#393) --- src/Defaults/index.ts | 2 - src/Signal/libsignal.ts | 18 +- src/Socket/socket.ts | 262 +++---- src/Types/Socket.ts | 18 - src/Utils/bounded-retry.ts | 393 ++++++++++ src/Utils/circuit-breaker.ts | 766 -------------------- src/Utils/health-status.ts | 45 +- src/Utils/index.ts | 2 +- src/Utils/prometheus-metrics.ts | 23 - src/Utils/retry-utils.ts | 161 +--- src/Utils/structured-logger.ts | 120 +-- src/Utils/unified-session.ts | 46 +- src/__tests__/Utils/bounded-retry.test.ts | 391 ++++++++++ src/__tests__/Utils/circuit-breaker.test.ts | 371 ---------- src/__tests__/Utils/retry-utils.test.ts | 101 +-- src/__tests__/Utils/unified-session.test.ts | 28 +- 16 files changed, 944 insertions(+), 1803 deletions(-) create mode 100644 src/Utils/bounded-retry.ts delete mode 100644 src/Utils/circuit-breaker.ts create mode 100644 src/__tests__/Utils/bounded-retry.test.ts delete mode 100644 src/__tests__/Utils/circuit-breaker.test.ts diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index 11885d93..e36976a3 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -142,8 +142,6 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { getMessage: async () => undefined, cachedGroupMetadata: async () => undefined, makeSignalRepository: makeLibSignalRepository, - // Circuit breaker configuration - enableCircuitBreaker: true, // Listener limits (memory leak prevention) // WebSocket: 8 core events (open, close, error, message, ping, pong, upgrade, unexpected-response) // + 10 dynamic listeners (reconnect handlers, custom events) diff --git a/src/Signal/libsignal.ts b/src/Signal/libsignal.ts index 8ef48a8f..26b6e8e9 100644 --- a/src/Signal/libsignal.ts +++ b/src/Signal/libsignal.ts @@ -6,7 +6,6 @@ import type { LIDMapping, SignalAuthState, SignalKeyStoreWithTransaction } from import type { BaileysEventEmitter } from '../Types/Events' import type { SignalRepositoryWithLIDStore } from '../Types/Signal' import { generateSignalPubKey } from '../Utils' -import { CircuitBreaker } from '../Utils/circuit-breaker.js' import type { ILogger } from '../Utils/logger' import { metrics } from '../Utils/prometheus-metrics.js' import { isAnyLidUser, isAnyPnUser, jidDecode, transferDevice, WAJIDDomains } from '../WABinary' @@ -67,8 +66,6 @@ export interface IdentitySaveResult { export interface LibSignalRepositoryOptions { /** Event emitter for broadcasting identity changes */ ev?: BaileysEventEmitter - /** Circuit breaker for prekey operations (optional) */ - preKeyCircuitBreaker?: CircuitBreaker } // ============================================ @@ -256,7 +253,7 @@ export function makeLibSignalRepository( pnToLIDFunc?: (jids: string[]) => Promise, options?: LibSignalRepositoryOptions ): SignalRepositoryWithLIDStore { - const { ev, preKeyCircuitBreaker } = options || {} + const { ev } = options || {} const lidMapping = new LIDMappingStore(auth.keys as SignalKeyStoreWithTransaction, logger, pnToLIDFunc) // Identity key cache to avoid repeated storage reads @@ -277,7 +274,7 @@ export function makeLibSignalRepository( cacheMetricsInterval.unref() } - const storage = signalStorage(auth, lidMapping, identityKeyCache, ev, preKeyCircuitBreaker, logger) + const storage = signalStorage(auth, lidMapping, identityKeyCache, ev, logger) const parsedKeys = auth.keys as SignalKeyStoreWithTransaction const migratedSessionCache = new LRUCache({ @@ -385,13 +382,9 @@ export function makeLibSignalRepository( 'Identity key changed - contact may have reinstalled WhatsApp, session will be re-established' ) - // Reset prekey circuit breaker since we identified the cause - // Reset regardless of state (could be open, half-open, or closed with accumulated failures) - // eslint-disable-next-line max-depth - if (preKeyCircuitBreaker) { - preKeyCircuitBreaker.reset() - logger.debug({ jid }, 'Reset prekey circuit breaker after identity key change detection') - } + // (No circuit breaker to reset — bounded-retry is stateless, + // each retry is independent so identity-change recovery + // happens automatically on the next operation.) } else if (saveResult.isNew) { logger.debug( { jid, addr: addrStr, fingerprint: saveResult.currentFingerprint }, @@ -769,7 +762,6 @@ function signalStorage( lidMapping: LIDMappingStore, identityKeyCache: LRUCache, ev?: BaileysEventEmitter, - preKeyCircuitBreaker?: CircuitBreaker, logger?: ILogger ): ExtendedSignalStorage { // Shared function to resolve PN signal address to LID if mapping exists diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index becfe657..992a3f1c 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -39,12 +39,7 @@ import { xmppSignedPreKey } from '../Utils' import { getPlatformId, isAndroidBrowser } from '../Utils/browser-utils' -import { - CircuitBreaker, - CircuitOpenError, - createConnectionCircuitBreaker, - createPreKeyCircuitBreaker -} from '../Utils/circuit-breaker' +import { withBoundedRetry } from '../Utils/bounded-retry' import { decrementActiveConnections, incrementActiveConnections, @@ -94,10 +89,6 @@ export const makeSocket = (config: SocketConfig) => { transactionOpts, qrTimeout, makeSignalRepository, - enableCircuitBreaker = true, - queryCircuitBreaker: queryCircuitBreakerConfig, - connectionCircuitBreaker: connectionCircuitBreakerConfig, - preKeyCircuitBreaker: preKeyCircuitBreakerConfig, // If enableUnifiedSession is explicitly set (true/false), use it // Otherwise (undefined), check env var, then default to true enableUnifiedSession: enableUnifiedSessionConfig @@ -107,57 +98,6 @@ export const makeSocket = (config: SocketConfig) => { const enableUnifiedSession = enableUnifiedSessionConfig !== undefined ? enableUnifiedSessionConfig : shouldEnableUnifiedSession() - // Initialize circuit breakers if enabled - let queryCircuitBreaker: CircuitBreaker | undefined - let connectionCircuitBreaker: CircuitBreaker | undefined - let preKeyCircuitBreaker: CircuitBreaker | undefined - - if (enableCircuitBreaker) { - // Circuit breaker for query operations (most critical) - queryCircuitBreaker = createConnectionCircuitBreaker({ - name: 'socket-query', - failureThreshold: 5, - failureWindow: 60000, - resetTimeout: 30000, - successThreshold: 2, - timeout: defaultQueryTimeoutMs || 60000, - onStateChange: (from, to) => { - logger.info({ from, to }, 'Query circuit breaker state changed') - }, - onOpen: () => { - logger.warn('Query circuit breaker OPENED - blocking requests') - }, - onClose: () => { - logger.info('Query circuit breaker CLOSED - resuming normal operation') - }, - ...queryCircuitBreakerConfig - }) - - // Circuit breaker for connection operations - connectionCircuitBreaker = createConnectionCircuitBreaker({ - name: 'socket-connection', - failureThreshold: 3, - failureWindow: 30000, - resetTimeout: 60000, - successThreshold: 1, - onStateChange: (from, to) => { - logger.info({ from, to }, 'Connection circuit breaker state changed') - }, - ...connectionCircuitBreakerConfig - }) - - // Circuit breaker for pre-key operations - preKeyCircuitBreaker = createPreKeyCircuitBreaker({ - name: 'socket-prekey', - onStateChange: (from, to) => { - logger.info({ from, to }, 'PreKey circuit breaker state changed') - }, - ...preKeyCircuitBreakerConfig - }) - - logger.info('Circuit breakers initialized for socket operations') - } - // Unified Session Manager will be initialized after sendNode is defined let unifiedSessionManager: UnifiedSessionManager | undefined @@ -238,20 +178,17 @@ export const makeSocket = (config: SocketConfig) => { }) } - /** send a raw buffer with circuit breaker protection */ + /** + * Send a raw buffer over the WebSocket. + * + * Replaces the previous connectionCircuitBreaker wrapping. WebSocket + * lifecycle errors (Connection Closed / Lost / Replaced) were already + * filtered out of the circuit's failure count, so the only failures it + * tripped on were rare native send() errors — for which a state-machine + * gate adds no value over a fast propagated error. Reconnection logic + * in makeSocket already handles WS recovery independently. + */ const sendRawMessage = async (data: Uint8Array | Buffer) => { - if (connectionCircuitBreaker) { - try { - return await connectionCircuitBreaker.execute(() => sendRawMessageInternal(data)) - } catch (error) { - if (error instanceof CircuitOpenError) { - logger.warn({ circuitName: error.circuitName }, 'Send blocked by connection circuit breaker') - } - - throw error - } - } - return sendRawMessageInternal(data) } @@ -266,7 +203,7 @@ export const makeSocket = (config: SocketConfig) => { } // Initialize Unified Session Manager now that sendNode is defined - // (single initialization to avoid duplicating circuit breakers and state) + // (single initialization to avoid duplicating manager state) if (enableUnifiedSession) { const sendNodeForSession = async (node: BinaryNode): Promise => { await sendNode(node) @@ -275,7 +212,6 @@ export const makeSocket = (config: SocketConfig) => { unifiedSessionManager = createUnifiedSessionManager({ enabled: true, logger, - enableCircuitBreaker, sendNode: sendNodeForSession }) logger.info('Unified session manager initialized') @@ -346,11 +282,8 @@ export const makeSocket = (config: SocketConfig) => { // Register the response listener BEFORE sending — avoids a race where the server // responds before we start listening. waitForMessage already handles its own // timeout (returns undefined) and connection-close errors (throws), so we do NOT - // wrap it in a second promiseTimeout. The outer wrapper caused a race condition - // where both timers fired at ~the same deadline: the outer one threw - // Boom('Timed Out') while sendNode was still pending, producing a spurious error - // whose message contained "socket-query" → matched the circuit-breaker's - // "socket" pattern → incorrectly tripped the breaker after 5 timeouts. + // wrap it in a second promiseTimeout. The outer wrapper would race the inner + // timeout near the deadline and throw a spurious Boom('Timed Out'). const responsePromise = waitForMessage(msgId, timeoutMs) // Prevent unhandled-rejection if sendNode throws before we reach // `await responsePromise` below. The error from sendNode still propagates @@ -372,23 +305,17 @@ export const makeSocket = (config: SocketConfig) => { return result } - /** send a query with circuit breaker protection */ + /** + * Send a query (iq stanza) and wait for its response. + * + * Replaces the previous queryCircuitBreaker wrapping (state-machine that + * blocked ALL queries for 30s after 5 failures). The cascade behavior + * was incompatible with WhatsApp Android's empirical retry pattern: + * each operation retries independently with exponential backoff, no + * global state. Callers that need retry semantics wrap query() with + * withBoundedRetry() at their level (assertSessions, etc.). + */ const query = async (node: BinaryNode, timeoutMs?: number) => { - // If circuit breaker is enabled, wrap the query - if (queryCircuitBreaker) { - try { - return await queryCircuitBreaker.execute(() => queryInternal(node, timeoutMs)) - } catch (error) { - // If circuit is open, log and rethrow with context - if (error instanceof CircuitOpenError) { - logger.warn({ circuitName: error.circuitName, state: error.state }, 'Query blocked by circuit breaker') - } - - throw error - } - } - - // Fallback to direct query if circuit breaker is disabled return queryInternal(node, timeoutMs) } @@ -685,20 +612,14 @@ export const makeSocket = (config: SocketConfig) => { let lastUploadTime = 0 /** generates and uploads a set of pre-keys to the server */ - const uploadPreKeys = async (count = MIN_PREKEY_COUNT, retryCount = 0) => { - // Check if pre-key circuit breaker is open - if (preKeyCircuitBreaker?.isOpen()) { - logger.warn('PreKey circuit breaker is open, skipping upload') - throw new CircuitOpenError('socket-prekey', 'open') - } - - // Check minimum interval (except for retries) - if (retryCount === 0) { - const timeSinceLastUpload = Date.now() - lastUploadTime - if (timeSinceLastUpload < MIN_UPLOAD_INTERVAL) { - logger.debug(`Skipping upload, only ${timeSinceLastUpload}ms since last upload`) - return - } + const uploadPreKeys = async (count = MIN_PREKEY_COUNT) => { + // Check minimum interval. (Previously this was guarded by `retryCount === 0` + // because the function recursed on retry; with bounded-retry handling + // retries internally there is no recursion and the guard is implicit.) + const timeSinceLastUpload = Date.now() - lastUploadTime + if (timeSinceLastUpload < MIN_UPLOAD_INTERVAL) { + logger.debug(`Skipping upload, only ${timeSinceLastUpload}ms since last upload`) + return } // Prevent multiple concurrent uploads — if one is already running, wait for it and return: @@ -709,8 +630,15 @@ export const makeSocket = (config: SocketConfig) => { return } + // Shared abort controller so the outer Promise.race can cancel the + // in-flight bounded-retry loop if its own timeout fires first. + // Without this, the outer race rejects but the bounded-retry loop + // keeps running in the background, mutating lastUploadTime / logs + // after the caller has given up (CodeRabbit + Copilot reviews). + const uploadAbort = new AbortController() + const uploadLogic = async () => { - logger.info({ count, retryCount }, 'uploading pre-keys') + logger.info({ count }, 'uploading pre-keys') // Generate and save pre-keys atomically (prevents ID collisions on retry) const node = await keys.transaction(async () => { @@ -721,45 +649,69 @@ export const makeSocket = (config: SocketConfig) => { return node // Only return node since update is already used }, creds?.me?.id || 'upload-pre-keys') - // Upload to server with circuit breaker protection - const uploadToServer = async () => { - await query(node) + // Upload to server. Bounded-retry handles backoff (replaces both + // the previous circuit breaker AND the manual exponential backoff + // retry loop with maxRetries=3). + // + // CRITICAL: query() returns `undefined` on timeout (it does NOT + // throw — see waitForMessage). If we just `await query(node)`, + // a timed-out upload would resolve with undefined and bounded-retry + // would treat it as success. Validate the response shape so a + // timeout actually triggers a retry (CodeRabbit review on PR #393). + // + // bounded-retry's ttlMs MUST be < UPLOAD_TIMEOUT (the outer + // Promise.race below). With margin (28s vs 30s) bounded-retry + // always reaches its natural give-up first. We additionally pass + // an AbortController signal so that if the outer race ever fires + // first, the bounded-retry loop is aborted and does not keep + // running in the background (CodeRabbit + Copilot reviews). + const PER_ATTEMPT_TIMEOUT_MS = 8_000 + const RETRY_TTL_MS = UPLOAD_TIMEOUT - 2_000 + + const uploadToServer = async (signal?: AbortSignal) => { + const result = await query(node, PER_ATTEMPT_TIMEOUT_MS) + if (signal?.aborted) { + throw new Error('aborted') + } + + if (!result) { + // query() returned undefined → underlying iq response + // timed out. Throw so bounded-retry retries. + throw new Boom('Pre-key upload query timed out (no response)', { statusCode: 408 }) + } + logger.info({ count }, 'uploaded pre-keys successfully') lastUploadTime = Date.now() } try { - // Use circuit breaker if available - if (preKeyCircuitBreaker) { - await preKeyCircuitBreaker.execute(uploadToServer) - } else { - await uploadToServer() - } + await withBoundedRetry(uploadToServer, { + name: 'uploadPreKeys', + // 1s -> 2s -> 4s -> 8s (cap). Max 4 attempts fit comfortably + // within RETRY_TTL_MS (28s) given perAttemptTimeoutMs=8s. + delays: [1000, 2000, 4000, 8000], + ttlMs: RETRY_TTL_MS, + perAttemptTimeoutMs: PER_ATTEMPT_TIMEOUT_MS, + signal: uploadAbort.signal, + logger + }) } catch (uploadError) { logger.error({ uploadError: (uploadError as Error).toString(), count }, 'Failed to upload pre-keys to server') - - // Don't retry if circuit breaker is open - if (uploadError instanceof CircuitOpenError) { - throw uploadError - } - - // Exponential backoff retry (max 3 retries) - if (retryCount < 3) { - const backoffDelay = Math.min(1000 * Math.pow(2, retryCount), 10000) - logger.info(`Retrying pre-key upload in ${backoffDelay}ms`) - await new Promise(resolve => setTimeout(resolve, backoffDelay)) - return uploadPreKeys(count, retryCount + 1) - } - throw uploadError } } - // Add timeout protection + // Outer timeout protection. With ttlMs=28s vs UPLOAD_TIMEOUT=30s, + // bounded-retry should always win — but if anything ever blocks + // uploadLogic outside bounded-retry's reach, this race aborts the + // whole thing cleanly via uploadAbort. uploadPreKeysPromise = Promise.race([ uploadLogic(), new Promise((_, reject) => - setTimeout(() => reject(new Boom('Pre-key upload timeout', { statusCode: 408 })), UPLOAD_TIMEOUT) + setTimeout(() => { + uploadAbort.abort() + reject(new Boom('Pre-key upload timeout', { statusCode: 408 })) + }, UPLOAD_TIMEOUT) ) ]) @@ -1231,12 +1183,6 @@ export const makeSocket = (config: SocketConfig) => { cleanupPreKeyAutoSync() cleanupSessionTTL() - // CRITICAL: Destroy circuit breakers AFTER cleanup functions complete - // This ensures cleanup functions can still use circuit breakers if needed - queryCircuitBreaker?.destroy() - connectionCircuitBreaker?.destroy() - preKeyCircuitBreaker?.destroy() - // IMPORTANT: Do NOT use removeAllListeners('connection.update') // It would remove consumer listeners, breaking their reconnection logic } @@ -1289,11 +1235,12 @@ export const makeSocket = (config: SocketConfig) => { return // connection closing — do not reschedule } else if (ws.isOpen) { // Send keep-alive ping via sendNode() (fire-and-forget) instead of query(). - // query() wraps the ping in the query circuit breaker — when that breaker is - // open or timing out, the ping is never sent, WA never responds, lastDateRecv - // goes stale, and the diff check above wrongly fires "Connection was lost". - // sendNode() bypasses the query circuit breaker entirely; WA's ping response - // still arrives as an incoming frame and updates lastDateRecv normally. + // We don't need the response correlator overhead — WA's pong arrives as an + // incoming frame and updates lastDateRecv via the standard receive path. + // (Historical context: this used to bypass the query circuit breaker, + // which would block pings during cascade failures. Circuit breaker has + // since been replaced by per-operation bounded-retry — but the + // fire-and-forget pattern stays correct on its own merits.) sendNode({ tag: 'iq', attrs: { @@ -1809,25 +1756,6 @@ export const makeSocket = (config: SocketConfig) => { sendWAMBuffer, executeUSyncQuery, onWhatsApp, - // Circuit breaker utilities - circuitBreakers: { - query: queryCircuitBreaker, - connection: connectionCircuitBreaker, - preKey: preKeyCircuitBreaker - }, - /** Get circuit breaker statistics */ - getCircuitBreakerStats: () => ({ - query: queryCircuitBreaker?.getStats(), - connection: connectionCircuitBreaker?.getStats(), - preKey: preKeyCircuitBreaker?.getStats() - }), - /** Reset all circuit breakers to closed state */ - resetCircuitBreakers: () => { - queryCircuitBreaker?.reset() - connectionCircuitBreaker?.reset() - preKeyCircuitBreaker?.reset() - logger.info('All circuit breakers reset to closed state') - }, // Unified Session Telemetry /** Send unified_session telemetry manually */ sendUnifiedSession, diff --git a/src/Types/Socket.ts b/src/Types/Socket.ts index 800f4cba..38e97cd6 100644 --- a/src/Types/Socket.ts +++ b/src/Types/Socket.ts @@ -1,7 +1,6 @@ import type { Agent } from 'https' import type { URL } from 'url' import { proto } from '../../WAProto/index.js' -import type { CircuitBreakerOptions } from '../Utils/circuit-breaker' import type { ILogger } from '../Utils/logger' import type { AuthenticationState, LIDMapping, SignalAuthState, TransactionCapabilityOptions } from './Auth' import type { GroupMetadata } from './GroupMetadata' @@ -203,23 +202,6 @@ export type SocketConfig = { pnToLIDFunc?: (jids: string[]) => Promise ) => SignalRepositoryWithLIDStore - // === Circuit Breaker Configuration === - - /** Enable circuit breaker protection for socket operations (default: true) */ - enableCircuitBreaker?: boolean - - /** Circuit breaker configuration for query operations */ - queryCircuitBreaker?: Partial - - /** Circuit breaker configuration for connection operations */ - connectionCircuitBreaker?: Partial - - /** Circuit breaker configuration for pre-key operations */ - preKeyCircuitBreaker?: Partial - - /** Circuit breaker configuration for message operations */ - messageCircuitBreaker?: Partial - // === Listener Limits (Memory Leak Prevention) === /** diff --git a/src/Utils/bounded-retry.ts b/src/Utils/bounded-retry.ts new file mode 100644 index 00000000..fdc376ca --- /dev/null +++ b/src/Utils/bounded-retry.ts @@ -0,0 +1,393 @@ +/** + * Bounded retry — WhatsApp-aligned per-operation retry without global state. + * + * # Why this exists + * + * Replaces the circuit breaker pattern that was causing cascading failures + * in production: 5 timeouts in 60s would trip the global breaker, blocking + * EVERY socket query (typing indicators, profile pic fetches, contact + * validation, presence updates) for 30s — even queries to peers that were + * perfectly healthy. + * + * # Empirical justification + * + * Captured WhatsApp Android's actual retry behavior via Frida hooks + * (`hook-circuit-breaker-re-v2.js`, `hook-retry-bounds.js`) on a real + * device under controlled WiFi off/on cycles. Findings: + * + * 1. Delay sequence (per-operation): 3s -> 10s -> 60s -> ~64s -> 120s + * (cap at 2 min). Last value reused for further attempts. + * 2. Memory profile during 5-min network outage: PSS dropped 53MB and + * stabilised. FDs closed (305 -> 295). NO unbounded accumulation. + * 3. Recovery on reconnect: ~10s. No retry storm. + * 4. NO global state machine observed. Each operation has its own timer. + * Failures of operation A do NOT block operation B. + * + * The default delays in WHATSAPP_BACKOFF_DELAYS below match the captured + * sequence directly. The default 10-min TTL is empirical 3-5 min stability + * + safety buffer. + * + * # Design properties + * + * - Per-operation isolation: each call to `withBoundedRetry` has its own + * timer. No shared state. Operation A failing has zero effect on B. + * - Bounded by `ttlMs` (wall-clock budget). The TTL is enforced strictly + * at three points: (a) before each attempt, (b) cap on per-attempt + * timeout, (c) cap on retry delay. + * - Per-attempt timeout: each attempt has its own deadline (default 30s), + * automatically capped to remaining TTL budget so total runtime never + * exceeds ttlMs. + * - AbortSignal cancellation: external cancellation supported, both for + * the sleep between retries AND (optionally) the in-flight operation + * when the operation accepts a signal parameter. + * - Memory bound: at most one outstanding retry timer per call. Listeners + * are explicitly removed when timers settle. Once the call resolves / + * rejects / aborts, all state is freed. + * + * # Logging + * + * Pass `logger` in options to get structured logs at each retry attempt, + * give-up, and post-failure recovery. The module emits Prometheus metrics + * regardless of whether a logger is provided. + * + * # When to use this vs. plain query() + * + * - Use plain `query()` when you want fast-fail semantics (caller decides + * what to do on failure). Most call sites in InfiniteAPI use this. + * - Use `withBoundedRetry(() => query(...), { name: 'X' })` when the + * operation is "must-eventually-succeed" with no upstream retry — e.g. + * `uploadPreKeys` or other write paths where the alternative is data loss. + * + * # Examples + * + * ```ts + * // Single attempt with 5-min TTL, give up after that: + * await withBoundedRetry( + * () => assertSessions([jid], true), + * { name: 'assertSessions', ttlMs: 5 * 60_000, logger } + * ) + * + * // Tight deadline, fast give-up: + * await withBoundedRetry( + * () => sendNode(node), + * { name: 'send', ttlMs: 30_000, perAttemptTimeoutMs: 5_000 } + * ) + * + * // Operation that respects abort signal (best practice): + * await withBoundedRetry( + * (signal) => fetchSomething({ signal }), + * { name: 'fetch', logger } + * ) + * ``` + * + * @module Utils/bounded-retry + */ + +import type { ILogger } from './logger' +import { metrics } from './prometheus-metrics.js' + +/** + * Default delay sequence (milliseconds) — matches WhatsApp Android empirical + * behavior. After exhausting the sequence, the last value is used (cap). + */ +export const WHATSAPP_BACKOFF_DELAYS = [3000, 10000, 60000, 60000, 120000] as const + +/** + * Default jitter (+/- 15%) to prevent thundering-herd retries. + */ +export const DEFAULT_JITTER_FACTOR = 0.15 as const + +/** + * Default time-to-live for retries: 10 minutes. + * + * Empirical justification: WhatsApp Android stabilises memory in ~3-5 min + * during a network outage; 10 min gives a generous safety buffer while + * preventing unbounded retry accumulation. + */ +export const DEFAULT_TTL_MS = 10 * 60 * 1000 + +/** + * Default per-attempt timeout: 30 seconds. + * + * Most WhatsApp queries respond within seconds. A 30s timeout is generous + * enough for slow networks but prevents a single hang from blocking retries. + */ +export const DEFAULT_PER_ATTEMPT_TIMEOUT_MS = 30000 + +export interface BoundedRetryOptions { + /** Operation name for logging/metrics */ + name?: string + /** Sequence of delays (ms). Last value is used as cap. */ + delays?: readonly number[] + /** Jitter factor 0..1 (default 0.15) */ + jitter?: number + /** Total wall-clock budget — gives up after this. Default 10 min. */ + ttlMs?: number + /** Per-attempt timeout (ms). Default 30s. Capped by remaining TTL. */ + perAttemptTimeoutMs?: number + /** Predicate: should we retry on this error? Default: always */ + shouldRetry?: (err: Error, attempt: number) => boolean + /** Hook fired before each retry */ + onRetry?: (err: Error, attempt: number, delayMs: number) => void + /** + * AbortSignal — when fired, the loop stops at the next observation point: + * - If the loop is sleeping between retries, the sleep rejects immediately. + * - If an attempt is in flight, the abort is forwarded to the operation + * via the per-attempt signal. The operation must ITSELF observe the + * signal (e.g. `(signal) => fetch({ signal })`) for cancellation to + * be truly immediate; otherwise it cancels at the next per-attempt + * timeout boundary. + * - Either way, on the next loop iteration BoundedRetryAbortedError is + * thrown, so the caller never sees more than one trailing attempt + * after the abort. + */ + signal?: AbortSignal + /** Optional logger for structured retry/give-up/recovery logs */ + logger?: ILogger +} + +export class BoundedRetryGiveUpError extends Error { + constructor( + public readonly opName: string, + public readonly attempts: number, + public readonly elapsedMs: number, + public readonly lastError: Error + ) { + super( + `bounded-retry "${opName}" gave up after ${attempts} attempts ` + + `(${elapsedMs}ms elapsed). Last error: ${lastError.message}` + ) + this.name = 'BoundedRetryGiveUpError' + } +} + +export class BoundedRetryAbortedError extends Error { + constructor(public readonly opName: string) { + super(`bounded-retry "${opName}" aborted via signal`) + this.name = 'BoundedRetryAbortedError' + } +} + +/** + * Apply jitter to a delay: returns delay * (1 +/- jitter) + */ +function withJitter(delayMs: number, jitter: number): number { + if (jitter <= 0) return delayMs + const factor = 1 + (Math.random() * 2 - 1) * jitter + return Math.max(0, Math.round(delayMs * factor)) +} + +/** + * Pick the delay for a given attempt index. Falls back to the last value + * (cap) once the sequence is exhausted. + */ +function pickDelay(attempt: number, delays: readonly number[]): number { + if (delays.length === 0) return 0 + if (attempt < delays.length) return delays[attempt]! + return delays[delays.length - 1]! +} + +/** + * Wrap a promise with a per-attempt timeout. Aborts the supplied controller + * when the timeout fires so the caller can cancel any in-flight work + * (operations that accept the signal). Always clears the timer on settlement. + */ +function withTimeout( + promise: Promise, + timeoutMs: number, + name: string, + abortOnTimeout: AbortController +): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + abortOnTimeout.abort() + reject(new Error(`bounded-retry "${name}" attempt timed out after ${timeoutMs}ms`)) + }, timeoutMs) + promise + .then(value => { + clearTimeout(timer) + resolve(value) + }) + .catch(err => { + clearTimeout(timer) + reject(err as Error) + }) + }) +} + +/** + * Sleep with abort support. Always removes the abort listener on settlement + * so listeners do not accumulate on long-lived signals. + */ +function sleep(ms: number, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new Error('aborted')) + return + } + + let onAbort: (() => void) | undefined + const cleanup = () => { + if (onAbort && signal) { + signal.removeEventListener('abort', onAbort) + } + } + + const timer = setTimeout(() => { + cleanup() + resolve() + }, ms) + + if (signal) { + onAbort = () => { + clearTimeout(timer) + cleanup() + reject(new Error('aborted')) + } + + signal.addEventListener('abort', onAbort, { once: true }) + } + }) +} + +/** + * Run an async operation with bounded exponential-backoff retry. + * + * Independent per-call: no global state, no cross-operation interaction. + * Memory bound: at most one outstanding retry timer per call. + * + * The operation may optionally accept an `AbortSignal` parameter — when the + * per-attempt timeout fires (or the outer signal is aborted), the inner + * signal is aborted so the operation can stop in-flight work cleanly. + * + * @example + * ```ts + * const result = await withBoundedRetry( + * (signal) => assertSessions([jid], true, { signal }), + * { name: 'assertSessions', ttlMs: 5 * 60_000, logger } + * ) + * ``` + */ +export async function withBoundedRetry( + operation: (signal?: AbortSignal) => Promise, + options: BoundedRetryOptions = {} +): Promise { + const name = options.name ?? 'bounded-retry' + const delays = options.delays ?? WHATSAPP_BACKOFF_DELAYS + const jitter = options.jitter ?? DEFAULT_JITTER_FACTOR + const ttlMs = options.ttlMs ?? DEFAULT_TTL_MS + const perAttemptTimeoutMs = options.perAttemptTimeoutMs ?? DEFAULT_PER_ATTEMPT_TIMEOUT_MS + const shouldRetry = options.shouldRetry ?? (() => true) + const logger = options.logger + + const start = Date.now() + let lastError: Error = new Error('unknown') + let attempt = 0 + + while (true) { + // Check abort + TTL BEFORE starting an attempt — strict wall-clock + // budget. Without this check the loop could begin a new attempt with + // 0ms remaining and then run for up to `perAttemptTimeoutMs`. + if (options.signal?.aborted) { + throw new BoundedRetryAbortedError(name) + } + + const elapsedBeforeAttempt = Date.now() - start + const remainingBudget = ttlMs - elapsedBeforeAttempt + if (remainingBudget <= 0) { + metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' }) + logger?.warn?.( + { op: name, attempts: attempt, elapsedMs: elapsedBeforeAttempt, ttlMs, lastError: lastError.message }, + 'bounded-retry: TTL exceeded before next attempt — giving up' + ) + throw new BoundedRetryGiveUpError(name, attempt, elapsedBeforeAttempt, lastError) + } + + // Cap the per-attempt timeout to the remaining TTL budget so a single + // attempt cannot run past the wall-clock deadline. + const attemptTimeoutMs = Math.min(perAttemptTimeoutMs, remainingBudget) + const attemptAbort = new AbortController() + + // Forward outer abort to the per-attempt controller so the in-flight + // operation is cancelled when the user aborts. Cleaned up below. + let onOuterAbort: (() => void) | undefined + if (options.signal) { + onOuterAbort = () => attemptAbort.abort() + if (options.signal.aborted) { + attemptAbort.abort() + } else { + options.signal.addEventListener('abort', onOuterAbort, { once: true }) + } + } + + try { + const result = await withTimeout(operation(attemptAbort.signal), attemptTimeoutMs, name, attemptAbort) + + if (attempt > 0) { + metrics.socketEvents?.inc({ event: 'bounded_retry_recovered' }) + logger?.info?.( + { op: name, attempts: attempt + 1, elapsedMs: Date.now() - start }, + 'bounded-retry: operation succeeded after retries' + ) + } + + return result + } catch (err) { + lastError = err as Error + attempt++ + + // (outer-abort listener detachment happens in the finally below — + // avoiding a double-remove that breaks listener-count assertions.) + + // If the outer signal aborted us mid-attempt, surface that explicitly + // rather than as a generic operation failure. + if (options.signal?.aborted) { + throw new BoundedRetryAbortedError(name) + } + + const elapsed = Date.now() - start + + if (!shouldRetry(lastError, attempt)) { + metrics.errors?.inc({ category: 'bounded_retry', code: 'predicate_no_retry' }) + logger?.warn?.( + { op: name, attempts: attempt, elapsedMs: elapsed, error: lastError.message }, + 'bounded-retry: shouldRetry returned false — giving up' + ) + throw lastError + } + + // Re-check budget after the failure so we do not sleep past TTL. + const remainingAfterFailure = ttlMs - elapsed + if (remainingAfterFailure <= 0) { + metrics.errors?.inc({ category: 'bounded_retry', code: 'ttl_exceeded' }) + logger?.warn?.( + { op: name, attempts: attempt, elapsedMs: elapsed, ttlMs, lastError: lastError.message }, + 'bounded-retry: TTL exceeded after attempt — giving up' + ) + throw new BoundedRetryGiveUpError(name, attempt, elapsed, lastError) + } + + const baseDelay = pickDelay(attempt - 1, delays) + const delayMs = Math.min(withJitter(baseDelay, jitter), remainingAfterFailure) + + options.onRetry?.(lastError, attempt, delayMs) + metrics.socketEvents?.inc({ event: 'bounded_retry_attempt' }) + logger?.debug?.( + { op: name, attempt, delayMs, elapsedMs: elapsed, error: lastError.message }, + 'bounded-retry: scheduling next attempt' + ) + + try { + await sleep(delayMs, options.signal) + } catch { + throw new BoundedRetryAbortedError(name) + } + } finally { + // Belt-and-suspenders: ensure the outer-abort listener is always + // removed even on early throws. + if (options.signal && onOuterAbort) { + options.signal.removeEventListener('abort', onOuterAbort) + } + } + } +} diff --git a/src/Utils/circuit-breaker.ts b/src/Utils/circuit-breaker.ts deleted file mode 100644 index 00be9cc6..00000000 --- a/src/Utils/circuit-breaker.ts +++ /dev/null @@ -1,766 +0,0 @@ -/** - * Circuit Breaker Pattern Implementation - * - * Provides protection against cascading failures by monitoring operation outcomes - * and temporarily blocking requests when failure thresholds are exceeded. - * - * States: - * - CLOSED: Normal operation, requests pass through - * - OPEN: Failures exceeded threshold, requests are blocked - * - HALF_OPEN: Testing recovery, limited requests allowed - * - * @module Utils/circuit-breaker - */ - -import { EventEmitter } from 'events' -import { metrics } from './prometheus-metrics.js' - -/** - * Circuit breaker operational states - */ -export type CircuitState = 'closed' | 'open' | 'half-open' - -/** - * Failure record with timestamp for sliding window tracking - */ -export interface FailureRecord { - timestamp: number - error: Error -} - -/** - * Circuit breaker configuration options - */ -export interface CircuitBreakerOptions { - /** Unique identifier for this circuit breaker (used in metrics/logging) */ - name: string - /** Number of failures within the window to trigger OPEN state (default: 5) */ - failureThreshold?: number - /** Time window in ms for counting failures (default: 60000) */ - failureWindow?: number - /** Number of successes required in HALF_OPEN to return to CLOSED (default: 2) */ - successThreshold?: number - /** Time in ms to wait before transitioning from OPEN to HALF_OPEN (default: 30000) */ - resetTimeout?: number - /** Timeout for individual operations in ms (default: 10000) */ - timeout?: number - /** Minimum number of requests before circuit can trip (default: 5) */ - volumeThreshold?: number - /** Predicate to determine if an error should count as a failure */ - shouldCountError?: (error: Error) => boolean - /** Whether to collect Prometheus metrics (default: true) */ - collectMetrics?: boolean - /** Fallback function when circuit is OPEN */ - fallback?: () => T | Promise - /** Callback when state changes */ - onStateChange?: (from: CircuitState, to: CircuitState) => void - /** Callback on failure */ - onFailure?: (error: Error) => void - /** Callback on success */ - onSuccess?: () => void - /** Callback when circuit opens */ - onOpen?: () => void - /** Callback when circuit closes */ - onClose?: () => void - /** Callback when circuit enters half-open */ - onHalfOpen?: () => void -} - -/** - * Circuit breaker statistics - */ -export interface CircuitBreakerStats { - state: CircuitState - failures: number - successes: number - consecutiveFailures: number - consecutiveSuccesses: number - totalCalls: number - totalFailures: number - totalSuccesses: number - totalRejected: number - failureRate: number - lastFailureTime?: number - lastSuccessTime?: number - lastStateChange?: number - isOpen: boolean - isClosed: boolean - isHalfOpen: boolean -} - -/** - * Error thrown when circuit is OPEN and request is rejected - */ -export class CircuitOpenError extends Error { - constructor( - public readonly circuitName: string, - public readonly state: CircuitState - ) { - super(`Circuit breaker "${circuitName}" is ${state}`) - this.name = 'CircuitOpenError' - } -} - -/** - * Error thrown when operation exceeds timeout - */ -export class CircuitTimeoutError extends Error { - constructor( - public readonly circuitName: string, - public readonly timeoutMs: number - ) { - super(`Circuit breaker "${circuitName}" operation timed out after ${timeoutMs}ms`) - this.name = 'CircuitTimeoutError' - } -} - -/** - * Circuit Breaker implementation with sliding window failure tracking - * - * @example - * ```typescript - * const breaker = new CircuitBreaker({ - * name: 'whatsapp-api', - * failureThreshold: 5, - * resetTimeout: 30000 - * }) - * - * try { - * const result = await breaker.execute(() => sendMessage(msg)) - * } catch (error) { - * if (error instanceof CircuitOpenError) { - * // Circuit is open, use fallback - * } - * } - * ``` - */ -export class CircuitBreaker extends EventEmitter { - private state: CircuitState = 'closed' - private failureRecords: FailureRecord[] = [] - private consecutiveFailures = 0 - private consecutiveSuccesses = 0 - private totalCalls = 0 - private totalFailures = 0 - private totalSuccesses = 0 - private totalRejected = 0 - private lastFailureTime?: number - private lastSuccessTime?: number - private lastStateChange: number - private resetTimer?: ReturnType - private readonly options: Required - - constructor(options: CircuitBreakerOptions) { - super() - - this.options = { - name: options.name, - failureThreshold: options.failureThreshold ?? 5, - failureWindow: options.failureWindow ?? 60000, - successThreshold: options.successThreshold ?? 2, - resetTimeout: options.resetTimeout ?? 30000, - timeout: options.timeout ?? 10000, - volumeThreshold: options.volumeThreshold ?? 5, - shouldCountError: options.shouldCountError ?? (() => true), - collectMetrics: options.collectMetrics ?? true, - fallback: - options.fallback ?? - (() => { - throw new CircuitOpenError(this.options.name, this.state) - }), - onStateChange: options.onStateChange ?? (() => {}), - onFailure: options.onFailure ?? (() => {}), - onSuccess: options.onSuccess ?? (() => {}), - onOpen: options.onOpen ?? (() => {}), - onClose: options.onClose ?? (() => {}), - onHalfOpen: options.onHalfOpen ?? (() => {}) - } - - this.lastStateChange = Date.now() - } - - /** - * Check if the circuit allows execution - */ - canExecute(): boolean { - if (this.state === 'closed') { - return true - } - - if (this.state === 'open') { - return false - } - - // HALF_OPEN: allow limited requests for testing - return true - } - - /** - * Execute an async operation with circuit breaker protection - */ - async execute(operation: () => T | Promise): Promise { - this.totalCalls++ - - // Check if circuit allows execution - if (this.state === 'open') { - this.totalRejected++ - - if (this.options.collectMetrics) { - metrics.errors.inc({ category: 'circuit_breaker', code: 'rejected' }) - } - - return this.options.fallback() as T - } - - // Execute with timeout protection - try { - const result = await this.executeWithTimeout(operation) - this.recordSuccess() - return result - } catch (error) { - this.recordFailure(error as Error) - throw error - } - } - - /** - * Execute a synchronous operation with circuit breaker protection - */ - executeSync(operation: () => T): T { - this.totalCalls++ - - if (this.state === 'open') { - this.totalRejected++ - - if (this.options.collectMetrics) { - metrics.errors.inc({ category: 'circuit_breaker', code: 'rejected' }) - } - - return this.options.fallback() as T - } - - try { - const result = operation() - this.recordSuccess() - return result - } catch (error) { - this.recordFailure(error as Error) - throw error - } - } - - /** - * Execute operation with timeout wrapper - */ - private async executeWithTimeout(operation: () => T | Promise): Promise { - return new Promise((resolve, reject) => { - const timer = setTimeout(() => { - reject(new CircuitTimeoutError(this.options.name, this.options.timeout)) - }, this.options.timeout) - - Promise.resolve(operation()) - .then(result => { - clearTimeout(timer) - resolve(result) - }) - .catch(error => { - clearTimeout(timer) - reject(error) - }) - }) - } - - /** - * Record a successful operation - */ - private recordSuccess(): void { - this.totalSuccesses++ - this.lastSuccessTime = Date.now() - this.consecutiveSuccesses++ - this.consecutiveFailures = 0 - - if (this.options.collectMetrics) { - metrics.socketEvents.inc({ event: 'circuit_success' }) - } - - this.options.onSuccess() - this.emit('success') - - // In HALF_OPEN state, check if we can close the circuit - if (this.state === 'half-open') { - if (this.consecutiveSuccesses >= this.options.successThreshold) { - this.transitionTo('closed') - } - } - } - - /** - * Record a failed operation - */ - private recordFailure(error: Error): void { - // Check if this error should count as a failure - if (!this.options.shouldCountError(error)) { - return - } - - const now = Date.now() - this.totalFailures++ - this.lastFailureTime = now - this.consecutiveFailures++ - this.consecutiveSuccesses = 0 - - // Add to failure records for sliding window - this.failureRecords.push({ timestamp: now, error }) - - // Clean old failures outside the window - this.cleanOldFailures() - - if (this.options.collectMetrics) { - metrics.errors.inc({ category: 'circuit_breaker', code: 'failure' }) - } - - this.options.onFailure(error) - this.emit('failure', error) - - // State transition logic - if (this.state === 'half-open') { - // Any failure in HALF_OPEN immediately reopens the circuit - this.transitionTo('open') - } else if (this.state === 'closed') { - // Check if we should trip the circuit - const recentFailures = this.failureRecords.length - - if (this.totalCalls >= this.options.volumeThreshold && recentFailures >= this.options.failureThreshold) { - this.transitionTo('open') - } - } - } - - /** - * Remove failure records outside the sliding window - */ - private cleanOldFailures(): void { - const cutoff = Date.now() - this.options.failureWindow - this.failureRecords = this.failureRecords.filter(record => record.timestamp > cutoff) - } - - /** - * Transition to a new state - */ - private transitionTo(newState: CircuitState): void { - const oldState = this.state - - if (oldState === newState) { - return - } - - this.state = newState - this.lastStateChange = Date.now() - - // Clear existing reset timer - if (this.resetTimer) { - clearTimeout(this.resetTimer) - this.resetTimer = undefined - } - - // State-specific actions - switch (newState) { - case 'closed': - this.consecutiveFailures = 0 - this.consecutiveSuccesses = 0 - this.failureRecords = [] - this.options.onClose() - this.emit('close') - break - - case 'open': - this.consecutiveSuccesses = 0 - this.options.onOpen() - this.emit('open') - - // Record circuit breaker trip metric - if (this.options.collectMetrics) { - metrics.circuitBreakerTrips?.inc({ name: this.options.name }) - } - - // Schedule transition to HALF_OPEN - this.resetTimer = setTimeout(() => { - this.transitionTo('half-open') - }, this.options.resetTimeout) - break - - case 'half-open': - this.consecutiveSuccesses = 0 - this.consecutiveFailures = 0 - this.options.onHalfOpen() - this.emit('half-open') - break - } - - this.options.onStateChange(oldState, newState) - this.emit('state-change', { from: oldState, to: newState }) - } - - /** - * Manually trip the circuit to OPEN state - */ - trip(): void { - this.transitionTo('open') - } - - /** - * Manually reset the circuit to CLOSED state - */ - reset(): void { - this.transitionTo('closed') - } - - /** - * Get current circuit state - */ - getState(): CircuitState { - return this.state - } - - /** - * Check if circuit is OPEN - */ - isOpen(): boolean { - return this.state === 'open' - } - - /** - * Check if circuit is CLOSED - */ - isClosed(): boolean { - return this.state === 'closed' - } - - /** - * Check if circuit is HALF_OPEN - */ - isHalfOpen(): boolean { - return this.state === 'half-open' - } - - /** - * Get circuit breaker statistics - */ - getStats(): CircuitBreakerStats { - this.cleanOldFailures() - - const failureRate = this.totalCalls > 0 ? (this.totalFailures / this.totalCalls) * 100 : 0 - - return { - state: this.state, - failures: this.failureRecords.length, - successes: this.consecutiveSuccesses, - consecutiveFailures: this.consecutiveFailures, - consecutiveSuccesses: this.consecutiveSuccesses, - totalCalls: this.totalCalls, - totalFailures: this.totalFailures, - totalSuccesses: this.totalSuccesses, - totalRejected: this.totalRejected, - failureRate, - lastFailureTime: this.lastFailureTime, - lastSuccessTime: this.lastSuccessTime, - lastStateChange: this.lastStateChange, - isOpen: this.isOpen(), - isClosed: this.isClosed(), - isHalfOpen: this.isHalfOpen() - } - } - - /** - * Get circuit breaker name - */ - getName(): string { - return this.options.name - } - - /** - * Destroy circuit breaker and clean up resources - */ - destroy(): void { - if (this.resetTimer) { - clearTimeout(this.resetTimer) - } - - this.failureRecords = [] - this.removeAllListeners() - } -} - -/** - * Factory function to create a circuit breaker - */ -export function createCircuitBreaker(options: CircuitBreakerOptions): CircuitBreaker { - return new CircuitBreaker(options) -} - -/** - * Registry for managing multiple circuit breakers - */ -export class CircuitBreakerRegistry { - private breakers: Map = new Map() - - /** - * Get or create a circuit breaker by name - */ - get(name: string, options?: Omit): CircuitBreaker { - if (!this.breakers.has(name)) { - const breaker = new CircuitBreaker({ ...options, name }) - this.breakers.set(name, breaker) - } - - return this.breakers.get(name)! - } - - /** - * Check if a circuit breaker exists - */ - has(name: string): boolean { - return this.breakers.has(name) - } - - /** - * Remove a circuit breaker - */ - remove(name: string): boolean { - const breaker = this.breakers.get(name) - if (breaker) { - breaker.destroy() - return this.breakers.delete(name) - } - - return false - } - - /** - * Get all circuit breakers - */ - getAll(): Map { - return new Map(this.breakers) - } - - /** - * Get statistics for all circuit breakers - */ - getAllStats(): Record { - const stats: Record = {} - for (const [name, breaker] of this.breakers) { - stats[name] = breaker.getStats() - } - - return stats - } - - /** - * Reset all circuit breakers to CLOSED state - */ - resetAll(): void { - for (const breaker of this.breakers.values()) { - breaker.reset() - } - } - - /** - * Destroy all circuit breakers - */ - destroyAll(): void { - for (const breaker of this.breakers.values()) { - breaker.destroy() - } - - this.breakers.clear() - } -} - -/** - * Global circuit breaker registry instance - */ -export const globalCircuitRegistry = new CircuitBreakerRegistry() - -/** - * Decorator to protect a method with circuit breaker - * - * @example - * ```typescript - * class MyService { - * @circuitBreaker({ failureThreshold: 3 }) - * async fetchData() { - * return await api.getData() - * } - * } - * ``` - */ -export function circuitBreaker(options: Omit & { name?: string } = {}) { - return function ( - _target: unknown, - propertyKey: string, - descriptor: TypedPropertyDescriptor<(...args: unknown[]) => unknown> - ) { - const originalMethod = descriptor.value - if (!originalMethod) return descriptor - - const name = options.name || propertyKey - const breaker = globalCircuitRegistry.get(name, options) - - descriptor.value = async function (...args: unknown[]): Promise { - return breaker.execute(() => originalMethod.apply(this, args)) - } - - return descriptor - } -} - -/** - * Wrap a function with circuit breaker protection - * - * @example - * ```typescript - * const protectedFetch = withCircuitBreaker( - * fetchData, - * { name: 'api-fetch', failureThreshold: 5 } - * ) - * ``` - */ -// eslint-disable-next-line space-before-function-paren -export function withCircuitBreaker unknown>( - fn: T, - options: CircuitBreakerOptions -): T { - const breaker = new CircuitBreaker(options) - - return (async (...args: Parameters): Promise> => { - return breaker.execute(() => fn(...args)) as Promise> - }) as unknown as T -} - -/** - * Get health status of all circuit breakers - */ -export function getCircuitHealth(): { - healthy: boolean - openCircuits: string[] - stats: Record -} { - const stats = globalCircuitRegistry.getAllStats() - const openCircuits: string[] = [] - - for (const [name, stat] of Object.entries(stats)) { - if (stat.isOpen) { - openCircuits.push(name) - } - } - - return { - healthy: openCircuits.length === 0, - openCircuits, - stats - } -} - -/** - * Create a pre-configured circuit breaker for WhatsApp PreKey operations - * - * This circuit breaker is optimized for handling encryption/session errors - * that commonly occur with WhatsApp's Signal protocol implementation. - * - * @example - * ```typescript - * const preKeyBreaker = createPreKeyCircuitBreaker() - * - * async function sendEncryptedMessage(msg) { - * return preKeyBreaker.execute(async () => { - * return await encryptAndSend(msg) - * }) - * } - * ``` - */ -export function createPreKeyCircuitBreaker(customOptions?: Partial): CircuitBreaker { - const preKeyErrorPatterns = ['prekey', 'pre-key', 'session', 'signal', 'encrypt', 'decrypt', 'cipher', 'key'] - - return new CircuitBreaker({ - name: 'prekey-operations', - failureThreshold: 5, - failureWindow: 60000, - resetTimeout: 30000, - successThreshold: 2, - shouldCountError: (error: Error) => { - const message = error.message.toLowerCase() - return preKeyErrorPatterns.some(pattern => message.includes(pattern)) - }, - ...customOptions - }) -} - -/** - * Create a circuit breaker for WebSocket connection operations - */ -export function createConnectionCircuitBreaker(customOptions?: Partial): CircuitBreaker { - const connectionErrorPatterns = [ - 'econnrefused', - 'econnreset', - 'etimedout', - 'enotfound', - 'socket', - 'websocket', - 'connection', - 'network' - ] - - // Normal WS lifecycle status codes that must NEVER trip the query circuit breaker. - // These are transient events that happen on every reconnect and carry no information - // about persistent server-side failures. The reconnection logic in makeSocket handles - // them independently. - const WS_LIFECYCLE_STATUS_CODES = new Set([ - 428, // connectionClosed - 408, // connectionLost / timedOut - 440, // connectionReplaced - ]) - - return new CircuitBreaker({ - name: 'connection-operations', - failureThreshold: 3, - failureWindow: 30000, - resetTimeout: 60000, - successThreshold: 1, - shouldCountError: (error: Error) => { - // CircuitTimeoutError messages embed the circuit name (e.g. "socket-query"), which - // accidentally matches the "socket" pattern below and causes a self-reinforcing loop - // where the breaker's own timeouts keep tripping the breaker. Exclude them explicitly. - if (error instanceof CircuitTimeoutError) return false - - // Exclude normal WS reconnect events (Connection Closed, Connection Lost, Timed Out, etc.). - // These are not server-side failures — they happen on every restart/redeploy and should - // never cause the circuit to open. If we counted them, the 3 concurrent parallel - // post-login queries (sendPassiveIq, uploadPreKeysToServerIfRequired, digestKeyBundle) could all - // fail simultaneously when the WS drops, instantly opening the circuit and blocking - // profile-picture fetches and message delivery for the next 30 s. - const statusCode = (error as { output?: { statusCode?: number } })?.output?.statusCode - if (statusCode !== undefined && WS_LIFECYCLE_STATUS_CODES.has(statusCode)) return false - - const message = error.message.toLowerCase() - const code = (error as NodeJS.ErrnoException).code?.toLowerCase() || '' - return connectionErrorPatterns.some(pattern => message.includes(pattern) || code.includes(pattern)) - }, - ...customOptions - }) -} - -/** - * Create a circuit breaker for message sending operations - */ -export function createMessageCircuitBreaker(customOptions?: Partial): CircuitBreaker { - return new CircuitBreaker({ - name: 'message-operations', - failureThreshold: 5, - failureWindow: 60000, - resetTimeout: 15000, - successThreshold: 2, - timeout: 30000, - ...customOptions - }) -} - -export default CircuitBreaker diff --git a/src/Utils/health-status.ts b/src/Utils/health-status.ts index 7486226d..ee6307b7 100644 --- a/src/Utils/health-status.ts +++ b/src/Utils/health-status.ts @@ -6,11 +6,12 @@ * @module Utils/health-status */ -import { globalCircuitRegistry } from './circuit-breaker.js' import { getVersionCacheStatus } from './version-cache.js' /** - * Circuit breaker health information + * Circuit breaker health information (kept for k8s probe schema stability — + * always reports an empty list now that circuit breakers were removed in + * favour of bounded-retry). */ export interface CircuitBreakerHealth { name: string @@ -105,44 +106,10 @@ export function getHealthStatus(): HealthStatus { }) } - // 2. Check circuit breakers + // 2. Circuit breakers were removed in favour of bounded-retry. + // Keep the field for backward-compat with k8s probe schema, always empty. const circuitBreakers: CircuitBreakerHealth[] = [] - let openCircuits = 0 - - for (const [name, breaker] of globalCircuitRegistry.getAll()) { - const stats = breaker.getStats() - const state = breaker.getState() - - circuitBreakers.push({ - name, - state, - failures: stats.totalFailures, - successes: stats.totalSuccesses, - totalCalls: stats.totalCalls - }) - - if (state === 'open') { - openCircuits++ - } - } - - if (openCircuits > 0) { - checks.push({ - name: 'circuit_breakers', - status: openCircuits > 2 ? 'fail' : 'warn', - message: `${openCircuits} circuit breaker(s) are open` - }) - if (openCircuits > 2) { - overallStatus = 'unhealthy' - } else if (overallStatus === 'healthy') { - overallStatus = 'degraded' - } - } else { - checks.push({ - name: 'circuit_breakers', - status: 'pass' - }) - } + checks.push({ name: 'circuit_breakers', status: 'pass' }) // 3. Check memory usage (warn if > 90%) const memUsage = process.memoryUsage() diff --git a/src/Utils/index.ts b/src/Utils/index.ts index 2c8e793b..74b4bc67 100644 --- a/src/Utils/index.ts +++ b/src/Utils/index.ts @@ -33,8 +33,8 @@ export * from './trace-context' export * from './prometheus-metrics' // Resilience and performance +export * from './bounded-retry' export * from './cache-utils' -export * from './circuit-breaker' export * from './retry-utils' // Telemetry and detection mitigation diff --git a/src/Utils/prometheus-metrics.ts b/src/Utils/prometheus-metrics.ts index 8aafd08a..2b2c14fc 100644 --- a/src/Utils/prometheus-metrics.ts +++ b/src/Utils/prometheus-metrics.ts @@ -1570,26 +1570,6 @@ export const metrics = { new Counter('socket_reconnects_total', 'Total socket reconnection attempts', ['reason']) ), - // ========== Circuit Breaker Metrics ========== - circuitBreakerState: baileysMetrics.register( - new Gauge('circuit_breaker_state', 'Circuit breaker state (0=closed, 1=open, 2=half-open)', ['name']) - ), - circuitBreakerTrips: baileysMetrics.register( - new Counter('circuit_breaker_trips_total', 'Total circuit breaker trips', ['name']) - ), - circuitBreakerRecoveries: baileysMetrics.register( - new Counter('circuit_breaker_recoveries_total', 'Total circuit breaker recoveries', ['name']) - ), - circuitBreakerRejections: baileysMetrics.register( - new Counter('circuit_breaker_rejections_total', 'Total requests rejected by circuit breaker', ['name']) - ), - circuitBreakerSuccesses: baileysMetrics.register( - new Counter('circuit_breaker_successes_total', 'Total successful requests through circuit breaker', ['name']) - ), - circuitBreakerFailures: baileysMetrics.register( - new Counter('circuit_breaker_failures_total', 'Total failed requests through circuit breaker', ['name']) - ), - // ========== Encryption Metrics ========== encryptionOperations: baileysMetrics.register( new Counter('encryption_operations_total', 'Total encryption operations', ['operation']) @@ -2167,9 +2147,6 @@ function initializeMetricsWithLabels(): void { metrics.messageFailures?.inc({ type: 'text', reason: 'max_retries' }, 0) metrics.messageFailures?.inc({ type: 'other', reason: 'max_retries' }, 0) - // Circuit breaker metric - metrics.circuitBreakerTrips?.inc({ name: 'main' }, 0) - console.log('[Prometheus] Initialized metrics with labels') } catch (error) { console.error('[Prometheus] Error initializing metrics with labels:', error) diff --git a/src/Utils/retry-utils.ts b/src/Utils/retry-utils.ts index b656bfac..2b85549e 100644 --- a/src/Utils/retry-utils.ts +++ b/src/Utils/retry-utils.ts @@ -6,15 +6,21 @@ * - Jitter to avoid thundering herd * - Configurable max attempts * - Customizable retry predicates - * - Circuit breaker integration - * - Event hooks - * - Cancellation support + * - Cancellation support (AbortSignal) + * + * Used by `decode-wa-message.ts` for Bad MAC retry recovery on the decrypt + * path. For socket-operation retries (uploadPreKeys etc.) prefer + * `withBoundedRetry` from `./bounded-retry.ts`, which is empirically + * aligned with WhatsApp Android's per-operation backoff. + * + * NOTE: the previous `circuitBreaker` integration option, the `withRetry` + * decorator, the `retryable` wrapper and the `RetryManager` class were + * removed when the socket-level circuit breaker was retired — see PR #393 + * for the rationale. * * @module Utils/retry-utils */ -import { EventEmitter } from 'events' -import type { CircuitBreaker } from './circuit-breaker.js' import { metrics } from './prometheus-metrics.js' /** @@ -59,8 +65,6 @@ export interface RetryOptions { operationName?: string /** Collect metrics */ collectMetrics?: boolean - /** Circuit breaker for integration */ - circuitBreaker?: CircuitBreaker /** Callback before each retry */ onRetry?: (error: Error, attempt: number, delay: number) => void | Promise /** Callback on success */ @@ -283,7 +287,6 @@ export async function retry( timeout: options.timeout, operationName: options.operationName ?? 'operation', collectMetrics: options.collectMetrics ?? true, - circuitBreaker: options.circuitBreaker, onRetry: options.onRetry ?? (() => {}), onSuccess: options.onSuccess ?? (() => {}), onFailure: options.onFailure ?? (() => {}), @@ -313,11 +316,6 @@ export async function retry( throw new RetryAbortedError(attempt) } - // Check circuit breaker - if (config.circuitBreaker?.isOpen()) { - throw new Error(`Circuit breaker "${config.circuitBreaker.getName()}" is open`) - } - try { // Execute operation let result: T @@ -425,138 +423,6 @@ export function createRetrier(defaultOptions: RetryOptions = {}) { } } -/** - * Decorator to add retry to method - */ -export function withRetry(options: RetryOptions = {}) { - return function ( - _target: unknown, - propertyKey: string, - descriptor: TypedPropertyDescriptor<(...args: unknown[]) => unknown> - ) { - const originalMethod = descriptor.value - if (!originalMethod) return descriptor - - descriptor.value = async function (...args: unknown[]): Promise { - return retry(() => originalMethod.apply(this, args), { - ...options, - operationName: options.operationName || propertyKey - }) - } - - return descriptor - } -} - -/** - * Wrapper for function with retry - */ -// eslint-disable-next-line space-before-function-paren -export function retryable unknown>( - fn: T, - options: RetryOptions = {} -): (...args: Parameters) => Promise> { - return async (...args: Parameters): Promise> => { - return retry(() => fn(...args), options) as Promise> - } -} - -/** - * Class to manage retries with state - */ -export class RetryManager extends EventEmitter { - private activeRetries: Map void; context: RetryContext }> = new Map() - private defaultOptions: RetryOptions - - constructor(defaultOptions: RetryOptions = {}) { - super() - this.defaultOptions = defaultOptions - } - - /** - * Execute operation with retry - */ - async execute( - id: string, - operation: (context: RetryContext) => T | Promise, - options?: RetryOptions - ): Promise { - // Cancel previous retry with same ID - this.cancel(id) - - const abortController = new AbortController() - const mergedOptions = { ...this.defaultOptions, ...options, abortSignal: abortController.signal } - - const retryPromise = retry(context => { - this.activeRetries.set(id, { - cancel: () => abortController.abort(), - context - }) - this.emit('attempt', { id, attempt: context.attempt }) - return operation(context) - }, mergedOptions) - - try { - const result = await retryPromise - this.emit('success', { id }) - return result - } catch (error) { - this.emit('failure', { id, error }) - throw error - } finally { - this.activeRetries.delete(id) - } - } - - /** - * Cancel in-progress retry - */ - cancel(id: string): boolean { - const active = this.activeRetries.get(id) - if (active) { - active.cancel() - this.activeRetries.delete(id) - this.emit('cancelled', { id }) - return true - } - - return false - } - - /** - * Cancel all retries - */ - cancelAll(): void { - for (const [id, active] of this.activeRetries) { - active.cancel() - this.emit('cancelled', { id }) - } - - this.activeRetries.clear() - } - - /** - * Check if there is an active retry - */ - isActive(id: string): boolean { - return this.activeRetries.has(id) - } - - /** - * Return active retry context - */ - getContext(id: string): RetryContext | undefined { - return this.activeRetries.get(id)?.context - } - - /** - * Return active retry IDs - */ - getActiveIds(): string[] { - return Array.from(this.activeRetries.keys()) - } -} - /** * Common predicates for shouldRetry */ @@ -660,9 +526,8 @@ export const retryConfigs = { * Uses fixed delay array: 1s, 2s, 5s, 10s, 20s (with ±15% jitter) * * NOTE: Values are hardcoded instead of referencing RETRY_BACKOFF_DELAYS/RETRY_JITTER_FACTOR - * to prevent "Cannot access before initialization" errors in ESM environments. - * This occurs when modules are loaded in specific orders due to indirect circular imports - * (e.g., via prometheus-metrics.ts -> circuit-breaker.ts chain). + * to prevent "Cannot access before initialization" errors in ESM environments + * caused by indirect circular imports through prometheus-metrics.ts. * Keep these values in sync with the constants above (lines 25, 31). */ rsocket: { diff --git a/src/Utils/structured-logger.ts b/src/Utils/structured-logger.ts index 57053172..371ddc32 100644 --- a/src/Utils/structured-logger.ts +++ b/src/Utils/structured-logger.ts @@ -6,7 +6,9 @@ * - Configurable log levels (trace, debug, info, warn, error, fatal) * - JSON formatting for log analysis * - Hierarchical context with child loggers - * - External system integration via hooks with circuit breaker + * - External system integration via fire-and-forget hooks (failures + * counted in hookFailures metric; no per-hook circuit breaker — chronic + * breakage is visible to operators via the metric) * - Logging metrics with Prometheus integration * - Sensitive data sanitization * - Log buffering for batch writes @@ -73,10 +75,6 @@ export interface StructuredLoggerConfig { maxLogsPerSecond?: number /** Enable async logging queue (default: false) */ enableAsyncQueue?: boolean - /** Circuit breaker failure threshold for external hooks (default: 5) */ - circuitBreakerThreshold?: number - /** Circuit breaker reset timeout in ms (default: 30000) */ - circuitBreakerResetMs?: number /** Enable Prometheus metrics integration (default: false) */ enableMetrics?: boolean } @@ -155,8 +153,6 @@ export interface LoggerMetrics { bufferFlushes: number /** External hook failures */ hookFailures: number - /** Circuit breaker trips */ - circuitBreakerTrips: number /** Average log processing time in ms */ avgProcessingTimeMs: number } @@ -169,8 +165,6 @@ export interface LoggerStatistics extends LoggerMetrics { bufferSize: number /** Rate limiter tokens available */ rateLimiterTokens: number - /** Circuit breaker state */ - circuitBreakerState: 'closed' | 'open' | 'half-open' /** Queue size (if async enabled) */ queueSize: number /** Created timestamp */ @@ -242,69 +236,6 @@ class RateLimiter { } } -// ============================================================================ -// CIRCUIT BREAKER -// ============================================================================ - -/** - * Circuit breaker for external hook protection - */ -class CircuitBreaker { - private failures = 0 - private lastFailure = 0 - private state: 'closed' | 'open' | 'half-open' = 'closed' - - constructor( - private readonly threshold: number, - private readonly resetTimeoutMs: number - ) {} - - async execute(operation: () => Promise): Promise { - if (this.state === 'open') { - if (Date.now() - this.lastFailure > this.resetTimeoutMs) { - this.state = 'half-open' - } else { - return null // Circuit is open, skip - } - } - - try { - const result = await operation() - this.onSuccess() - return result - } catch (error) { - this.onFailure() - throw error - } - } - - private onSuccess(): void { - this.failures = 0 - this.state = 'closed' - } - - private onFailure(): void { - this.failures++ - this.lastFailure = Date.now() - if (this.failures >= this.threshold) { - this.state = 'open' - } - } - - getState(): 'closed' | 'open' | 'half-open' { - // Check if should transition from open to half-open - if (this.state === 'open' && Date.now() - this.lastFailure > this.resetTimeoutMs) { - this.state = 'half-open' - } - - return this.state - } - - getFailures(): number { - return this.failures - } -} - // ============================================================================ // ASYNC LOG QUEUE // ============================================================================ @@ -403,8 +334,6 @@ export class StructuredLogger implements ILogger { private rateLimiter: RateLimiter | null = null // Circuit breaker for external hook - private circuitBreaker: CircuitBreaker | null = null - // Async queue private asyncQueue: AsyncLogQueue | null = null @@ -435,8 +364,6 @@ export class StructuredLogger implements ILogger { enableRateLimiting: config.enableRateLimiting ?? envConfig.enableRateLimiting ?? false, maxLogsPerSecond: config.maxLogsPerSecond ?? envConfig.maxLogsPerSecond ?? 1000, enableAsyncQueue: config.enableAsyncQueue ?? envConfig.enableAsyncQueue ?? false, - circuitBreakerThreshold: config.circuitBreakerThreshold ?? 5, - circuitBreakerResetMs: config.circuitBreakerResetMs ?? 30000, enableMetrics: config.enableMetrics ?? envConfig.enableMetrics ?? false } @@ -455,7 +382,6 @@ export class StructuredLogger implements ILogger { droppedLogs: 0, bufferFlushes: 0, hookFailures: 0, - circuitBreakerTrips: 0, avgProcessingTimeMs: 0 } @@ -464,11 +390,6 @@ export class StructuredLogger implements ILogger { this.rateLimiter = new RateLimiter(this.config.maxLogsPerSecond) } - // Initialize circuit breaker for external hook - if (this.config.externalHook) { - this.circuitBreaker = new CircuitBreaker(this.config.circuitBreakerThreshold, this.config.circuitBreakerResetMs) - } - // Initialize async queue if (this.config.enableAsyncQueue) { this.asyncQueue = new AsyncLogQueue() @@ -569,19 +490,27 @@ export class StructuredLogger implements ILogger { this.output(entry) } - // External hook with circuit breaker + // External hook — fire-and-forget. Failures are counted in + // hookFailures metric so chronic breakage is visible to operators + // who can fix their hook. We do not gate further calls — that is + // the consumer's responsibility (consistent with the rest of the + // codebase after the circuit-breaker removal). const externalHook = this.config.externalHook - if (externalHook && this.circuitBreaker) { - this.circuitBreaker - .execute(async () => { - await Promise.resolve(externalHook(entry)) - }) - .catch(() => { - this.metrics.hookFailures++ - if (this.circuitBreaker?.getState() === 'open') { - this.metrics.circuitBreakerTrips++ - } - }) + if (externalHook) { + try { + const hookResult = externalHook(entry) as unknown + // Use Promise.resolve().catch() so any thenable (including + // Promise-likes that only implement .then) is handled + // safely — calling .catch directly would throw on such + // objects (Copilot review on PR #393). + if (hookResult && typeof (hookResult as PromiseLike).then === 'function') { + void Promise.resolve(hookResult as PromiseLike).catch(() => { + this.metrics.hookFailures++ + }) + } + } catch { + this.metrics.hookFailures++ + } } // Track processing time @@ -834,7 +763,6 @@ export class StructuredLogger implements ILogger { ...this.metrics, bufferSize: this.buffer.length, rateLimiterTokens: this.rateLimiter?.getTokens() ?? 0, - circuitBreakerState: this.circuitBreaker?.getState() ?? 'closed', queueSize: this.asyncQueue?.getSize() ?? 0, createdAt: this.createdAt, uptimeMs: Date.now() - this.createdAt @@ -860,7 +788,6 @@ export class StructuredLogger implements ILogger { droppedLogs: 0, bufferFlushes: 0, hookFailures: 0, - circuitBreakerTrips: 0, avgProcessingTimeMs: 0 } this.totalProcessingTime = 0 @@ -893,7 +820,6 @@ export class StructuredLogger implements ILogger { // Clear references this.rateLimiter = null - this.circuitBreaker = null this.metricsModule = null } } diff --git a/src/Utils/unified-session.ts b/src/Utils/unified-session.ts index cff005f7..9b9ba1d4 100644 --- a/src/Utils/unified-session.ts +++ b/src/Utils/unified-session.ts @@ -15,7 +15,6 @@ */ import type { BinaryNode } from '../WABinary/types.js' -import { CircuitBreaker, CircuitOpenError, createConnectionCircuitBreaker } from './circuit-breaker.js' import type { ILogger } from './logger.js' import { metrics } from './prometheus-metrics.js' @@ -58,8 +57,6 @@ export interface UnifiedSessionOptions { enabled?: boolean /** Logger instance for debugging */ logger?: ILogger - /** Enable circuit breaker protection for send operations */ - enableCircuitBreaker?: boolean /** Function to send binary nodes to WhatsApp */ sendNode?: (node: BinaryNode) => Promise } @@ -93,7 +90,9 @@ export type UnifiedSessionTrigger = 'login' | 'pairing' | 'presence' | 'manual' * Manages the unified_session telemetry feature with: * - Server time synchronization * - Rate limiting (prevents spam) - * - Circuit breaker protection + * - Best-effort send semantics (failures are swallowed and counted — + * telemetry is non-critical so a single attempt with the underlying + * sendNode timeout is enough; no retry, no circuit breaker) * - Prometheus metrics integration * - Structured logging * @@ -121,7 +120,6 @@ export class UnifiedSessionManager { } private readonly options: Required - private circuitBreaker: CircuitBreaker | null = null /** Minimum interval between unified_session sends (1 minute) */ private static readonly MIN_SEND_INTERVAL_MS = TimeMs.Minute @@ -130,7 +128,6 @@ export class UnifiedSessionManager { this.options = { enabled: options.enabled ?? true, logger: options.logger ?? (console as unknown as ILogger), - enableCircuitBreaker: options.enableCircuitBreaker ?? true, sendNode: options.sendNode ?? (async () => { @@ -138,25 +135,6 @@ export class UnifiedSessionManager { }) } - // Initialize circuit breaker if enabled - if (this.options.enableCircuitBreaker) { - this.circuitBreaker = createConnectionCircuitBreaker({ - name: 'unified-session', - failureThreshold: 3, - failureWindow: 60000, - resetTimeout: 30000, - successThreshold: 1, - timeout: 10000, - onStateChange: (from, to) => { - this.options.logger.debug?.({ from, to }, 'Unified session circuit breaker state changed') - }, - onOpen: () => { - this.options.logger.warn?.('Unified session circuit breaker OPENED') - metrics.circuitBreakerTrips?.inc({ name: 'unified-session' }) - } - }) - } - this.state.isInitialized = true this.options.logger.debug?.('UnifiedSessionManager initialized') } @@ -292,20 +270,12 @@ export class UnifiedSessionManager { } try { - // Execute with circuit breaker if available - if (this.circuitBreaker) { - await this.circuitBreaker.execute(sendOperation) - } else { - await sendOperation() - } + // Telemetry is non-critical: a single send attempt with the + // underlying sendNode timeout is enough. No retry/circuit needed. + await sendOperation() } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error) - - if (error instanceof CircuitOpenError) { - this.options.logger.warn?.({ trigger, circuitState: error.state }, 'Unified session blocked by circuit breaker') - } else { - this.options.logger.warn?.({ trigger, error: errorMessage }, 'Failed to send unified session telemetry') - } + this.options.logger.warn?.({ trigger, error: errorMessage }, 'Failed to send unified session telemetry') // Record failure metric metrics.errors?.inc({ category: 'unified_session', code: 'send_failed' }) @@ -331,7 +301,6 @@ export class UnifiedSessionManager { sendCount: 0, isInitialized: true } - this.circuitBreaker?.reset() this.options.logger.debug?.('UnifiedSessionManager reset') } @@ -339,7 +308,6 @@ export class UnifiedSessionManager { * Destroy the manager and clean up resources. */ destroy(): void { - this.circuitBreaker?.destroy() this.state.isInitialized = false this.options.logger.debug?.('UnifiedSessionManager destroyed') } diff --git a/src/__tests__/Utils/bounded-retry.test.ts b/src/__tests__/Utils/bounded-retry.test.ts new file mode 100644 index 00000000..c5d0f661 --- /dev/null +++ b/src/__tests__/Utils/bounded-retry.test.ts @@ -0,0 +1,391 @@ +import { jest } from '@jest/globals' +import { + withBoundedRetry, + BoundedRetryGiveUpError, + BoundedRetryAbortedError, + WHATSAPP_BACKOFF_DELAYS, + DEFAULT_TTL_MS +} from '../../Utils/bounded-retry' + +// All tests use very short real delays (1-50ms) so they run fast. +// This sidesteps the fragility of jest fake timers + async chains. + +describe('bounded-retry — WhatsApp-aligned per-operation retry', () => { + test('default delay sequence matches WhatsApp Android empirical capture', () => { + // Frida trace: 3000 -> 10000 -> 60000 -> ~64000 -> 120000 ms + expect(WHATSAPP_BACKOFF_DELAYS).toEqual([3000, 10000, 60000, 60000, 120000]) + }) + + test('default TTL is 10 minutes (matches empirical stabilisation + safety buffer)', () => { + expect(DEFAULT_TTL_MS).toBe(600_000) + }) + + test('first-attempt success returns immediately, no retries', async () => { + const op = jest.fn<() => Promise>().mockResolvedValue('ok') + const result = await withBoundedRetry(op, { name: 'fast-success' }) + expect(result).toBe('ok') + expect(op).toHaveBeenCalledTimes(1) + }) + + test('retries with the configured delay sequence after failures', async () => { + const op = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('try1')) + .mockRejectedValueOnce(new Error('try2')) + .mockResolvedValueOnce(42) + + const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>() + const result = await withBoundedRetry(op, { + name: 'three-tries', + delays: [10, 20, 40], + jitter: 0, + onRetry + }) + + expect(result).toBe(42) + expect(op).toHaveBeenCalledTimes(3) + expect(onRetry).toHaveBeenCalledTimes(2) + expect(onRetry.mock.calls[0]![2]).toBe(10) + expect(onRetry.mock.calls[1]![2]).toBe(20) + }) + + test('TTL exhaustion throws BoundedRetryGiveUpError (no infinite retries)', async () => { + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('always-fails')) + + await expect( + withBoundedRetry(op, { + name: 'ttl-test', + delays: [10], + jitter: 0, + ttlMs: 50 + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + // Should have tried multiple times before giving up + expect(op.mock.calls.length).toBeGreaterThanOrEqual(2) + }) + + test('per-attempt timeout fires for slow operations', async () => { + const slowOp = jest.fn<() => Promise>(() => new Promise(() => {})) // hangs + + await expect( + withBoundedRetry(slowOp, { + name: 'slow', + delays: [5], + jitter: 0, + ttlMs: 200, + perAttemptTimeoutMs: 30 + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + // Multiple attempts should have been tried (each timed out at 30ms) + expect(slowOp.mock.calls.length).toBeGreaterThan(1) + }) + + test('shouldRetry predicate can short-circuit retries', async () => { + const op = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('transient')) + .mockRejectedValueOnce(new Error('FATAL: do not retry')) + + await expect( + withBoundedRetry(op, { + name: 'predicate-test', + delays: [1], + jitter: 0, + shouldRetry: err => !err.message.startsWith('FATAL') + }) + ).rejects.toThrow('FATAL: do not retry') + + expect(op).toHaveBeenCalledTimes(2) + }) + + test('AbortSignal cancels in-flight retry sleep', async () => { + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('fail')) + const ctrl = new AbortController() + + const promise = withBoundedRetry(op, { + name: 'abort-test', + delays: [10_000], // long sleep so abort can happen during it + jitter: 0, + signal: ctrl.signal + }) + + // First attempt fires immediately; sleep starts + await new Promise(resolve => setTimeout(resolve, 10)) + ctrl.abort() + + await expect(promise).rejects.toBeInstanceOf(BoundedRetryAbortedError) + }) + + test('two concurrent retry chains do NOT interfere with each other (per-op isolation)', async () => { + // Key advantage over circuit breaker: failures in one chain do not + // block another chain. With circuit breaker, 5 failures from chain A + // would block chain B too. With bounded-retry, B is independent. + + const flakyOp = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('a1')) + .mockResolvedValueOnce('a-ok') + + const fastOp = jest.fn<() => Promise>().mockResolvedValue('b-ok') + + // Start a (will retry once after 30ms) and b (immediate) + const aPromise = withBoundedRetry(flakyOp, { name: 'a', delays: [30], jitter: 0 }) + const bPromise = withBoundedRetry(fastOp, { name: 'b', delays: [30], jitter: 0 }) + + // b should resolve right away + const bResult = await bPromise + expect(bResult).toBe('b-ok') + expect(fastOp).toHaveBeenCalledTimes(1) + + // a still running — wait for it + const aResult = await aPromise + expect(aResult).toBe('a-ok') + expect(flakyOp).toHaveBeenCalledTimes(2) + }) + + test('delay sequence caps at last value (no unbounded growth)', async () => { + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('fail')) + const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>() + + await expect( + withBoundedRetry(op, { + name: 'cap-test', + delays: [5, 10, 15], // 3 distinct steps; 4th+ retry should reuse 15 + jitter: 0, + ttlMs: 300, + onRetry + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + // Find delays >= 4th retry — all should be capped at 15 + const lateRetries = onRetry.mock.calls.filter(c => c[1] >= 4) + expect(lateRetries.length).toBeGreaterThan(0) + for (const call of lateRetries) { + expect(call[2]).toBeLessThanOrEqual(15) + } + }) + + test('jitter randomises delay within +/- factor', async () => { + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('fail')) + const observed: number[] = [] + + // Single run, capture multiple onRetry calls (each with different jitter) + await withBoundedRetry(op, { + name: 'jitter-test', + delays: [50], // base delay + jitter: 0.3, // +/- 30% -> [35, 65] + ttlMs: 1000, // ttl high enough for ~10 retries + perAttemptTimeoutMs: 5, + onRetry: (_, __, d) => observed.push(d) + }).catch(() => {}) + + expect(observed.length).toBeGreaterThanOrEqual(3) + const unique = new Set(observed) + expect(unique.size).toBeGreaterThan(1) + // Every observed delay must be within jitter window — but the LAST one + // may be capped by remaining ttl budget, so check all but the last + const inWindow = observed.slice(0, -1) + for (const d of inWindow) { + expect(d).toBeGreaterThanOrEqual(35) + expect(d).toBeLessThanOrEqual(65) + } + }) + + test('delay does not exceed remaining TTL budget', async () => { + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('fail')) + const onRetry = jest.fn<(err: Error, attempt: number, delayMs: number) => void>() + + await withBoundedRetry(op, { + name: 'budget-cap', + delays: [200], // would exceed ttl + jitter: 0, + ttlMs: 50, // very short + onRetry + }).catch(() => {}) + + // All scheduled delays should be capped to remaining budget + for (const call of onRetry.mock.calls) { + expect(call[2]).toBeLessThanOrEqual(50) + } + }) + + // ─── Tests for fixes from PR #393 review ────────────────────────────── + + test('TTL is enforced BEFORE the next attempt (not just after)', async () => { + // Regression: previously TTL was only checked after a failure, so a + // new attempt could start with 0ms remaining and run for the full + // per-attempt timeout, overshooting wall-clock budget by that much. + const slowOp = jest.fn<() => Promise>(() => new Promise(resolve => setTimeout(() => resolve('late'), 200))) + + const start = Date.now() + await expect( + withBoundedRetry(slowOp, { + name: 'ttl-strict', + delays: [10], + jitter: 0, + ttlMs: 50, + perAttemptTimeoutMs: 1000 // huge, must be capped by TTL + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + const elapsed = Date.now() - start + // With strict TTL enforcement, total runtime should be very close to ttlMs. + // Allow up to 100ms slack for Node timer + microtask scheduling. + expect(elapsed).toBeLessThan(150) + }) + + test('per-attempt timeout is capped by remaining TTL budget', async () => { + // If user passes perAttemptTimeoutMs > ttlMs, the cap should apply. + const slowOp = jest.fn<() => Promise>(() => new Promise(() => {})) // hangs forever + + const start = Date.now() + await expect( + withBoundedRetry(slowOp, { + name: 'attempt-cap', + delays: [1], + jitter: 0, + ttlMs: 30, + perAttemptTimeoutMs: 5000 // way bigger than ttl + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + const elapsed = Date.now() - start + expect(elapsed).toBeLessThan(150) // not 5000ms! + }) + + test('operation receives an AbortSignal that is aborted on per-attempt timeout', async () => { + // New API: operation can opt into cancellation via the signal arg. + const aborts: boolean[] = [] + const opThatRespectsSignal = jest.fn( + (signal?: AbortSignal) => + new Promise((resolve, reject) => { + const timer = setTimeout(() => resolve('late'), 100) + signal?.addEventListener('abort', () => { + clearTimeout(timer) + aborts.push(true) + reject(new Error('aborted by signal')) + }) + }) + ) + + await withBoundedRetry(opThatRespectsSignal, { + name: 'signal-aware', + delays: [1], + jitter: 0, + ttlMs: 80, + perAttemptTimeoutMs: 20 // smaller than 100ms op duration + }).catch(() => {}) + + expect(aborts.length).toBeGreaterThan(0) // at least one attempt was aborted + }) + + test('sleep listener is removed when timeout completes normally (no leak)', async () => { + // Regression: sleep used signal.addEventListener with { once: true } but + // never removed the listener on timer-resolve. With many retries on a + // shared signal, listeners would accumulate. + // + // Make the failure mode observable by spying on add/removeEventListener + // directly so the assertion fails if cleanup is removed. + const ctrl = new AbortController() + const realAdd = ctrl.signal.addEventListener.bind(ctrl.signal) + const realRemove = ctrl.signal.removeEventListener.bind(ctrl.signal) + let addCount = 0 + let removeCount = 0 + ctrl.signal.addEventListener = ((type: string, listener: never, opts?: never) => { + if (type === 'abort') addCount++ + return realAdd(type, listener, opts) + }) as typeof ctrl.signal.addEventListener + ctrl.signal.removeEventListener = ((type: string, listener: never, opts?: never) => { + if (type === 'abort') removeCount++ + return realRemove(type, listener, opts) + }) as typeof ctrl.signal.removeEventListener + + const op = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('1')) + .mockRejectedValueOnce(new Error('2')) + .mockRejectedValueOnce(new Error('3')) + .mockResolvedValueOnce('ok') + + const result = await withBoundedRetry(op, { + name: 'no-leak', + delays: [5], + jitter: 0, + ttlMs: 500, + signal: ctrl.signal + }) + + expect(result).toBe('ok') + expect(op).toHaveBeenCalledTimes(4) + // If sleep() leaks listeners, addCount > removeCount. Each retry adds + // listeners (1 sleep + 1 outer-abort forwarding) and must remove the + // same number on settlement. + expect(addCount).toBeGreaterThan(0) + expect(addCount).toBe(removeCount) + }) + + test('logger receives structured logs for retry, give-up, and recovery', async () => { + const debugCalls: unknown[][] = [] + const infoCalls: unknown[][] = [] + const warnCalls: unknown[][] = [] + const fakeLogger = { + debug: (...args: unknown[]) => debugCalls.push(args), + info: (...args: unknown[]) => infoCalls.push(args), + warn: (...args: unknown[]) => warnCalls.push(args), + error: () => {}, + fatal: () => {}, + trace: () => {}, + child: () => fakeLogger, + level: 'debug' + } + + const op = jest + .fn<() => Promise>() + .mockRejectedValueOnce(new Error('boom')) + .mockResolvedValueOnce('ok') + + const result = await withBoundedRetry(op, { + name: 'logged-op', + delays: [5], + jitter: 0, + ttlMs: 200, + logger: fakeLogger as never + }) + + expect(result).toBe('ok') + // Should have logged the scheduling at debug level + expect(debugCalls.length).toBeGreaterThanOrEqual(1) + // Should have logged the recovery at info level + expect(infoCalls.length).toBeGreaterThanOrEqual(1) + }) + + test('logger reports give-up via warn when TTL exceeded', async () => { + const warnCalls: unknown[][] = [] + const fakeLogger = { + debug: () => {}, + info: () => {}, + warn: (...args: unknown[]) => warnCalls.push(args), + error: () => {}, + fatal: () => {}, + trace: () => {}, + child: () => fakeLogger, + level: 'debug' + } + + const op = jest.fn<() => Promise>().mockRejectedValue(new Error('always-fails')) + + await expect( + withBoundedRetry(op, { + name: 'logged-fail', + delays: [5], + jitter: 0, + ttlMs: 30, + logger: fakeLogger as never + }) + ).rejects.toBeInstanceOf(BoundedRetryGiveUpError) + + expect(warnCalls.length).toBeGreaterThanOrEqual(1) + }) +}) diff --git a/src/__tests__/Utils/circuit-breaker.test.ts b/src/__tests__/Utils/circuit-breaker.test.ts deleted file mode 100644 index cdb792cf..00000000 --- a/src/__tests__/Utils/circuit-breaker.test.ts +++ /dev/null @@ -1,371 +0,0 @@ -/** - * Testes unitários para circuit-breaker.ts - */ - -import { afterEach, beforeEach, describe, expect, it } from '@jest/globals' -import { - CircuitBreaker, - CircuitBreakerRegistry, - CircuitOpenError, - type CircuitState, - CircuitTimeoutError, - createCircuitBreaker, - getCircuitHealth, - globalCircuitRegistry, - withCircuitBreaker -} from '../../Utils/circuit-breaker.js' - -describe('CircuitBreaker', () => { - let breaker: CircuitBreaker - - beforeEach(() => { - breaker = createCircuitBreaker({ - name: 'test', - failureThreshold: 3, - successThreshold: 2, - resetTimeout: 100, - timeout: 1000, - volumeThreshold: 3, - collectMetrics: false - }) - }) - - afterEach(() => { - breaker.destroy() - }) - - describe('initial state', () => { - it('should start in closed state', () => { - expect(breaker.getState()).toBe('closed') - expect(breaker.isClosed()).toBe(true) - expect(breaker.isOpen()).toBe(false) - expect(breaker.isHalfOpen()).toBe(false) - }) - }) - - describe('successful operations', () => { - it('should execute successful operations', async () => { - const result = await breaker.execute(() => 'success') - expect(result).toBe('success') - }) - - it('should execute async operations', async () => { - const result = await breaker.execute(async () => { - await new Promise(resolve => setTimeout(resolve, 10)) - return 'async success' - }) - expect(result).toBe('async success') - }) - - it('should track successful calls in stats', async () => { - await breaker.execute(() => 'success') - await breaker.execute(() => 'success') - - const stats = breaker.getStats() - expect(stats.totalCalls).toBe(2) - expect(stats.totalSuccesses).toBe(2) - expect(stats.totalFailures).toBe(0) - }) - }) - - describe('failure handling', () => { - it('should open after reaching failure threshold', async () => { - const failingOp = () => { - throw new Error('Failure') - } - - // Reach failure threshold - for (let i = 0; i < 3; i++) { - await expect(breaker.execute(failingOp)).rejects.toThrow('Failure') - } - - expect(breaker.isOpen()).toBe(true) - }) - - it('should not open before reaching threshold', async () => { - const failingOp = () => { - throw new Error('Failure') - } - - // Below threshold - for (let i = 0; i < 2; i++) { - await expect(breaker.execute(failingOp)).rejects.toThrow('Failure') - } - - expect(breaker.isClosed()).toBe(true) - }) - - it('should reset failure count on success', async () => { - const failingOp = () => { - throw new Error('Failure') - } - - await expect(breaker.execute(failingOp)).rejects.toThrow() - await expect(breaker.execute(failingOp)).rejects.toThrow() - - // Success resets failures - await breaker.execute(() => 'success') - - // Only 1 more failure needed (already have 2 in window, need 3 total) - await expect(breaker.execute(failingOp)).rejects.toThrow() - - // Should open now (3 failures in window) - expect(breaker.isOpen()).toBe(true) - }) - }) - - describe('open state', () => { - beforeEach(async () => { - // Force open - breaker.trip() - }) - - it('should reject operations when open', async () => { - await expect(breaker.execute(() => 'success')).rejects.toThrow(CircuitOpenError) - }) - - it('should transition to half-open after reset timeout', async () => { - await new Promise(resolve => setTimeout(resolve, 150)) - expect(breaker.isHalfOpen()).toBe(true) - }) - }) - - describe('half-open state', () => { - beforeEach(async () => { - breaker.trip() - await new Promise(resolve => setTimeout(resolve, 150)) - }) - - it('should close after success threshold', async () => { - expect(breaker.isHalfOpen()).toBe(true) - - // Success threshold is 2 - await breaker.execute(() => 'success') - await breaker.execute(() => 'success') - - expect(breaker.isClosed()).toBe(true) - }) - - it('should reopen on failure', async () => { - expect(breaker.isHalfOpen()).toBe(true) - - await expect( - breaker.execute(() => { - throw new Error('Failure') - }) - ).rejects.toThrow() - - expect(breaker.isOpen()).toBe(true) - }) - }) - - describe('manual controls', () => { - it('should force open with trip()', () => { - breaker.trip() - expect(breaker.isOpen()).toBe(true) - }) - - it('should force close with reset()', async () => { - breaker.trip() - breaker.reset() - expect(breaker.isClosed()).toBe(true) - }) - }) - - describe('timeout', () => { - it('should timeout slow operations', async () => { - const slowBreaker = createCircuitBreaker({ - name: 'slow', - timeout: 50, - collectMetrics: false - }) - - await expect(slowBreaker.execute(() => new Promise(resolve => setTimeout(resolve, 200)))).rejects.toThrow( - CircuitTimeoutError - ) - - slowBreaker.destroy() - }) - }) - - describe('events', () => { - it('should emit state-change event', async () => { - const stateChanges: Array<{ from: CircuitState; to: CircuitState }> = [] - - breaker.on('state-change', change => { - stateChanges.push(change) - }) - - breaker.trip() - breaker.reset() - - expect(stateChanges).toHaveLength(2) - expect(stateChanges[0]).toEqual({ from: 'closed', to: 'open' }) - expect(stateChanges[1]).toEqual({ from: 'open', to: 'closed' }) - }) - - it('should emit success and failure events', async () => { - let successCount = 0 - let failureCount = 0 - - breaker.on('success', () => successCount++) - breaker.on('failure', () => failureCount++) - - await breaker.execute(() => 'success') - await expect( - breaker.execute(() => { - throw new Error() - }) - ).rejects.toThrow() - - expect(successCount).toBe(1) - expect(failureCount).toBe(1) - }) - }) - - describe('isFailure predicate', () => { - it('should use custom isFailure predicate', async () => { - const customBreaker = createCircuitBreaker({ - name: 'custom', - failureThreshold: 1, - volumeThreshold: 1, - shouldCountError: (error: any) => error.message !== 'Ignored', - collectMetrics: false - }) - - // This error should be ignored - await expect( - customBreaker.execute(() => { - throw new Error('Ignored') - }) - ).rejects.toThrow() - - expect(customBreaker.isClosed()).toBe(true) - - // This should trip the breaker - await expect( - customBreaker.execute(() => { - throw new Error('Real failure') - }) - ).rejects.toThrow() - - expect(customBreaker.isOpen()).toBe(true) - - customBreaker.destroy() - }) - }) - - describe('sync operations', () => { - it('should execute sync operations', () => { - const result = breaker.executeSync(() => 'sync result') - expect(result).toBe('sync result') - }) - - it('should handle sync failures', () => { - expect(() => - breaker.executeSync(() => { - throw new Error('Sync failure') - }) - ).toThrow('Sync failure') - }) - }) -}) - -describe('CircuitBreakerRegistry', () => { - let registry: CircuitBreakerRegistry - - beforeEach(() => { - registry = new CircuitBreakerRegistry() - }) - - afterEach(() => { - registry.destroyAll() - }) - - it('should create and retrieve circuit breakers', () => { - const breaker1 = registry.get('test1') - const breaker2 = registry.get('test1') - - expect(breaker1).toBe(breaker2) - }) - - it('should check if breaker exists', () => { - registry.get('exists') - - expect(registry.has('exists')).toBe(true) - expect(registry.has('notexists')).toBe(false) - }) - - it('should remove breaker', () => { - registry.get('toRemove') - expect(registry.remove('toRemove')).toBe(true) - expect(registry.has('toRemove')).toBe(false) - }) - - it('should get all stats', () => { - registry.get('breaker1') - registry.get('breaker2') - - const stats = registry.getAllStats() - - expect(stats).toHaveProperty('breaker1') - expect(stats).toHaveProperty('breaker2') - }) - - it('should reset all breakers', async () => { - const breaker = registry.get('test') - breaker.trip() - - registry.resetAll() - - expect(breaker.isClosed()).toBe(true) - }) -}) - -describe('withCircuitBreaker', () => { - it('should wrap function with circuit breaker', async () => { - let callCount = 0 - - const protectedFn = withCircuitBreaker( - async () => { - callCount++ - return 'result' - }, - { - name: 'wrapped-fn', - collectMetrics: false - } - ) - - const result = await protectedFn() - - expect(result).toBe('result') - expect(callCount).toBe(1) - }) -}) - -describe('getCircuitHealth', () => { - beforeEach(() => { - globalCircuitRegistry.destroyAll() - }) - - it('should report healthy when all circuits closed', () => { - globalCircuitRegistry.get('healthy1') - globalCircuitRegistry.get('healthy2') - - const health = getCircuitHealth() - - expect(health.healthy).toBe(true) - expect(health.openCircuits).toHaveLength(0) - }) - - it('should report unhealthy when circuit is open', () => { - const breaker = globalCircuitRegistry.get('unhealthy') - breaker.trip() - - const health = getCircuitHealth() - - expect(health.healthy).toBe(false) - expect(health.openCircuits).toContain('unhealthy') - }) -}) diff --git a/src/__tests__/Utils/retry-utils.test.ts b/src/__tests__/Utils/retry-utils.test.ts index 14ff473d..3fad8824 100644 --- a/src/__tests__/Utils/retry-utils.test.ts +++ b/src/__tests__/Utils/retry-utils.test.ts @@ -2,7 +2,7 @@ * Testes unitários para retry-utils.ts */ -import { beforeEach, describe, expect, it, jest } from '@jest/globals' +import { describe, expect, it, jest } from '@jest/globals' import { calculateDelay, createRetrier, @@ -11,12 +11,10 @@ import { retry, RETRY_BACKOFF_DELAYS, RETRY_JITTER_FACTOR, - retryable, RetryAbortedError, retryConfigs, type RetryContext, RetryExhaustedError, - RetryManager, retryPredicates, retryWithResult } from '../../Utils/retry-utils.js' @@ -315,103 +313,6 @@ describe('createRetrier', () => { }) }) -describe('retryable', () => { - it('should wrap function with retry', async () => { - let attempts = 0 - - const fn = retryable( - () => { - attempts++ - if (attempts < 2) throw new Error('Failing') - return 'wrapped result' - }, - { - maxAttempts: 5, - baseDelay: 10, - collectMetrics: false - } - ) - - const result = await fn() - - expect(result).toBe('wrapped result') - expect(attempts).toBe(2) - }) -}) - -describe('RetryManager', () => { - let manager: RetryManager - - beforeEach(() => { - manager = new RetryManager({ baseDelay: 200, collectMetrics: false }) - }) - - it('should execute operation with id', async () => { - const result = await manager.execute('op1', () => 'result') - expect(result).toBe('result') - }) - - it('should cancel active retry', async () => { - let attempt = 0 - - const promise = manager.execute('cancelable', async () => { - attempt++ - if (attempt === 1) { - throw new Error('Temporary failure') - } - - await new Promise(resolve => setTimeout(resolve, 1000)) - return 'result' - }) - - // Cancel during retry delay - setTimeout(() => manager.cancel('cancelable'), 100) - - await expect(promise).rejects.toThrow() - }) - - it('should check if operation is active', async () => { - let resolve: () => void - const promise = manager.execute( - 'active', - () => - new Promise(r => { - resolve = () => r('done') - }) - ) - - expect(manager.isActive('active')).toBe(true) - - resolve!() - await promise - - expect(manager.isActive('active')).toBe(false) - }) - - it('should cancel all operations', async () => { - const promises = [ - manager.execute('op1', () => new Promise((_, reject) => setTimeout(() => reject(new Error()), 1000))), - manager.execute('op2', () => new Promise((_, reject) => setTimeout(() => reject(new Error()), 1000))) - ] - - setTimeout(() => manager.cancelAll(), 20) - - await expect(Promise.all(promises)).rejects.toThrow() - }) - - it('should emit events', async () => { - const events: string[] = [] - - manager.on('attempt', () => events.push('attempt')) - manager.on('success', () => events.push('success')) - - await manager.execute('test', () => 'result') - - expect(events).toContain('attempt') - expect(events).toContain('success') - }) -}) - describe('retryPredicates', () => { describe('always', () => { it('should always return true', () => { diff --git a/src/__tests__/Utils/unified-session.test.ts b/src/__tests__/Utils/unified-session.test.ts index 3f3a17b9..4c3ee642 100644 --- a/src/__tests__/Utils/unified-session.test.ts +++ b/src/__tests__/Utils/unified-session.test.ts @@ -29,8 +29,7 @@ const TimeMs = { jest.mock('../../Utils/prometheus-metrics.js', () => ({ metrics: { socketEvents: { inc: jest.fn() }, - errors: { inc: jest.fn() }, - circuitBreakerTrips: { inc: jest.fn() } + errors: { inc: jest.fn() } } })) @@ -59,7 +58,6 @@ describe('UnifiedSessionManager', () => { manager = createUnifiedSessionManager({ enabled: true, logger: mockLogger as any, - enableCircuitBreaker: false, // Disable for simpler testing sendNode: mockSendNode }) }) @@ -356,7 +354,11 @@ describe('shouldEnableUnifiedSession', () => { }) }) -describe('UnifiedSessionManager with CircuitBreaker', () => { +describe('UnifiedSessionManager — non-critical send semantics', () => { + // The previous "with CircuitBreaker" suite is gone — telemetry no longer + // uses a circuit breaker. Verify equivalent behavior still holds: + // failures are swallowed (telemetry is non-critical) and don't propagate. + let manager: UnifiedSessionManager let mockSendNode: MockSendNode @@ -366,7 +368,6 @@ describe('UnifiedSessionManager with CircuitBreaker', () => { manager = createUnifiedSessionManager({ enabled: true, logger: createMockLogger() as any, - enableCircuitBreaker: true, sendNode: mockSendNode }) }) @@ -375,24 +376,13 @@ describe('UnifiedSessionManager with CircuitBreaker', () => { manager.destroy() }) - it('should work with circuit breaker enabled', async () => { + it('should send telemetry once per call', async () => { await manager.send('login') expect(mockSendNode).toHaveBeenCalledTimes(1) }) - it('should handle circuit breaker failures gracefully', async () => { - // Make send fail multiple times to trigger circuit breaker - let callCount = 0 - mockSendNode.mockImplementation(async () => { - callCount++ - if (callCount <= 3) { - throw new Error(`Fail ${callCount}`) - } - }) - - // These should not throw even with failures - await expect(manager.send('login')).resolves.toBeUndefined() - await expect(manager.send('login')).resolves.toBeUndefined() + it('should swallow send failures gracefully (telemetry is non-critical)', async () => { + mockSendNode.mockRejectedValueOnce(new Error('boom')) await expect(manager.send('login')).resolves.toBeUndefined() }) })