Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| cfa74127a1 | |||
| ce38e7836d |
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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 })
|
||||||
|
|||||||
@@ -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(() => {
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
Reference in New Issue
Block a user