Compare commits

...

2 Commits

Author SHA1 Message Date
Renato Alcara cfa74127a1 fix: address PR review — stale comments, mutex key normalization, test sync
- Update stale comments in event-buffer.ts (15s→3s) and socket.ts (5s→2s)
  to reflect actual timeout values
- Normalize mutex keys with jidNormalizedUser() for receiptMutex and
  notificationMutex to handle device suffix variants (e.g. user:5@s.whatsapp.net)
- Update offline-buffer-timeout.test.ts to match new 2s timeout value
  (was hardcoded at 5s, causing test/source divergence)
- Rejected suggestion to use attrs.id as fallback (false positive — attrs.id
  is a message ID, not a chat identifier; 'unknown' is the correct catch-all)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-28 17:21:11 -03:00
Renato Alcara ce38e7836d perf: reduce message delivery latency across all connection scenarios
Surgical optimizations to eliminate message delivery delays after QR scan,
pm2 restart, and code updates:

Buffer/Flush:
- bufferTimeoutMs: 5s→3s (safety auto-flush cap)
- maxBufferTimeoutMs: 8s→3s (adaptive timeout ceiling)
- flushDebounceMs: 100ms→10ms (createBufferedFunction debounce)
- OFFLINE_BUFFER_TIMEOUT_MS: 5s→2s (first-connection buffer)
- AwaitingInitialSync: 4s→2s (history sync wait)

Mutex parallelism:
- receiptMutex: global→keyed by remoteJid (parallel receipt processing)
- notificationMutex: global→keyed by remoteJid (parallel notification processing)

Delay reduction:
- retryRequestDelayMs: 250ms→150ms
- transactionOpts.delayBetweenTriesMs: 3s→1s

Offline processing:
- BATCH_SIZE: 10→25 (fewer event loop yields)

