diff --git a/src/Signal/lid-mapping.ts b/src/Signal/lid-mapping.ts index 636f2868..66b064fb 100644 --- a/src/Signal/lid-mapping.ts +++ b/src/Signal/lid-mapping.ts @@ -173,10 +173,6 @@ export class LIDMappingStore { lastOperationAt: null } - // Metrics integration (lazy loaded) - private metricsModule: typeof import('../Utils/prometheus-metrics') | null = null - private metricsQueue: Array<() => void> = [] - constructor( keys: SignalKeyStoreWithTransaction, logger: ILogger, @@ -196,19 +192,6 @@ export class LIDMappingStore { updateAgeOnGet: this.config.updateAgeOnGet }) - // Load metrics module if enabled - if (this.config.enableMetrics) { - import('../Utils/prometheus-metrics').then(m => { - this.metricsModule = m - this.logger.debug('📊 Prometheus metrics loaded for LID mapping, flushing buffered metrics', { queuedCount: this.metricsQueue.length }) - // Flush buffered metrics - this.metricsQueue.forEach(fn => fn()) - this.metricsQueue = [] - }).catch(() => { - this.logger.debug('Prometheus metrics not available for LID mapping') - }) - } - this.logger.debug({ config: this.config }, 'LIDMappingStore initialized') } @@ -970,20 +953,11 @@ export class LIDMappingStore { /** * Record metrics if enabled (with buffer support for async loading) + * Note: Actual metric recording is not yet implemented */ private recordMetrics(operation: string, count: number): void { - if (this.metricsModule) { - try { - // Use the metrics registry to record LID mapping operations - // This integrates with the existing Prometheus metrics - } catch { - // Ignore metrics errors - } - } else { - // Buffer metric call until module loads - this.metricsQueue.push(() => { - // Metrics will be recorded when module loads - }) - } + // Metrics implementation pending - currently a no-op + // When implemented, should record LID mapping operations to Prometheus + // For now, we don't buffer since there's no actual recording function } } diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 644b7b12..eb5a488b 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -505,14 +505,11 @@ export const makeSocket = (config: SocketConfig) => { let qrTimer: NodeJS.Timeout let closed = false - // Auto-reconnect state for session errors - const MAX_RECONNECT_ATTEMPTS = 5 - let reconnectAttempts = 0 - // Session TTL and cleanup const SESSION_TTL = 7 * 24 * 60 * 60 * 1000 // 7 days let sessionStartTime: number | undefined let ttlTimer: NodeJS.Timeout | undefined + let ttlGraceTimer: NodeJS.Timeout | undefined /** log & process any unexpected errors */ const onUnexpectedError = (err: Error | Boom, msg: string) => { @@ -739,6 +736,7 @@ export const makeSocket = (config: SocketConfig) => { /** * PreKey Auto-Sync: Proactive validation every 6 hours * Prevents "Identity key field not found" errors by ensuring keys are always valid + * Returns cleanup function to remove event listener */ const startPreKeyAutoSync = () => { const SYNC_INTERVAL = 6 * 60 * 60 * 1000 // 6 hours @@ -769,18 +767,21 @@ export const makeSocket = (config: SocketConfig) => { isRunning = false } - // PROTECTION 5: Reschedule AFTER completion (not from start) - syncTimer = setTimeout(syncLoop, SYNC_INTERVAL) + // PROTECTION 3: Prevent timer accumulation + // Only reschedule if connection is still open + if (!closed && ws.isOpen) { + syncTimer = setTimeout(syncLoop, SYNC_INTERVAL) + } } - // PROTECTION 6: Initial delay (avoid duplicate at startup) + // PROTECTION 4: Initial delay (avoid duplicate at startup) // CB:success already calls uploadPreKeysToServerIfRequired(), so wait 6h before first auto-sync - ev.on('connection.update', ({ connection }) => { + const connectionHandler = ({ connection }: { connection: any }) => { if (connection === 'open') { logger.info('🔑 Starting PreKey auto-sync timer (first sync in 6h)') syncTimer = setTimeout(syncLoop, SYNC_INTERVAL) } else if (connection === 'close') { - // PROTECTION 7: Cleanup on disconnect + // PROTECTION 5: Cleanup on disconnect if (syncTimer) { clearTimeout(syncTimer) syncTimer = undefined @@ -788,18 +789,30 @@ export const makeSocket = (config: SocketConfig) => { logger.info('🔑 PreKey auto-sync stopped') } } - }) + } + + ev.on('connection.update', connectionHandler) + + // PROTECTION 6: Return cleanup function + return () => { + ev.off('connection.update', connectionHandler) + if (syncTimer) { + clearTimeout(syncTimer) + syncTimer = undefined + } + } } - // Initialize PreKey auto-sync - startPreKeyAutoSync() + // Initialize PreKey auto-sync and store cleanup function + const cleanupPreKeyAutoSync = startPreKeyAutoSync() /** * Session TTL Management: Graceful cleanup after 7 days * Prevents memory leaks and allows credential rotation in long-running sessions + * Returns cleanup function to remove event listener */ const startSessionTTL = () => { - ev.on('connection.update', ({ connection }) => { + const connectionHandler = ({ connection }: { connection: any }) => { if (connection === 'open') { sessionStartTime = Date.now() @@ -816,8 +829,8 @@ export const makeSocket = (config: SocketConfig) => { duration: duration }) - // PROTECTION 4: Graceful delay before cleanup - setTimeout(() => { + // PROTECTION 3: Graceful delay before cleanup (with proper cleanup) + ttlGraceTimer = setTimeout(() => { logger.info('🕐 Proceeding with TTL cleanup after grace period') end(new Error('SESSION_TTL_EXPIRED')) }, 5000) // 5s grace period @@ -826,19 +839,38 @@ export const makeSocket = (config: SocketConfig) => { const ttlHours = SESSION_TTL / 1000 / 60 / 60 logger.info(`🕐 Session TTL started (${ttlHours}h = 7 days)`) } else if (connection === 'close') { - // PROTECTION 3: Cleanup timer on disconnect + // PROTECTION 4: Cleanup ALL timers on disconnect if (ttlTimer) { clearTimeout(ttlTimer) ttlTimer = undefined - sessionStartTime = undefined - logger.info('🕐 Session TTL timer cleared') } + if (ttlGraceTimer) { + clearTimeout(ttlGraceTimer) + ttlGraceTimer = undefined + } + sessionStartTime = undefined + logger.info('🕐 Session TTL timers cleared') } - }) + } + + ev.on('connection.update', connectionHandler) + + // PROTECTION 5: Return cleanup function + return () => { + ev.off('connection.update', connectionHandler) + if (ttlTimer) { + clearTimeout(ttlTimer) + ttlTimer = undefined + } + if (ttlGraceTimer) { + clearTimeout(ttlGraceTimer) + ttlGraceTimer = undefined + } + } } - // Initialize Session TTL - startSessionTTL() + // Initialize Session TTL and store cleanup function + const cleanupSessionTTL = startSessionTTL() const onMessageReceived = async (data: Buffer) => { await noise.decodeFrame(data, frame => { @@ -948,6 +980,12 @@ export const makeSocket = (config: SocketConfig) => { // Clean up transaction capability (PreKeyManager + queues) keys.destroy?.() + // Clean up PreKey auto-sync event listener + cleanupPreKeyAutoSync() + + // Clean up Session TTL event listener + cleanupSessionTTL() + ws.removeAllListeners('close') ws.removeAllListeners('open') ws.removeAllListeners('message') @@ -1366,46 +1404,24 @@ export const makeSocket = (config: SocketConfig) => { // update credentials when required ev.on('creds.update', async update => { - // CRITICAL: Handle session errors with auto-reconnect + // CRITICAL: Handle session errors by emitting close event for consumer-level reconnect if (update.error) { - logger.error({ error: update.error }, '🔴 Session error detected - initiating auto-reconnect') + logger.error({ error: update.error }, '🔴 Session error detected') - // PROTECTION 1: Max attempts guard - if (reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) { - logger.error(`❌ Max reconnect attempts (${MAX_RECONNECT_ATTEMPTS}) reached, giving up`) - ev.emit('connection.update', { - connection: 'close', - lastDisconnect: { - error: update.error, - date: new Date() - } - }) - return - } - - reconnectAttempts++ - logger.info(`🔄 Reconnect attempt ${reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS}`) - - // PROTECTION 4: Cleanup before reconnect - await end(update.error) - - // PROTECTION 2: Exponential backoff - const delay = Math.min(1000 * Math.pow(2, reconnectAttempts - 1), 30000) - logger.info(`⏳ Waiting ${delay}ms before reconnect (exponential backoff)`) - await new Promise(resolve => setTimeout(resolve, delay)) - - // Attempt reconnect - logger.info(`🔄 Reconnecting now (attempt ${reconnectAttempts})`) - await connect() - - // PROTECTION 3: Reset counter on success - ev.once('connection.update', ({ connection }) => { - if (connection === 'open') { - logger.info(`✅ Reconnect successful, resetting attempt counter`) - reconnectAttempts = 0 - } + // Session errors indicate keys are desynchronized - socket must be recreated + // Emit close event so consumer can call makeWASocket() again + ev.emit('connection.update', { + connection: 'close', + lastDisconnect: { + error: update.error, + date: new Date() + }, + // Include session error flag for consumer to detect + isSessionError: true }) + // Cleanup current socket + await end(update.error) return } diff --git a/src/Types/Events.ts b/src/Types/Events.ts index d15fa6c7..9277cf2c 100644 --- a/src/Types/Events.ts +++ b/src/Types/Events.ts @@ -153,6 +153,18 @@ export type BaileysEventMap = { /** Whether this is a new contact (true) or an existing contact with changed key (false) */ isNewContact: boolean } + + /** + * Emitted when the session TTL (Time-To-Live) expires after 7 days. + * Applications can listen to this event to perform graceful cleanup, + * flush pending operations, or rotate credentials before the socket closes. + */ + 'session.ttl-expired': { + /** Timestamp when the session started (Date.now()) */ + startTime: number | undefined + /** Duration of the session in milliseconds */ + duration: number + } } export type BufferedEventData = { diff --git a/src/Types/State.ts b/src/Types/State.ts index 1b9e40df..72558b83 100644 --- a/src/Types/State.ts +++ b/src/Types/State.ts @@ -40,4 +40,10 @@ export type ConnectionState = { * If this is false, the primary phone and other devices will receive notifs * */ isOnline?: boolean + /** + * indicates the disconnect was caused by a session error (keys desynchronized). + * When true, the consumer should recreate the socket with makeWASocket() + * to establish a fresh session. + */ + isSessionError?: boolean } diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index e1617f07..612f9f1b 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -421,6 +421,8 @@ export const makeEventBuffer = ( // Metrics integration (lazy loaded to avoid circular deps) let metricsModule: typeof import('./prometheus-metrics') | null = null let metricsQueue: Array<() => void> = [] + let metricsImportFailed = false + const MAX_METRICS_QUEUE_SIZE = 1000 // Cap to prevent unbounded growth if (config.enableMetrics) { import('./prometheus-metrics').then(m => { @@ -431,6 +433,9 @@ export const makeEventBuffer = ( metricsQueue = [] }).catch(() => { logger.debug('Prometheus metrics not available for event buffer') + metricsImportFailed = true + // Clear queue to prevent memory leak + metricsQueue = [] }) } @@ -438,17 +443,17 @@ export const makeEventBuffer = ( const recordMetrics = (eventType: string, count: number) => { if (metricsModule) { metricsModule.recordEventBuffered(eventType, count) - } else { - // Buffer metric call until module loads + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { + // Buffer metric call until module loads (with size limit) metricsQueue.push(() => metricsModule?.recordEventBuffered(eventType, count)) } + // If import failed or queue is full, silently drop metric } const recordFlushMetrics = (eventCount: number, forced: boolean, cacheSize: number) => { if (metricsModule) { metricsModule.recordBufferFlush(eventCount, forced, cacheSize) - } else { - // Buffer metric call until module loads + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { metricsQueue.push(() => metricsModule?.recordBufferFlush(eventCount, forced, cacheSize)) } } @@ -489,7 +494,7 @@ export const makeEventBuffer = ( // Record overflow metric if (metricsModule) { metricsModule.recordBufferOverflow() - } else { + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { metricsQueue.push(() => metricsModule?.recordBufferOverflow()) } flush(true) @@ -528,7 +533,7 @@ export const makeEventBuffer = ( // Record metrics for cache cleanup if (metricsModule) { metricsModule.recordCacheCleanup(removed.length) - } else { + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { metricsQueue.push(() => metricsModule?.recordCacheCleanup(removed.length)) } } @@ -608,7 +613,7 @@ export const makeEventBuffer = ( if (config.enableAdaptiveTimeout) { if (metricsModule) { metricsModule.updateAdaptiveMetrics(adaptiveTimeout.getEventRate(), adaptiveTimeout.isHealthy()) - } else { + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { metricsQueue.push(() => metricsModule?.updateAdaptiveMetrics(adaptiveTimeout.getEventRate(), adaptiveTimeout.isHealthy())) } } @@ -666,7 +671,7 @@ export const makeEventBuffer = ( // Record final flush metric if (metricsModule) { metricsModule.recordBufferFinalFlush() - } else { + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { metricsQueue.push(() => metricsModule?.recordBufferFinalFlush()) } } @@ -684,7 +689,7 @@ export const makeEventBuffer = ( // Record buffer destroyed metric if (metricsModule) { metricsModule.recordBufferDestroyed('normal', hadPendingFlush) - } else { + } else if (!metricsImportFailed && metricsQueue.length < MAX_METRICS_QUEUE_SIZE) { metricsQueue.push(() => metricsModule?.recordBufferDestroyed('normal', hadPendingFlush)) }