From 72e78a12e0ef6b7ea2cda22011b04912940c5278 Mon Sep 17 00:00:00 2001 From: Renato Alcara Date: Sun, 26 Apr 2026 13:47:14 -0300 Subject: [PATCH] fix(inbound): preserve per-chat ordering + key-share await in detached upsert MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses bot review on PR #392: 1. Codex P1 + CodeRabbit Major: removing the `await` on Promise.all broke the per-chat ordering guarantee enforced by `messageMutex` in messages-recv.ts:2416 (`await messageMutex.mutex(... await upsertMessage(...))`). The outer mutex unlocked before the detached processMessage finished, so a follow-up message from the same chat could overtake state mutations (chat.unreadCount, LID/PN mapping, messages.update emits, history downloads). Fix: wrap the fire-and-forget Promise.all in `messageMutex.mutex(chatId, ...)` inside chats.ts. The outer mutex (in messages-recv) releases as soon as upsertMessage returns (fast — work is detached), then this inner mutex enqueues per-chat. Reentrancy is safe because outer always releases before inner acquires. 2. CodeRabbit Critical: appStateSyncKeyShare race. processMessage persists the new app-state-sync key via keyStore.transaction at process-message.ts:650. Previously the awaited Promise.all guaranteed the key was in the store before the follow-up `await doAppStateSync()` (which decodes patches via that key) ran. After detaching, doAppStateSync could hit isMissingKeyError and park collections in blockedCollections, requiring an extra retry cycle. Fix: for the rare appStateSyncKeyShare + Syncing branch, await the postUpsertWork before triggering doAppStateSync. All other messages stay fire-and-forget — the latency win is preserved for the hot path. 3. Copilot nit: enrich the warn payload with messageId, remoteJid and shouldProcessHistoryMsg so background failures can be correlated. 4. CodeRabbit nit: replace the IIFE around the conditional doAppStateSync with a ternary expression. --- src/Socket/chats.ts | 74 +++++++++++++++++++++++++++++++-------------- 1 file changed, 51 insertions(+), 23 deletions(-) diff --git a/src/Socket/chats.ts b/src/Socket/chats.ts index 3d1152e7..8673346a 100644 --- a/src/Socket/chats.ts +++ b/src/Socket/chats.ts @@ -1415,31 +1415,59 @@ export const makeChatsSocket = (config: SocketConfig) => { // must NOT block the buffered function. Awaiting here keeps `messages.upsert` // pinned in the event buffer (createBufferedFunction only schedules flush // after work() resolves), delaying delivery to the consumer by the duration - // of processMessage. Detaching this work releases the emit on the next debounce - // tick while signal/store/tcToken side-effects continue in the background. - Promise.all([ - (async () => { - if (shouldProcessHistoryMsg) { - await doAppStateSync() - } - })(), - processMessage(msg, { - signalRepository, - shouldProcessHistoryMsg, - placeholderResendCache, - ev, - creds: authState.creds, - keyStore: authState.keys, - logger, - options: config.options, - getMessage - }) - ]).catch(err => logger?.warn({ err }, 'background post-upsert work failed')) + // of processMessage. + // + // We wrap the work in `messageMutex.mutex(chatId, ...)` so per-chat ordering + // of processMessage side-effects (chat.unreadCount, LID/PN mapping, messages.update, + // history downloads) is preserved across messages of the same chat. The outer + // mutex acquired by messages-recv.ts releases as soon as upsertMessage returns + // (fast, since work is detached), then this inner mutex enqueues per-chat — + // no deadlock, strict per-chat ordering kept. + const postUpsertChatId = msg.key.remoteJid || msg.key.id || 'unknown' + const postUpsertWork = messageMutex.mutex(postUpsertChatId, () => + Promise.all([ + shouldProcessHistoryMsg ? doAppStateSync() : Promise.resolve(), + processMessage(msg, { + signalRepository, + shouldProcessHistoryMsg, + placeholderResendCache, + ev, + creds: authState.creds, + keyStore: authState.keys, + logger, + options: config.options, + getMessage + }) + ]) + ) - // If the app state key arrives and we are waiting to sync, trigger the sync now. + // appStateSyncKeyShare path: processMessage persists the new app-state-sync key + // via keyStore.transaction. The follow-up doAppStateSync() needs that key to + // decrypt patches, so for this rare branch we MUST await postUpsertWork first. + // All other messages stay fire-and-forget (the latency win). if (msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing) { - logger.info('App state sync key arrived, triggering app state sync') - await doAppStateSync() + logger.info('App state sync key arrived, awaiting persistence before triggering sync') + try { + await postUpsertWork + await doAppStateSync() + } catch (err) { + logger?.warn( + { err, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid }, + 'app-state key-share post-upsert work failed' + ) + } + } else { + postUpsertWork.catch(err => + logger?.warn( + { + err, + messageId: msg.key?.id, + remoteJid: msg.key?.remoteJid, + shouldProcessHistoryMsg + }, + 'background post-upsert work failed' + ) + ) } })