diff --git a/src/Defaults/index.ts b/src/Defaults/index.ts index 0c49914c..192b24b1 100644 --- a/src/Defaults/index.ts +++ b/src/Defaults/index.ts @@ -67,7 +67,7 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { emitOwnEvents: true, defaultQueryTimeoutMs: 30_000, customUploadHosts: [], - retryRequestDelayMs: 250, + retryRequestDelayMs: 150, maxMsgRetryCount: 5, fireInitQueries: true, auth: undefined as unknown as AuthenticationState, @@ -78,7 +78,7 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = { shouldSyncHistoryMessage: () => true, shouldIgnoreJid: () => false, linkPreviewImageThumbnailWidth: 192, - transactionOpts: { maxCommitRetries: 10, delayBetweenTriesMs: 3000 }, + transactionOpts: { maxCommitRetries: 10, delayBetweenTriesMs: 1000 }, generateHighQualityLinkPreview: false, enableAutoSessionRecreation: true, enableRecentMessageCache: true, diff --git a/src/Socket/chats.ts b/src/Socket/chats.ts index 72b951c5..94e64830 100644 --- a/src/Socket/chats.ts +++ b/src/Socket/chats.ts @@ -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 */ const messageMutex = makeKeyedMutex() - /** this mutex ensures that receipts are processed in order */ - const receiptMutex = makeMutex() + /** this mutex ensures that receipts from the same chat are processed in order, while allowing parallel processing across chats */ + const receiptMutex = makeKeyedMutex() /** this mutex ensures that app state patches are processed in order */ const appStatePatchMutex = makeMutex() - /** this mutex ensures that notifications are processed in order */ - const notificationMutex = makeMutex() + /** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */ + const notificationMutex = makeKeyedMutex() // Timeout for AwaitingInitialSync state let awaitingSyncTimeout: NodeJS.Timeout | undefined @@ -1495,7 +1495,7 @@ export const makeChatsSocket = (config: SocketConfig) => { awaitingSyncTimeout = setTimeout(() => { 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 ev.flush() @@ -1505,7 +1505,7 @@ export const makeChatsSocket = (config: SocketConfig) => { const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1 ev.emit('creds.update', { accountSyncCounter }) } - }, 4_000) + }, 2_000) }) // When an app state sync key arrives (myAppStateKeyId is set) and there are diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index 6ff06546..85f51099 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -1282,7 +1282,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { try { await Promise.all([ - receiptMutex.mutex(async () => { + receiptMutex.mutex(jidNormalizedUser(remoteJid) || 'unknown', async () => { const status = getStatusFromReceiptType(attrs.type) if ( typeof status !== 'undefined' && @@ -1356,7 +1356,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { try { await Promise.all([ - notificationMutex.mutex(async () => { + notificationMutex.mutex(jidNormalizedUser(remoteJid) || 'unknown', async () => { const msg = await processNotification(node) if (msg) { const fromMe = areJidsSameUser(node.attrs.participant || remoteJid, authState.creds.me!.id) @@ -1985,7 +1985,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { let isProcessing = false // Number of nodes to process before yielding to event loop - const BATCH_SIZE = 10 + const BATCH_SIZE = 25 const enqueue = (type: MessageType, node: BinaryNode) => { nodes.push({ type, node }) diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 3f394e51..aa26ec98 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -1667,8 +1667,8 @@ export const makeSocket = (config: SocketConfig) => { // 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 // server to finish flushing the pending-message backlog (CB:ib,,offline). - // For normal restarts (no stale routingInfo) the standard 5 s safety cap applies. - const OFFLINE_BUFFER_TIMEOUT_MS = 5_000 + // For normal restarts (no stale routingInfo) the standard 2 s safety cap applies. + const OFFLINE_BUFFER_TIMEOUT_MS = 2_000 let offlineBufferTimeout: NodeJS.Timeout | undefined process.nextTick(() => { diff --git a/src/Utils/event-buffer.ts b/src/Utils/event-buffer.ts index 24bf36bb..1b19a3e8 100644 --- a/src/Utils/event-buffer.ts +++ b/src/Utils/event-buffer.ts @@ -54,15 +54,15 @@ export interface BufferConfig { */ export function loadBufferConfig(): BufferConfig { 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), // 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), - 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), 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', enableMetrics: process.env.BAILEYS_BUFFER_METRICS === 'true' || process.env.BAILEYS_PROMETHEUS_ENABLED === 'true', lruCleanupRatio: parseFloat(process.env.BAILEYS_BUFFER_LRU_CLEANUP_RATIO || '0.2'), diff --git a/src/__tests__/Socket/offline-buffer-timeout.test.ts b/src/__tests__/Socket/offline-buffer-timeout.test.ts index 6ad2a05c..c138b2dd 100644 --- a/src/__tests__/Socket/offline-buffer-timeout.test.ts +++ b/src/__tests__/Socket/offline-buffer-timeout.test.ts @@ -20,7 +20,7 @@ import { jest } from '@jest/globals' * 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 */ 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) 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', () => { startBuffer(state, mockFlush, mockWarn) - // Server responds before the 5 s timeout + // Server responds before the 2 s timeout jest.advanceTimersByTime(1_000) 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) expect(mockFlush).toHaveBeenCalledTimes(1) @@ -192,7 +192,7 @@ describe('offline-buffer safety timer (socket.ts)', () => { 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) expect(mockFlush).not.toHaveBeenCalled() @@ -236,7 +236,7 @@ describe('offline-buffer safety timer (socket.ts)', () => { // 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) jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS - 1) @@ -244,7 +244,7 @@ describe('offline-buffer safety timer (socket.ts)', () => { expect(mockFlush).not.toHaveBeenCalled() }) - it('flushes at exactly the 5 s boundary', () => { + it('flushes at exactly the 2 s boundary', () => { startBuffer(state, mockFlush, mockWarn) jest.advanceTimersByTime(OFFLINE_BUFFER_TIMEOUT_MS)