refactor(utils): remove circuit-breaker integration from helpers
- Utils/index.ts: re-export bounded-retry instead of circuit-breaker - Utils/unified-session.ts: drop the optional circuit-breaker wrapping. Telemetry is non-critical; a single send attempt with the underlying sendNode timeout is sufficient. Removed enableCircuitBreaker option, circuitBreaker field, and CircuitOpenError handling. - Utils/retry-utils.ts: drop the optional circuitBreaker option from withRetry. Retry mechanics are now self-contained; callers that need bounded retries should prefer withBoundedRetry directly. - Utils/health-status.ts: stop importing globalCircuitRegistry. Keep the CircuitBreakerHealth shape for k8s probe schema stability — always reports an empty list now. Note: src/Utils/structured-logger.ts has its own *private* CircuitBreaker class for log-throttling — independent of the socket-level circuit breaker being removed. Untouched. baileys-logger.ts logCircuitBreaker() is a generic logging helper that takes state strings as input — no import dependency. Untouched (no-op for callers that no longer have circuit breakers).
This commit is contained in:
@@ -6,11 +6,12 @@
|
|||||||
* @module Utils/health-status
|
* @module Utils/health-status
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import { globalCircuitRegistry } from './circuit-breaker.js'
|
|
||||||
import { getVersionCacheStatus } from './version-cache.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 {
|
export interface CircuitBreakerHealth {
|
||||||
name: string
|
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[] = []
|
const circuitBreakers: CircuitBreakerHealth[] = []
|
||||||
let openCircuits = 0
|
checks.push({ name: 'circuit_breakers', status: 'pass' })
|
||||||
|
|
||||||
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'
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
// 3. Check memory usage (warn if > 90%)
|
// 3. Check memory usage (warn if > 90%)
|
||||||
const memUsage = process.memoryUsage()
|
const memUsage = process.memoryUsage()
|
||||||
|
|||||||
+1
-1
@@ -33,8 +33,8 @@ export * from './trace-context'
|
|||||||
export * from './prometheus-metrics'
|
export * from './prometheus-metrics'
|
||||||
|
|
||||||
// Resilience and performance
|
// Resilience and performance
|
||||||
|
export * from './bounded-retry'
|
||||||
export * from './cache-utils'
|
export * from './cache-utils'
|
||||||
export * from './circuit-breaker'
|
|
||||||
export * from './retry-utils'
|
export * from './retry-utils'
|
||||||
|
|
||||||
// Telemetry and detection mitigation
|
// Telemetry and detection mitigation
|
||||||
|
|||||||
@@ -14,7 +14,6 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import { EventEmitter } from 'events'
|
import { EventEmitter } from 'events'
|
||||||
import type { CircuitBreaker } from './circuit-breaker.js'
|
|
||||||
import { metrics } from './prometheus-metrics.js'
|
import { metrics } from './prometheus-metrics.js'
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -59,8 +58,6 @@ export interface RetryOptions {
|
|||||||
operationName?: string
|
operationName?: string
|
||||||
/** Collect metrics */
|
/** Collect metrics */
|
||||||
collectMetrics?: boolean
|
collectMetrics?: boolean
|
||||||
/** Circuit breaker for integration */
|
|
||||||
circuitBreaker?: CircuitBreaker
|
|
||||||
/** Callback before each retry */
|
/** Callback before each retry */
|
||||||
onRetry?: (error: Error, attempt: number, delay: number) => void | Promise<void>
|
onRetry?: (error: Error, attempt: number, delay: number) => void | Promise<void>
|
||||||
/** Callback on success */
|
/** Callback on success */
|
||||||
@@ -283,7 +280,6 @@ export async function retry<T>(
|
|||||||
timeout: options.timeout,
|
timeout: options.timeout,
|
||||||
operationName: options.operationName ?? 'operation',
|
operationName: options.operationName ?? 'operation',
|
||||||
collectMetrics: options.collectMetrics ?? true,
|
collectMetrics: options.collectMetrics ?? true,
|
||||||
circuitBreaker: options.circuitBreaker,
|
|
||||||
onRetry: options.onRetry ?? (() => {}),
|
onRetry: options.onRetry ?? (() => {}),
|
||||||
onSuccess: options.onSuccess ?? (() => {}),
|
onSuccess: options.onSuccess ?? (() => {}),
|
||||||
onFailure: options.onFailure ?? (() => {}),
|
onFailure: options.onFailure ?? (() => {}),
|
||||||
@@ -313,11 +309,6 @@ export async function retry<T>(
|
|||||||
throw new RetryAbortedError(attempt)
|
throw new RetryAbortedError(attempt)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check circuit breaker
|
|
||||||
if (config.circuitBreaker?.isOpen()) {
|
|
||||||
throw new Error(`Circuit breaker "${config.circuitBreaker.getName()}" is open`)
|
|
||||||
}
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Execute operation
|
// Execute operation
|
||||||
let result: T
|
let result: T
|
||||||
|
|||||||
@@ -15,7 +15,6 @@
|
|||||||
*/
|
*/
|
||||||
|
|
||||||
import type { BinaryNode } from '../WABinary/types.js'
|
import type { BinaryNode } from '../WABinary/types.js'
|
||||||
import { CircuitBreaker, CircuitOpenError, createConnectionCircuitBreaker } from './circuit-breaker.js'
|
|
||||||
import type { ILogger } from './logger.js'
|
import type { ILogger } from './logger.js'
|
||||||
import { metrics } from './prometheus-metrics.js'
|
import { metrics } from './prometheus-metrics.js'
|
||||||
|
|
||||||
@@ -58,8 +57,6 @@ export interface UnifiedSessionOptions {
|
|||||||
enabled?: boolean
|
enabled?: boolean
|
||||||
/** Logger instance for debugging */
|
/** Logger instance for debugging */
|
||||||
logger?: ILogger
|
logger?: ILogger
|
||||||
/** Enable circuit breaker protection for send operations */
|
|
||||||
enableCircuitBreaker?: boolean
|
|
||||||
/** Function to send binary nodes to WhatsApp */
|
/** Function to send binary nodes to WhatsApp */
|
||||||
sendNode?: (node: BinaryNode) => Promise<void>
|
sendNode?: (node: BinaryNode) => Promise<void>
|
||||||
}
|
}
|
||||||
@@ -121,7 +118,6 @@ export class UnifiedSessionManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private readonly options: Required<UnifiedSessionOptions>
|
private readonly options: Required<UnifiedSessionOptions>
|
||||||
private circuitBreaker: CircuitBreaker | null = null
|
|
||||||
|
|
||||||
/** Minimum interval between unified_session sends (1 minute) */
|
/** Minimum interval between unified_session sends (1 minute) */
|
||||||
private static readonly MIN_SEND_INTERVAL_MS = TimeMs.Minute
|
private static readonly MIN_SEND_INTERVAL_MS = TimeMs.Minute
|
||||||
@@ -130,7 +126,6 @@ export class UnifiedSessionManager {
|
|||||||
this.options = {
|
this.options = {
|
||||||
enabled: options.enabled ?? true,
|
enabled: options.enabled ?? true,
|
||||||
logger: options.logger ?? (console as unknown as ILogger),
|
logger: options.logger ?? (console as unknown as ILogger),
|
||||||
enableCircuitBreaker: options.enableCircuitBreaker ?? true,
|
|
||||||
sendNode:
|
sendNode:
|
||||||
options.sendNode ??
|
options.sendNode ??
|
||||||
(async () => {
|
(async () => {
|
||||||
@@ -138,25 +133,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.state.isInitialized = true
|
||||||
this.options.logger.debug?.('UnifiedSessionManager initialized')
|
this.options.logger.debug?.('UnifiedSessionManager initialized')
|
||||||
}
|
}
|
||||||
@@ -292,20 +268,12 @@ export class UnifiedSessionManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Execute with circuit breaker if available
|
// Telemetry is non-critical: a single send attempt with the
|
||||||
if (this.circuitBreaker) {
|
// underlying sendNode timeout is enough. No retry/circuit needed.
|
||||||
await this.circuitBreaker.execute(sendOperation)
|
|
||||||
} else {
|
|
||||||
await sendOperation()
|
await sendOperation()
|
||||||
}
|
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
const errorMessage = error instanceof Error ? error.message : String(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
|
// Record failure metric
|
||||||
metrics.errors?.inc({ category: 'unified_session', code: 'send_failed' })
|
metrics.errors?.inc({ category: 'unified_session', code: 'send_failed' })
|
||||||
@@ -331,7 +299,6 @@ export class UnifiedSessionManager {
|
|||||||
sendCount: 0,
|
sendCount: 0,
|
||||||
isInitialized: true
|
isInitialized: true
|
||||||
}
|
}
|
||||||
this.circuitBreaker?.reset()
|
|
||||||
this.options.logger.debug?.('UnifiedSessionManager reset')
|
this.options.logger.debug?.('UnifiedSessionManager reset')
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -339,7 +306,6 @@ export class UnifiedSessionManager {
|
|||||||
* Destroy the manager and clean up resources.
|
* Destroy the manager and clean up resources.
|
||||||
*/
|
*/
|
||||||
destroy(): void {
|
destroy(): void {
|
||||||
this.circuitBreaker?.destroy()
|
|
||||||
this.state.isInitialized = false
|
this.state.isInitialized = false
|
||||||
this.options.logger.debug?.('UnifiedSessionManager destroyed')
|
this.options.logger.debug?.('UnifiedSessionManager destroyed')
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user