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>
This commit is contained in:
@@ -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,
|
||||
|
||||
+6
-6
@@ -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
|
||||
|
||||
@@ -1282,7 +1282,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
|
||||
try {
|
||||
await Promise.all([
|
||||
receiptMutex.mutex(async () => {
|
||||
receiptMutex.mutex(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(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 })
|
||||
|
||||
@@ -1668,7 +1668,7 @@ export const makeSocket = (config: SocketConfig) => {
|
||||
// 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
|
||||
const OFFLINE_BUFFER_TIMEOUT_MS = 2_000
|
||||
let offlineBufferTimeout: NodeJS.Timeout | undefined
|
||||
|
||||
process.nextTick(() => {
|
||||
|
||||
@@ -57,12 +57,12 @@ export function loadBufferConfig(): BufferConfig {
|
||||
// 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),
|
||||
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'),
|
||||
|
||||
Reference in New Issue
Block a user