perf: eliminates a delay in message delivery between phone→app + cache device-list
perf: eliminates a delay in message delivery between phone→app + cache device-list
This commit is contained in:
@@ -62,10 +62,10 @@ export const DEFAULT_CONNECTION_CONFIG: SocketConfig = {
|
|||||||
browser: Browsers.macOS('Chrome'),
|
browser: Browsers.macOS('Chrome'),
|
||||||
waWebSocketUrl: 'wss://web.whatsapp.com/ws/chat',
|
waWebSocketUrl: 'wss://web.whatsapp.com/ws/chat',
|
||||||
connectTimeoutMs: 20_000,
|
connectTimeoutMs: 20_000,
|
||||||
keepAliveIntervalMs: 30_000,
|
keepAliveIntervalMs: 15_000,
|
||||||
logger: logger.child({ class: 'baileys' }),
|
logger: logger.child({ class: 'baileys' }),
|
||||||
emitOwnEvents: true,
|
emitOwnEvents: true,
|
||||||
defaultQueryTimeoutMs: 60_000,
|
defaultQueryTimeoutMs: 30_000,
|
||||||
customUploadHosts: [],
|
customUploadHosts: [],
|
||||||
retryRequestDelayMs: 250,
|
retryRequestDelayMs: 250,
|
||||||
maxMsgRetryCount: 5,
|
maxMsgRetryCount: 5,
|
||||||
|
|||||||
+25
-2
@@ -286,6 +286,17 @@ export function makeLibSignalRepository(
|
|||||||
updateAgeOnGet: true
|
updateAgeOnGet: true
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Cache for device-list DB reads in migrateSession.
|
||||||
|
// The device list changes rarely (new linked device), so a 5-minute TTL avoids
|
||||||
|
// a DB round-trip on every incoming LID message without risking stale state:
|
||||||
|
// new devices are picked up at most 5 minutes late (still migrated via their own
|
||||||
|
// decrypt transaction when the session is first used).
|
||||||
|
const deviceListCache = new LRUCache<string, string[]>({
|
||||||
|
max: 500,
|
||||||
|
ttl: 5 * 60 * 1000, // 5 minutes
|
||||||
|
ttlAutopurge: true
|
||||||
|
})
|
||||||
|
|
||||||
const repository: SignalRepositoryWithLIDStore = {
|
const repository: SignalRepositoryWithLIDStore = {
|
||||||
decryptGroupMessage({ group, authorJid, msg }) {
|
decryptGroupMessage({ group, authorJid, msg }) {
|
||||||
const senderName = jidToSignalSenderKeyName(group, authorJid)
|
const senderName = jidToSignalSenderKeyName(group, authorJid)
|
||||||
@@ -493,12 +504,24 @@ export function makeLibSignalRepository(
|
|||||||
|
|
||||||
logger.debug({ fromJid }, 'bulk device migration - loading all user devices')
|
logger.debug({ fromJid }, 'bulk device migration - loading all user devices')
|
||||||
|
|
||||||
// Get user's device list from storage
|
// Get user's device list — use in-memory cache to avoid a DB round-trip on
|
||||||
const { [user]: userDevices } = await parsedKeys.get('device-list', [user])
|
// every incoming LID message. Cache is invalidated after 5 minutes.
|
||||||
|
let userDevices: string[] | undefined = deviceListCache.get(user)
|
||||||
|
if (!userDevices) {
|
||||||
|
const result = await parsedKeys.get('device-list', [user])
|
||||||
|
userDevices = result[user]
|
||||||
|
if (userDevices) {
|
||||||
|
deviceListCache.set(user, userDevices)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (!userDevices) {
|
if (!userDevices) {
|
||||||
return { migrated: 0, skipped: 0, total: 0 }
|
return { migrated: 0, skipped: 0, total: 0 }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Work on a copy so we don't mutate the cached array
|
||||||
|
userDevices = [...userDevices]
|
||||||
|
|
||||||
const { device: fromDevice } = decoded1
|
const { device: fromDevice } = decoded1
|
||||||
const fromDeviceStr = fromDevice?.toString() || '0'
|
const fromDeviceStr = fromDevice?.toString() || '0'
|
||||||
if (!userDevices.includes(fromDeviceStr)) {
|
if (!userDevices.includes(fromDeviceStr)) {
|
||||||
|
|||||||
@@ -1330,6 +1330,7 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
if (syncState === SyncState.Syncing) {
|
if (syncState === SyncState.Syncing) {
|
||||||
// All collections will be synced, so clear any blocked ones
|
// All collections will be synced, so clear any blocked ones
|
||||||
blockedCollections.clear()
|
blockedCollections.clear()
|
||||||
|
|
||||||
logger.info('Doing app state sync')
|
logger.info('Doing app state sync')
|
||||||
await resyncAppState(ALL_WA_PATCH_NAMES, true)
|
await resyncAppState(ALL_WA_PATCH_NAMES, true)
|
||||||
|
|
||||||
@@ -1430,6 +1431,27 @@ 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
|
||||||
|
// session. Skip the 20s AwaitingInitialSync wait and go directly to Online so that
|
||||||
|
// live incoming messages are not held in the buffer for up to 20-60 seconds.
|
||||||
|
// History messages that arrive later are still processed via processMessage regardless
|
||||||
|
// of the state machine phase (see the Syncing → Online path below).
|
||||||
|
const isReconnection = (authState.creds.accountSyncCounter ?? 0) > 0
|
||||||
|
if (isReconnection) {
|
||||||
|
logger.info('Reconnection detected (accountSyncCounter > 0), skipping AwaitingInitialSync wait. Transitioning to Online immediately.')
|
||||||
|
blockedCollections.clear()
|
||||||
|
syncState = SyncState.Online
|
||||||
|
const accountSyncCounter = (authState.creds.accountSyncCounter || 0) + 1
|
||||||
|
ev.emit('creds.update', { accountSyncCounter })
|
||||||
|
// Fire-and-forget: pick up patches missed during downtime (mute/archive/pin/read state).
|
||||||
|
// Runs in background so live incoming messages are not blocked.
|
||||||
|
resyncAppState(ALL_WA_PATCH_NAMES, true).catch(err =>
|
||||||
|
logger.warn({ err }, 'Background app state resync failed (non-critical on reconnection)')
|
||||||
|
)
|
||||||
|
setTimeout(() => ev.flush(), 0)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
const willSyncHistory = shouldSyncHistoryMessage(
|
const willSyncHistory = shouldSyncHistoryMessage(
|
||||||
proto.Message.HistorySyncNotification.create({
|
proto.Message.HistorySyncNotification.create({
|
||||||
syncType: proto.HistorySync.HistorySyncType.RECENT
|
syncType: proto.HistorySync.HistorySyncType.RECENT
|
||||||
|
|||||||
@@ -54,9 +54,9 @@ export interface BufferConfig {
|
|||||||
*/
|
*/
|
||||||
export function loadBufferConfig(): BufferConfig {
|
export function loadBufferConfig(): BufferConfig {
|
||||||
return {
|
return {
|
||||||
bufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_TIMEOUT_MS || '30000', 10),
|
bufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_TIMEOUT_MS || '15000', 10),
|
||||||
minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '5000', 10),
|
minBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MIN_TIMEOUT_MS || '3000', 10),
|
||||||
maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '60000', 10),
|
maxBufferTimeoutMs: parseInt(process.env.BAILEYS_BUFFER_MAX_TIMEOUT_MS || '20000', 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 || '100', 10),
|
||||||
|
|||||||
Reference in New Issue
Block a user