perf(inbound-latency): fix 35-60 s message delivery delay by tightening buffer timeouts

perf(inbound-latency): fix 35-60 s message delivery delay by tightening buffer timeouts

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 (8s), forcing state to Online and flushing buffer')
-				logger.warn('Timeout in AwaitingInitialSync, 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

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')
 		}
 

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),
+		// perf(inbound-latency): reduced from 15s → 5s so the safety auto-flush fires sooner.
-		minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '3000', 10),
+		// processNodeWithBuffer always calls ev.flush() explicitly (no-op for this timer),
-		maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '20000', 10),
+		// 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),

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)
+	})
+})