diff --git a/src/Socket/chats.ts b/src/Socket/chats.ts index 8ee92f57..ac9eb6ae 100644 --- a/src/Socket/chats.ts +++ b/src/Socket/chats.ts @@ -1465,7 +1465,12 @@ export const makeChatsSocket = (config: SocketConfig) => { return } - logger.info('First connection, awaiting history sync notification with a 20s timeout.') + // perf(inbound-latency): reduced from 20s → 8s. On first connection we wait for the + // history-sync notification so that doAppStateSync runs before live messages are emitted. + // If the notification does not arrive within 8s we stop waiting, go Online, and flush + // so that any live message arriving after connection is never held more than ~8s. + // History that arrives late is still processed via processMessage regardless of state. + logger.info('First connection, awaiting history sync notification with an 8s timeout.') if (awaitingSyncTimeout) { clearTimeout(awaitingSyncTimeout) @@ -1473,18 +1478,17 @@ export const makeChatsSocket = (config: SocketConfig) => { awaitingSyncTimeout = setTimeout(() => { if (syncState === SyncState.AwaitingInitialSync) { - // TODO: investigate - logger.warn('Timeout in AwaitingInitialSync, forcing state to Online and flushing buffer') + logger.warn('Timeout in AwaitingInitialSync (8s), forcing state to Online and flushing buffer') syncState = SyncState.Online ev.flush() - // Increment so subsequent reconnections skip the 20s wait. + // Increment so subsequent reconnections skip the wait entirely. // Late-arriving history is still processed via processMessage // regardless of the state machine phase. const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1 ev.emit('creds.update', { accountSyncCounter }) } - }, 20_000) + }, 8_000) }) // When an app state sync key arrives (myAppStateKeyId is set) and there are diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 75f69d7b..342f1d5f 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -1110,6 +1110,17 @@ export const makeSocket = (config: SocketConfig) => { clearInterval(keepAliveReq) clearTimeout(qrTimer) + // Clear offline-buffer safety timer so its callback cannot call ev.flush() + // on an already-closed socket (e.g. auth failure or early network drop before + // CB:ib,,offline ever arrives). Mirrors how awaitingSyncTimeout is cleared in + // chats.ts on connection close. + if (offlineBufferTimeout) { + clearTimeout(offlineBufferTimeout) + offlineBufferTimeout = undefined + } + + didStartBuffer = false + // Stop session cleanup scheduler sessionCleanup.stop() @@ -1572,12 +1583,38 @@ export const makeSocket = (config: SocketConfig) => { }) let didStartBuffer = false + // perf(inbound-latency): safety timer that caps how long the offline-phase buffer may + // block live message delivery. The server sends CB:ib,,offline only after transmitting + // ALL queued offline messages; on busy accounts this can take 10-30+ seconds, holding + // every buffered event hostage. This timer fires after OFFLINE_BUFFER_TIMEOUT_MS and + // force-flushes so live messages are never delayed beyond that cap. + // CB:ib,,offline clears the timer when it fires normally (fast path). + // + // INTENTIONALLY hardcoded — not controlled by BAILEYS_BUFFER_TIMEOUT_MS or any other + // env var. BAILEYS_BUFFER_TIMEOUT_MS governs general-purpose buffer batching (for + // Prometheus / history consolidation) and is typically set to 5-30 s by operators. + // This constant must remain short regardless so that a large offline backlog cannot + // hold live incoming messages hostage for minutes. + const OFFLINE_BUFFER_TIMEOUT_MS = 5_000 + let offlineBufferTimeout: NodeJS.Timeout | undefined + process.nextTick(() => { if (creds.me?.id) { // start buffering important events // if we're logged in ev.buffer() didStartBuffer = true + + // Cap the offline-phase buffer so a large backlog of missed messages + // does not delay delivery of the very next live message beyond OFFLINE_BUFFER_TIMEOUT_MS. + offlineBufferTimeout = setTimeout(() => { + offlineBufferTimeout = undefined + if (didStartBuffer) { + logger.warn({ timeoutMs: OFFLINE_BUFFER_TIMEOUT_MS }, 'perf: offline-buffer safety timeout reached, force-flushing before CB:ib,,offline') + ev.flush() + didStartBuffer = false + } + }, OFFLINE_BUFFER_TIMEOUT_MS) } ev.emit('connection.update', { connection: 'connecting', receivedPendingNotifications: false, qr: undefined }) @@ -1589,8 +1626,16 @@ export const makeSocket = (config: SocketConfig) => { const offlineNotifs = +(child?.attrs.count || 0) logger.info(`handled ${offlineNotifs} offline messages/notifications`) + + // Clear safety timeout — CB:ib,,offline arrived in time, do the normal flush. + if (offlineBufferTimeout) { + clearTimeout(offlineBufferTimeout) + offlineBufferTimeout = undefined + } + if (didStartBuffer) { ev.flush() + didStartBuffer = false logger.trace('flushed events for initial buffer') } diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index f49fb021..24bf36bb 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -54,9 +54,12 @@ export interface BufferConfig { */ export function loadBufferConfig(): BufferConfig { return { - bufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_TIMEOUT_MS || '15000', 10), - minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '3000', 10), - maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '20000', 10), + // perf(inbound-latency): reduced from 15s → 5s so the safety auto-flush fires sooner. + // processNodeWithBuffer always calls ev.flush() explicitly (no-op for this timer), + // but socket.ts's offline-phase buffer and any stalled buffer benefit from the lower cap. + bufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_TIMEOUT_MS || '5000', 10), + minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '1000', 10), + maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '8000', 10), maxHistoryCacheSize: parseInt(process.env.BAILEYS_BUFFER_MAX_HISTORY_CACHE || '10000', 10), maxBufferSize: parseInt(process.env.BAILEYS_BUFFER_MAX_SIZE || '5000', 10), flushDebounceMs: parseInt(process.env.BAILEYS_BUFFER_FLUSH_DEBOUNCE_MS || '100', 10), diff --git a/src/__tests__/Socket/offline-buffer-timeout.test.ts b/src/__tests__/Socket/offline-buffer-timeout.test.ts new file mode 100644 index 00000000..3b4fc69c --- /dev/null +++ b/src/__tests__/Socket/offline-buffer-timeout.test.ts @@ -0,0 +1,258 @@ +import { jest } from '@jest/globals' + +/** + * Tests for the offline-buffer safety timer introduced in socket.ts. + * + * The timer caps how long the CB:ib,,offline phase can block live message + * delivery. Its behaviour spans three interaction points that must all be + * correct for the feature to work safely: + * + * 1. startBuffer() – called inside process.nextTick when the socket connects + * and creds.me?.id is set. Arms the timer. + * 2. onOffline() – called when CB:ib,,offline arrives (happy path). + * Must cancel the timer and flush exactly once. + * 3. onClose() – called inside end() when the socket closes for any + * reason. Must cancel the timer so the callback cannot + * call ev.flush() on a dead session. + * + * Because these closures live deep inside makeSocket we mirror their logic + * here as standalone functions, exactly the same approach used in + * bad-ack-handling.test.ts. + */ + +const OFFLINE_BUFFER_TIMEOUT_MS = 5_000 + +/** Mirrors the state variables declared at the top of makeSocket */ +interface OfflineBufferState { + didStartBuffer: boolean + offlineBufferTimeout: NodeJS.Timeout | undefined +} + +function makeState(): OfflineBufferState { + return { didStartBuffer: false, offlineBufferTimeout: undefined } +} + +/** + * Mirrors the process.nextTick block that arms the offline-buffer timer. + * Only called when creds.me?.id is set (reconnection path). + */ +function startBuffer( + state: OfflineBufferState, + flush: () => void, + warn: () => void +): void { + state.didStartBuffer = true + state.offlineBufferTimeout = setTimeout(() => { + state.offlineBufferTimeout = undefined + if(state.didStartBuffer) { + warn() + flush() + state.didStartBuffer = false + } + }, OFFLINE_BUFFER_TIMEOUT_MS) +} + +/** + * Mirrors the CB:ib,,offline handler — the happy path where the server + * delivers all offline notifications before the safety timer fires. + */ +function onOffline(state: OfflineBufferState, flush: () => void): void { + if(state.offlineBufferTimeout) { + clearTimeout(state.offlineBufferTimeout) + state.offlineBufferTimeout = undefined + } + + if(state.didStartBuffer) { + flush() + state.didStartBuffer = false + } +} + +/** + * Mirrors the relevant portion of end() — clears the timer and resets the + * flag so a closing socket cannot emit stale events after the fact. + */ +function onClose(state: OfflineBufferState): void { + if(state.offlineBufferTimeout) { + clearTimeout(state.offlineBufferTimeout) + state.offlineBufferTimeout = undefined + } + + state.didStartBuffer = false +} + +// --------------------------------------------------------------------------- + +describe('offline-buffer safety timer (socket.ts)', () => { + let state: OfflineBufferState + let mockFlush: jest.Mock + let mockWarn: jest.Mock + + beforeEach(() => { + jest.useFakeTimers() + state = makeState() + mockFlush = jest.fn() + mockWarn = jest.fn() + }) + + afterEach(() => { + // Clean up any remaining timer to avoid cross-test interference + if(state.offlineBufferTimeout) { + clearTimeout(state.offlineBufferTimeout) + } + + jest.useRealTimers() + }) + + // ------------------------------------------------------------------------- + // 1. Timeout path — CB:ib,,offline never arrives within 5 s + // ------------------------------------------------------------------------- + + it('fires after 5 s and flushes when CB:ib,,offline is delayed', () => { + startBuffer(state, mockFlush, mockWarn) + + expect(mockFlush).not.toHaveBeenCalled() + + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(mockWarn).toHaveBeenCalledTimes(1) + expect(mockFlush).toHaveBeenCalledTimes(1) + }) + + it('resets didStartBuffer to false after the timeout fires', () => { + startBuffer(state, mockFlush, mockWarn) + expect(state.didStartBuffer).toBe(true) + + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(state.didStartBuffer).toBe(false) + }) + + it('sets offlineBufferTimeout to undefined after the callback executes', () => { + startBuffer(state, mockFlush, mockWarn) + + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(state.offlineBufferTimeout).toBeUndefined() + }) + + it('does not flush if didStartBuffer is already false when timeout fires', () => { + startBuffer(state, mockFlush, mockWarn) + // Simulate external reset (e.g. onClose was called before the timer fired) + state.didStartBuffer = false + + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(mockFlush).not.toHaveBeenCalled() + expect(mockWarn).not.toHaveBeenCalled() + }) + + // ------------------------------------------------------------------------- + // 2. Happy path — CB:ib,,offline arrives before the 5 s timer fires + // ------------------------------------------------------------------------- + + it('CB:ib,,offline cancels the timer and flushes exactly once', () => { + startBuffer(state, mockFlush, mockWarn) + + // Server responds before the 5 s timeout + jest.advanceTimersByTime(1_000) + onOffline(state, mockFlush) + + // Timer should be cancelled — advancing past 5 s must not cause a second flush + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(mockFlush).toHaveBeenCalledTimes(1) + expect(mockWarn).not.toHaveBeenCalled() + }) + + it('CB:ib,,offline resets didStartBuffer to false', () => { + startBuffer(state, mockFlush, mockWarn) + onOffline(state, mockFlush) + + expect(state.didStartBuffer).toBe(false) + }) + + it('CB:ib,,offline clears offlineBufferTimeout reference', () => { + startBuffer(state, mockFlush, mockWarn) + onOffline(state, mockFlush) + + expect(state.offlineBufferTimeout).toBeUndefined() + }) + + it('CB:ib,,offline is idempotent when called twice (no double flush)', () => { + startBuffer(state, mockFlush, mockWarn) + onOffline(state, mockFlush) + onOffline(state, mockFlush) // spurious second call + + expect(mockFlush).toHaveBeenCalledTimes(1) + }) + + // ------------------------------------------------------------------------- + // 3. Socket close path — end() called before CB:ib,,offline or timer fires + // ------------------------------------------------------------------------- + + it('end() cancels the timer so the callback never flushes after socket close', () => { + startBuffer(state, mockFlush, mockWarn) + + onClose(state) + + // Timer must be gone — advancing past 5 s must not trigger any flush + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(mockFlush).not.toHaveBeenCalled() + expect(mockWarn).not.toHaveBeenCalled() + }) + + it('end() resets didStartBuffer to false', () => { + startBuffer(state, mockFlush, mockWarn) + expect(state.didStartBuffer).toBe(true) + + onClose(state) + + expect(state.didStartBuffer).toBe(false) + }) + + it('end() clears offlineBufferTimeout reference', () => { + startBuffer(state, mockFlush, mockWarn) + onClose(state) + + expect(state.offlineBufferTimeout).toBeUndefined() + }) + + it('end() is safe to call when no buffer was started', () => { + // State has never been touched — must not throw + expect(() => onClose(state)).not.toThrow() + expect(state.didStartBuffer).toBe(false) + expect(state.offlineBufferTimeout).toBeUndefined() + }) + + it('end() after CB:ib,,offline has already arrived is a no-op', () => { + startBuffer(state, mockFlush, mockWarn) + onOffline(state, mockFlush) + + // end() should not throw and should leave state clean + expect(() => onClose(state)).not.toThrow() + expect(state.offlineBufferTimeout).toBeUndefined() + expect(state.didStartBuffer).toBe(false) + }) + + // ------------------------------------------------------------------------- + // 4. Boundary / timing edge cases + // ------------------------------------------------------------------------- + + it('does not flush before exactly 5 s have elapsed', () => { + startBuffer(state, mockFlush, mockWarn) + + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS - 1) + + expect(mockFlush).not.toHaveBeenCalled() + }) + + it('flushes at exactly the 5 s boundary', () => { + startBuffer(state, mockFlush, mockWarn) + + jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) + + expect(mockFlush).toHaveBeenCalledTimes(1) + }) +})