Expected impact: QR scan worst-case 8s→2s, pm2 restart 100ms→10ms.
All existing customizations preserved (Circuit Breaker, Prometheus,
TcToken, CTWA Recovery, sanitizeCallerPn, Session Cleanup, LID Mapping,
Identity Key cache, clearRoutingInfoOnStart, wasm-bridge).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-28 17:07:13 -03:00
6 changed files with 26 additions and 26 deletions
+2 -2
View File
@@ -67,7 +67,7 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
emitOwnEvents: true, emitOwnEvents: true,
defaultQueryTimeoutMs: 30_000, defaultQueryTimeoutMs: 30_000,
customUploadHosts: [], customUploadHosts: [],
retryRequestDelayMs: 250, retryRequestDelayMs: 150,
maxMsgRetryCount: 5, maxMsgRetryCount: 5,
fireInitQueries: true, fireInitQueries: true,
auth: undefined as unknown as AuthenticationState, auth: undefined as unknown as AuthenticationState,
@@ -78,7 +78,7 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
shouldSyncHistoryMessage: () => true, shouldSyncHistoryMessage: () => true,
shouldIgnoreJid: () => false, shouldIgnoreJid: () => false,
linkPreviewImageThumbnailWidth: 192, linkPreviewImageThumbnailWidth: 192,
transactionOpts: { maxCommitRetries: 10, delayBetweenTriesMs: 3000 }, transactionOpts: { maxCommitRetries: 10, delayBetweenTriesMs: 1000 },
generateHighQualityLinkPreview: false, generateHighQualityLinkPreview: false,
enableAutoSessionRecreation: true, enableAutoSessionRecreation: true,
enableRecentMessageCache: true, enableRecentMessageCache: true,
+6 -6
View File
@@ -102,14 +102,14 @@ export const makeChatsSocket = (config: SocketConfig) => {
/** this mutex ensures that messages from the same chat are processed in order, while allowing parallel processing of messages from different chats */ /** this mutex ensures that messages from the same chat are processed in order, while allowing parallel processing of messages from different chats */
const messageMutex = makeKeyedMutex() const messageMutex = makeKeyedMutex()
/** this mutex ensures that receipts are processed in order */ /** this mutex ensures that receipts from the same chat are processed in order, while allowing parallel processing across chats */
const receiptMutex = makeMutex() const receiptMutex = makeKeyedMutex()
/** this mutex ensures that app state patches are processed in order */ /** this mutex ensures that app state patches are processed in order */
const appStatePatchMutex = makeMutex() const appStatePatchMutex = makeMutex()
/** this mutex ensures that notifications are processed in order */ /** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */
const notificationMutex = makeMutex() const notificationMutex = makeKeyedMutex()
// Timeout for AwaitingInitialSync state // Timeout for AwaitingInitialSync state
let awaitingSyncTimeout: NodeJS.Timeout | undefined let awaitingSyncTimeout: NodeJS.Timeout | undefined
@@ -1495,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 (4s), forcing state to Online and flushing buffer') logger.warn('Timeout in AwaitingInitialSync (2s), forcing state to Online and flushing buffer')
syncState = SyncState.Online syncState = SyncState.Online
ev.flush() ev.flush()
@@ -1505,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 })
} }
}, 4_000) }, 2_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
+3 -3
View File
@@ -1282,7 +1282,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
try { try {
await Promise.all([ await Promise.all([
receiptMutex.mutex(async () => { receiptMutex.mutex(jidNormalizedUser(remoteJid) || 'unknown', async () => {
const status = getStatusFromReceiptType(attrs.type) const status = getStatusFromReceiptType(attrs.type)
if ( if (
typeof status !== 'undefined' && typeof status !== 'undefined' &&
@@ -1356,7 +1356,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
try { try {
await Promise.all([ await Promise.all([
notificationMutex.mutex(async () => { notificationMutex.mutex(jidNormalizedUser(remoteJid) || 'unknown', async () => {
const msg = await processNotification(node) const msg = await processNotification(node)
if (msg) { if (msg) {
const fromMe = areJidsSameUser(node.attrs.participant || remoteJid, authState.creds.me!.id) const fromMe = areJidsSameUser(node.attrs.participant || remoteJid, authState.creds.me!.id)
@@ -1985,7 +1985,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
let isProcessing = false let isProcessing = false
// Number of nodes to process before yielding to event loop // Number of nodes to process before yielding to event loop
const BATCH_SIZE = 10 const BATCH_SIZE = 25
const enqueue = (type: MessageType, node: BinaryNode) => { const enqueue = (type: MessageType, node: BinaryNode) => {
nodes.push({ type, node }) nodes.push({ type, node })
+2 -2
View File
@@ -1667,8 +1667,8 @@ export const makeSocket = (config: SocketConfig) => {
// already-authenticated session (no QR re-scan needed). In this case we skip // already-authenticated session (no QR re-scan needed). In this case we skip
// the offline buffer entirely so live messages are not held hostage waiting for the // the offline buffer entirely so live messages are not held hostage waiting for the
// server to finish flushing the pending-message backlog (CB:ib,,offline). // server to finish flushing the pending-message backlog (CB:ib,,offline).
// For normal restarts (no stale routingInfo) the standard 5 s safety cap applies. // For normal restarts (no stale routingInfo) the standard 2 s safety cap applies.
const OFFLINE_BUFFER_TIMEOUT_MS = 5_000 const OFFLINE_BUFFER_TIMEOUT_MS = 2_000
let offlineBufferTimeout: NodeJS.Timeout | undefined let offlineBufferTimeout: NodeJS.Timeout | undefined
process.nextTick(() => { process.nextTick(() => {
+4 -4
View File
@@ -54,15 +54,15 @@ export interface BufferConfig {
*/ */
export function loadBufferConfig(): BufferConfig { export function loadBufferConfig(): BufferConfig {
return { return {
// perf(inbound-latency): reduced from 15s → 5s so the safety auto-flush fires sooner. // perf(inbound-latency): reduced from 15s → 3s so the safety auto-flush fires sooner.
// processNodeWithBuffer always calls ev.flush() explicitly (no-op for this timer), // 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. // 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), bufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_TIMEOUT_MS || '3000', 10),
minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '1000', 10), minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '1000', 10),
maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '8000', 10), maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '3000', 10),
maxHistoryCacheSize: parseInt(process.env.BAILEYS_BUFFER_MAX_HISTORY_CACHE || '10000', 10), maxHistoryCacheSize: parseInt(process.env.BAILEYS_BUFFER_MAX_HISTORY_CACHE || '10000', 10),
maxBufferSize: parseInt(process.env.BAILEYS_BUFFER_MAX_SIZE || '5000', 10), maxBufferSize: parseInt(process.env.BAILEYS_BUFFER_MAX_SIZE || '5000', 10),
flushDebounceMs: parseInt(process.env.BAILEYS_BUFFER_FLUSH_DEBOUNCE_MS || '100', 10), flushDebounceMs: parseInt(process.env.BAILEYS_BUFFER_FLUSH_DEBOUNCE_MS || '10', 10),
enableAdaptiveTimeout: process.env.BAILEYS_BUFFER_ADAPTIVE_TIMEOUT !== 'false', enableAdaptiveTimeout: process.env.BAILEYS_BUFFER_ADAPTIVE_TIMEOUT !== 'false',
enableMetrics: process.env.BAILEYS_BUFFER_METRICS === 'true' || process.env.BAILEYS_PROMETHEUS_ENABLED === 'true', enableMetrics: process.env.BAILEYS_BUFFER_METRICS === 'true' || process.env.BAILEYS_PROMETHEUS_ENABLED === 'true',
lruCleanupRatio: parseFloat(process.env.BAILEYS_BUFFER_LRU_CLEANUP_RATIO || '0.2'), lruCleanupRatio: parseFloat(process.env.BAILEYS_BUFFER_LRU_CLEANUP_RATIO || '0.2'),
@@ -20,7 +20,7 @@ import { jest } from '@jest/globals'
* bad-ack-handling.test.ts. * bad-ack-handling.test.ts.
*/ */
const OFFLINE_BUFFER_TIMEOUT_MS = 5_000 const OFFLINE_BUFFER_TIMEOUT_MS = 2_000
/** Mirrors the state variables declared at the top of makeSocket */ /** Mirrors the state variables declared at the top of makeSocket */
interface OfflineBufferState { interface OfflineBufferState {
@@ -101,10 +101,10 @@ describe('offline-buffer safety timer (socket.ts)', () => {
}) })
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
// 1. Timeout path — CB:ib,,offline never arrives within 5 s // 1. Timeout path — CB:ib,,offline never arrives within 2 s
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
it('fires after 5 s and flushes when CB:ib,,offline is delayed', () => { it('fires after 2 s and flushes when CB:ib,,offline is delayed', () => {
startBuffer(state, mockFlush, mockWarn) startBuffer(state, mockFlush, mockWarn)
expect(mockFlush).not.toHaveBeenCalled() expect(mockFlush).not.toHaveBeenCalled()
@@ -144,17 +144,17 @@ describe('offline-buffer safety timer (socket.ts)', () => {
}) })
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
// 2. Happy path — CB:ib,,offline arrives before the 5 s timer fires // 2. Happy path — CB:ib,,offline arrives before the 2 s timer fires
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
it('CB:ib,,offline cancels the timer and flushes exactly once', () => { it('CB:ib,,offline cancels the timer and flushes exactly once', () => {
startBuffer(state, mockFlush, mockWarn) startBuffer(state, mockFlush, mockWarn)
// Server responds before the 5 s timeout // Server responds before the 2 s timeout
jest.advanceTimersByTime(1_000) jest.advanceTimersByTime(1_000)
onOffline(state, mockFlush) onOffline(state, mockFlush)
// Timer should be cancelled — advancing past 5 s must not cause a second flush // Timer should be cancelled — advancing past 2 s must not cause a second flush
jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS)
expect(mockFlush).toHaveBeenCalledTimes(1) expect(mockFlush).toHaveBeenCalledTimes(1)
@@ -192,7 +192,7 @@ describe('offline-buffer safety timer (socket.ts)', () => {
onClose(state) onClose(state)
// Timer must be gone — advancing past 5 s must not trigger any flush // Timer must be gone — advancing past 2 s must not trigger any flush
jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS)
expect(mockFlush).not.toHaveBeenCalled() expect(mockFlush).not.toHaveBeenCalled()
@@ -236,7 +236,7 @@ describe('offline-buffer safety timer (socket.ts)', () => {
// 4. Boundary / timing edge cases // 4. Boundary / timing edge cases
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------
it('does not flush before exactly 5 s have elapsed', () => { it('does not flush before exactly 2 s have elapsed', () => {
startBuffer(state, mockFlush, mockWarn) startBuffer(state, mockFlush, mockWarn)
jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS - 1) jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS - 1)
@@ -244,7 +244,7 @@ describe('offline-buffer safety timer (socket.ts)', () => {
expect(mockFlush).not.toHaveBeenCalled() expect(mockFlush).not.toHaveBeenCalled()
}) })
it('flushes at exactly the 5 s boundary', () => { it('flushes at exactly the 2 s boundary', () => {
startBuffer(state, mockFlush, mockWarn) startBuffer(state, mockFlush, mockWarn)
jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS) jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS)