ruicesar/InfiniteAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(socket): integrate circuit breaker with Socket operations

Browse Source 
- Add circuit breaker configuration options to SocketConfig type
- Initialize circuit breakers for query, connection, and pre-key operations
- Wrap query() with circuit breaker protection for all WhatsApp queries
- Wrap sendRawMessage() with connection circuit breaker
- Integrate pre-key circuit breaker with uploadPreKeys()
- Add circuit breaker utilities to socket return object (stats, reset)
- Clean up circuit breakers on connection close
- Fix TypeScript errors in utility files (prometheus-metrics, logger-adapter, etc.)

Circuit breakers provide:
- Sliding window failure tracking
- Automatic state transitions (closed -> open -> half-open)
- Configurable thresholds and timeouts
- Prometheus metrics integration
- Event callbacks for monitoring
This commit is contained in:
Claude
2026-01-20 02:31:23 +00:00
parent 64c7712da7
commit 61f5f76ef0
9 changed files with 212 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Defaults/index.ts
+3 -1
View File
@@ -77,7 +77,9 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
countryCode: 'US',
getMessage: async () => undefined,
cachedGroupMetadata: async () => undefined,
makeSignalRepository: makeLibSignalRepository
makeSignalRepository: makeLibSignalRepository,
// Circuit breaker configuration
enableCircuitBreaker: true
}
export const MEDIA_PATH_MAP: { [T in MediaType]?: string } = {
src/Socket/socket.ts
+147 -7
View File
@@ -35,6 +35,12 @@ import {
xmppSignedPreKey
} from '../Utils'
import { getPlatformId } from '../Utils/browser-utils'
import {
CircuitBreaker,
CircuitOpenError,
createConnectionCircuitBreaker,
createPreKeyCircuitBreaker
} from '../Utils/circuit-breaker'
import {
assertNodeErrorFree,
type BinaryNode,
@@ -71,9 +77,64 @@ export const makeSocket = (config: SocketConfig) => {
defaultQueryTimeoutMs,
transactionOpts,
qrTimeout,
makeSignalRepository
makeSignalRepository,
enableCircuitBreaker = true,
queryCircuitBreaker: queryCircuitBreakerConfig,
connectionCircuitBreaker: connectionCircuitBreakerConfig,
preKeyCircuitBreaker: preKeyCircuitBreakerConfig
} = config
// 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')
}
const publicWAMBuffer = new BinaryInfo()
const uqTagId = generateMdTagPrefix()
@@ -110,8 +171,8 @@ export const makeSocket = (config: SocketConfig) => {
ws.connect()
const sendPromise = promisify(ws.send)
/** send a raw buffer */
const sendRawMessage = async (data: Uint8Array | Buffer) => {
/** send a raw buffer (internal implementation) */
const sendRawMessageInternal = async (data: Uint8Array | Buffer) => {
if (!ws.isOpen) {
throw new Boom('Connection Closed', { statusCode: DisconnectReason.connectionClosed })
}
@@ -127,6 +188,22 @@ export const makeSocket = (config: SocketConfig) => {
})
}
/** send a raw buffer with circuit breaker protection */
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)
}
/** send a binary node */
const sendNode = (frame: BinaryNode) => {
if (logger.level === 'trace') {
@@ -185,7 +262,7 @@ export const makeSocket = (config: SocketConfig) => {
}
/** send a query, and wait for its response. auto-generates message ID if not provided */
const query = async (node: BinaryNode, timeoutMs?: number) => {
const queryInternal = async (node: BinaryNode, timeoutMs?: number) => {
if (!node.attrs.id) {
node.attrs.id = generateMessageTag()
}
@@ -206,6 +283,25 @@ export const makeSocket = (config: SocketConfig) => {
return result
}
/** send a query with circuit breaker protection */
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)
}
// Validate current key-bundle on server; on failure, trigger pre-key upload and rethrow
const digestKeyBundle = async (): Promise<void> => {
const res = await query({
@@ -460,6 +556,12 @@ export const makeSocket = (config: SocketConfig) => {
/** 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
@@ -487,14 +589,28 @@ export const makeSocket = (config: SocketConfig) => {
return node // Only return node since update is already used
}, creds?.me?.id || 'upload-pre-keys')
// Upload to server (outside transaction, can fail without affecting local keys)
try {
// Upload to server with circuit breaker protection
const uploadToServer = async () => {
await query(node)
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()
}
} 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)
@@ -617,6 +733,11 @@ export const makeSocket = (config: SocketConfig) => {
clearInterval(keepAliveReq)
clearTimeout(qrTimer)
// Clean up circuit breakers
queryCircuitBreaker?.destroy()
connectionCircuitBreaker?.destroy()
preKeyCircuitBreaker?.destroy()
ws.removeAllListeners('close')
ws.removeAllListeners('open')
ws.removeAllListeners('message')
@@ -1058,7 +1179,26 @@ export const makeSocket = (config: SocketConfig) => {
waitForConnectionUpdate: bindWaitForConnectionUpdate(ev),
sendWAMBuffer,
executeUSyncQuery,
onWhatsApp
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')
}
}
}
src/Types/Socket.ts
+18
View File
@@ -2,6 +2,7 @@ import type { Agent } from 'https'
import type { URL } from 'url'
import { proto } from '../../WAProto/index.js'
import type { ILogger } from '../Utils/logger'
import type { CircuitBreakerOptions } from '../Utils/circuit-breaker'
import type { AuthenticationState, LIDMapping, SignalAuthState, TransactionCapabilityOptions } from './Auth'
import type { GroupMetadata } from './GroupMetadata'
import { type MediaConnInfo, type WAMessageKey } from './Message'
@@ -147,4 +148,21 @@ export type SocketConfig = {
logger: ILogger,
pnToLIDFunc?: (jids: string[]) => Promise<LIDMapping[] | undefined>
) => SignalRepositoryWithLIDStore
// === Circuit Breaker Configuration ===
/** Enable circuit breaker protection for socket operations (default: true) */
enableCircuitBreaker?: boolean
/** Circuit breaker configuration for query operations */
queryCircuitBreaker?: Partial<CircuitBreakerOptions>
/** Circuit breaker configuration for connection operations */
connectionCircuitBreaker?: Partial<CircuitBreakerOptions>
/** Circuit breaker configuration for pre-key operations */
preKeyCircuitBreaker?: Partial<CircuitBreakerOptions>
/** Circuit breaker configuration for message operations */
messageCircuitBreaker?: Partial<CircuitBreakerOptions>
}
src/Utils/baileys-event-stream.ts
+2 -1
View File
@@ -321,7 +321,8 @@ export class BaileysEventStream extends EventEmitter {
// Find correct position
let insertIndex = this.buffer.length
for (let i = 0; i < this.buffer.length; i++) {
if (PRIORITY_VALUES[this.buffer[i].priority] > eventPriorityValue) {
const bufferEvent = this.buffer[i]
if (bufferEvent && PRIORITY_VALUES[bufferEvent.priority] > eventPriorityValue) {
insertIndex = i
break
}
src/Utils/baileys-logger.ts
+27 -4
View File
@@ -331,8 +331,29 @@ export class BaileysLogger implements ILogger {
...(typeof sanitizedObj === 'object' && sanitizedObj !== null ? sanitizedObj : { value: sanitizedObj }),
}
// Structured log
this.structuredLogger[level](enrichedObj, msg)
// Structured log (using type-safe method call)
switch (level) {
case 'trace':
this.structuredLogger.trace(enrichedObj, msg)
break
case 'debug':
this.structuredLogger.debug(enrichedObj, msg)
break
case 'info':
this.structuredLogger.info(enrichedObj, msg)
break
case 'warn':
this.structuredLogger.warn(enrichedObj, msg)
break
case 'error':
this.structuredLogger.error(enrichedObj, msg)
break
case 'fatal':
this.structuredLogger.fatal(enrichedObj, msg)
break
default:
this.structuredLogger.info(enrichedObj, msg)
}
// Event handler
if (this.config.eventHandler) {
@@ -429,8 +450,10 @@ export class BaileysLogger implements ILogger {
if (process.env.NODE_ENV === 'production') {
// In production, mask part of the number
const parts = jid.split('@')
if (parts.length === 2 && parts[0].length > 4) {
return `${parts[0].substring(0, 4)}****@${parts[1]}`
const localPart = parts[0]
const domainPart = parts[1]
if (parts.length === 2 && localPart && domainPart && localPart.length > 4) {
return `${localPart.substring(0, 4)}****@${domainPart}`
}
}
return jid
src/Utils/cache-utils.ts
+2 -1
View File
@@ -500,7 +500,8 @@ export function withCache<T extends (...args: unknown[]) => unknown>(
/**
* Global singleton cache by namespace
*/
const globalCaches: Map<string, Cache<unknown>> = new Map()
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const globalCaches: Map<string, Cache<any>> = new Map()
export function getGlobalCache<V>(namespace: string, options?: CacheOptions<V>): Cache<V> {
if (!globalCaches.has(namespace)) {
src/Utils/logger-adapter.ts
+7 -3
View File
@@ -120,7 +120,7 @@ export class LoggerAdapter implements ILogger {
}
if (this.config.levelMapping && level in this.config.levelMapping) {
return this.config.levelMapping[level]
return this.config.levelMapping[level] || 'info'
}
return (level as LogLevel) || 'info'
@@ -327,8 +327,12 @@ function createLogMethod(
return (obj: unknown, msg?: string) => {
if (typeof logger[level] === 'function') {
;(logger[level] as (obj: unknown, msg?: string) => void)(obj, msg)
} else if (typeof (console as Record<string, unknown>)[level] === 'function') {
;(console as Record<string, (...args: unknown[]) => void>)[level](obj, msg)
} else {
// Fallback to console
const consoleMethod = (console as unknown as Record<string, (...args: unknown[]) => void>)[level]
if (typeof consoleMethod === 'function') {
consoleMethod(obj, msg)
}
}
}
}
src/Utils/prometheus-metrics.ts
+3 -3
View File
@@ -612,10 +612,10 @@ export class MetricsRegistry {
/**
* Return metrics in Prometheus format
*/
async metrics(): Promise<string> {
async getMetricsOutput(): Promise<string> {
const lines: string[] = []
for (const [name, metric] of this.metrics) {
for (const [name, metric] of this.metrics.entries()) {
lines.push(`# HELP ${name} ${metric.help}`)
lines.push(`# TYPE ${name} ${metric.type}`)
@@ -776,7 +776,7 @@ export const metrics = {
*/
export function createMetricsHandler(registry: MetricsRegistry = baileysMetrics) {
return async (_req: unknown, res: { setHeader: (name: string, value: string) => void; end: (body: string) => void }) => {
const metricsOutput = await registry.metrics()
const metricsOutput = await registry.getMetricsOutput()
res.setHeader('Content-Type', registry.contentType())
res.end(metricsOutput)
}
src/Utils/trace-context.ts
+3 -2
View File
@@ -554,9 +554,10 @@ export function extractTraceHeaders(headers: Record<string, string | undefined>)
}
// Parse baggage
if (headers[TRACE_HEADERS.BAGGAGE]) {
const baggageHeader = headers[TRACE_HEADERS.BAGGAGE]
if (baggageHeader) {
options.baggage = {}
const pairs = headers[TRACE_HEADERS.BAGGAGE].split(',')
const pairs = baggageHeader.split(',')
for (const pair of pairs) {
const [key, value] = pair.split('=')
if (key && value) {