fix(latency): resolve message delivery slowness on restart and fresh

fix(latency): resolve message delivery slowness on restart and fresh
This commit is contained in:
Renato Alcara
2026-02-27 11:57:16 -03:00
committed by GitHub
parent d233a7856f
commit 2faa107549
3 changed files with 52 additions and 16 deletions
+25 -15
View File
@@ -89,7 +89,8 @@ export const makeChatsSocket = (config: SocketConfig) => {
query, query,
signalRepository, signalRepository,
onUnexpectedError, onUnexpectedError,
sendUnifiedSession sendUnifiedSession,
skipOfflineBuffer: socketSkippedOfflineBuffer
} = sock } = sock
const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping) const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
@@ -1436,15 +1437,21 @@ export const makeChatsSocket = (config: SocketConfig) => {
logger.info('Connection is now AwaitingInitialSync, buffering events') logger.info('Connection is now AwaitingInitialSync, buffering events')
ev.buffer() ev.buffer()
// On reconnections (accountSyncCounter > 0), app state was already synced in a previous // On reconnections, app state was already synced in a previous session.
// session. Skip the 20s AwaitingInitialSync wait and go directly to Online so that // Skip the AwaitingInitialSync wait and go directly to Online so that
// live incoming messages are not held in the buffer for up to 20-60 seconds. // live incoming messages are not held in the buffer for up to 4 seconds.
// History messages that arrive later are still processed via processMessage regardless //
// of the state machine phase (see the Syncing → Online path below). // Two signals indicate a reconnect (either is sufficient):
const isReconnection = (authState.creds.accountSyncCounter ?? 0) > 0 // 1. accountSyncCounter > 0 — at least one full sync completed before
// 2. socketSkippedOfflineBuffer — socket.ts already determined this is a
// reconnect (e.g. stale routingInfo was cleared) and skipped the offline
// phase buffer. Keeping the second buffer active while the first was already
// skipped would cause a mismatch: events flow immediately then stall for 4s.
const isReconnection = (authState.creds.accountSyncCounter ?? 0) > 0 || socketSkippedOfflineBuffer
if (isReconnection) { if (isReconnection) {
logger.info( logger.info(
'Reconnection detected (accountSyncCounter > 0), skipping AwaitingInitialSync wait. Transitioning to Online immediately.' { accountSyncCounter: authState.creds.accountSyncCounter, socketSkippedOfflineBuffer },
'Reconnection detected, skipping AwaitingInitialSync wait. Transitioning to Online immediately.'
) )
blockedCollections.clear() blockedCollections.clear()
syncState = SyncState.Online syncState = SyncState.Online
@@ -1472,12 +1479,15 @@ export const makeChatsSocket = (config: SocketConfig) => {
return return
} }
// perf(inbound-latency): reduced from 20s → 8s. On first connection we wait for the // perf(inbound-latency): reduced from 20s → 8s → 4s. On first connection we wait for
// history-sync notification so that doAppStateSync runs before live messages are emitted. // the history-sync notification so that doAppStateSync runs before live messages are
// If the notification does not arrive within 8s we stop waiting, go Online, and flush // emitted. If the notification does not arrive within 4s we stop waiting, go Online,
// so that any live message arriving after connection is never held more than ~8s. // and flush so that any live message arriving after connection is never held more than 4s.
// History that arrives late is still processed via processMessage regardless of state. // History that arrives late is still processed via processMessage regardless of state.
logger.info('First connection, awaiting history sync notification with an 8s timeout.') // This 4s timeout fires before the event-buffer's own adaptive safety timer
// (BAILEYS_BUFFER_TIMEOUT_MS defaults to 5s), ensuring the buffer cannot stall
// beyond 4s on a first connect regardless of event rate.
logger.info('First connection, awaiting history sync notification with a 4s timeout.')
if (awaitingSyncTimeout) { if (awaitingSyncTimeout) {
clearTimeout(awaitingSyncTimeout) clearTimeout(awaitingSyncTimeout)
@@ -1485,7 +1495,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
awaitingSyncTimeout = setTimeout(() => { awaitingSyncTimeout = setTimeout(() => {
if (syncState === SyncState.AwaitingInitialSync) { if (syncState === SyncState.AwaitingInitialSync) {
logger.warn('Timeout in AwaitingInitialSync (8s), forcing state to Online and flushing buffer') logger.warn('Timeout in AwaitingInitialSync (4s), forcing state to Online and flushing buffer')
syncState = SyncState.Online syncState = SyncState.Online
ev.flush() ev.flush()
@@ -1495,7 +1505,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1 const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1
ev.emit('creds.update', { accountSyncCounter }) ev.emit('creds.update', { accountSyncCounter })
} }
}, 8_000) }, 4_000)
}) })
// When an app state sync key arrives (myAppStateKeyId is set) and there are // When an app state sync key arrives (myAppStateKeyId is set) and there are
+7 -1
View File
@@ -1797,7 +1797,13 @@ export const makeSocket = (config: SocketConfig) => {
/** Update server time offset (call when receiving server timestamps) */ /** Update server time offset (call when receiving server timestamps) */
updateServerTimeOffset: (serverTime: string | number) => { updateServerTimeOffset: (serverTime: string | number) => {
unifiedSessionManager?.updateServerTimeOffset(serverTime) unifiedSessionManager?.updateServerTimeOffset(serverTime)
} },
/**
* Whether the offline-phase buffer was skipped for this connection.
* true → this is a reconnect of an existing session (skip all sync waits in chats.ts too)
* false → fresh QR-scan or first connection (normal sync flow applies)
*/
skipOfflineBuffer
} }
} }
+20
View File
@@ -709,6 +709,16 @@ export function createConnectionCircuitBreaker(customOptions?: Partial<CircuitBr
'network' 'network'
] ]
// Normal WS lifecycle status codes that must NEVER trip the query circuit breaker.
// These are transient events that happen on every reconnect and carry no information
// about persistent server-side failures. The reconnection logic in makeSocket handles
// them independently.
const WS_LIFECYCLE_STATUS_CODES = new Set([
428, // connectionClosed
408, // connectionLost / timedOut
440, // connectionReplaced
])
return new CircuitBreaker({ return new CircuitBreaker({
name: 'connection-operations', name: 'connection-operations',
failureThreshold: 3, failureThreshold: 3,
@@ -720,6 +730,16 @@ export function createConnectionCircuitBreaker(customOptions?: Partial<CircuitBr
// accidentally matches the "socket" pattern below and causes a self-reinforcing loop // accidentally matches the "socket" pattern below and causes a self-reinforcing loop
// where the breaker's own timeouts keep tripping the breaker. Exclude them explicitly. // where the breaker's own timeouts keep tripping the breaker. Exclude them explicitly.
if (error instanceof CircuitTimeoutError) return false if (error instanceof CircuitTimeoutError) return false
// Exclude normal WS reconnect events (Connection Closed, Connection Lost, Timed Out, etc.).
// These are not server-side failures — they happen on every restart/redeploy and should
// never cause the circuit to open. If we counted them, the 3 concurrent parallel
// post-login queries (sendPassiveIq, uploadPreKeysToServerIfRequired, digestKeyBundle) could all
// fail simultaneously when the WS drops, instantly opening the circuit and blocking
// profile-picture fetches and message delivery for the next 30 s.
const statusCode = (error as { output?: { statusCode?: number } })?.output?.statusCode
if (statusCode !== undefined && WS_LIFECYCLE_STATUS_CODES.has(statusCode)) return false
const message = error.message.toLowerCase() const message = error.message.toLowerCase()
const code = (error as NodeJS.ErrnoException).code?.toLowerCase() || '' const code = (error as NodeJS.ErrnoException).code?.toLowerCase() || ''
return connectionErrorPatterns.some(pattern => message.includes(pattern) || code.includes(pattern)) return connectionErrorPatterns.some(pattern => message.includes(pattern) || code.includes(pattern))