fix(inbound): preserve per-chat ordering + key-share await in detached upsert
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.
This commit is contained in:
+51
-23
@@ -1415,31 +1415,59 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
// must NOT block the buffered function. Awaiting here keeps `messages.upsert`
|
// must NOT block the buffered function. Awaiting here keeps `messages.upsert`
|
||||||
// pinned in the event buffer (createBufferedFunction only schedules flush
|
// pinned in the event buffer (createBufferedFunction only schedules flush
|
||||||
// after work() resolves), delaying delivery to the consumer by the duration
|
// after work() resolves), delaying delivery to the consumer by the duration
|
||||||
// of processMessage. Detaching this work releases the emit on the next debounce
|
// of processMessage.
|
||||||
// tick while signal/store/tcToken side-effects continue in the background.
|
//
|
||||||
Promise.all([
|
// We wrap the work in `messageMutex.mutex(chatId, ...)` so per-chat ordering
|
||||||
(async () => {
|
// of processMessage side-effects (chat.unreadCount, LID/PN mapping, messages.update,
|
||||||
if (shouldProcessHistoryMsg) {
|
// history downloads) is preserved across messages of the same chat. The outer
|
||||||
await doAppStateSync()
|
// 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.
|
||||||
processMessage(msg, {
|
const postUpsertChatId = msg.key.remoteJid || msg.key.id || 'unknown'
|
||||||
signalRepository,
|
const postUpsertWork = messageMutex.mutex(postUpsertChatId, () =>
|
||||||
shouldProcessHistoryMsg,
|
Promise.all([
|
||||||
placeholderResendCache,
|
shouldProcessHistoryMsg ? doAppStateSync() : Promise.resolve(),
|
||||||
ev,
|
processMessage(msg, {
|
||||||
creds: authState.creds,
|
signalRepository,
|
||||||
keyStore: authState.keys,
|
shouldProcessHistoryMsg,
|
||||||
logger,
|
placeholderResendCache,
|
||||||
options: config.options,
|
ev,
|
||||||
getMessage
|
creds: authState.creds,
|
||||||
})
|
keyStore: authState.keys,
|
||||||
]).catch(err => logger?.warn({ err }, 'background post-upsert work failed'))
|
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) {
|
if (msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing) {
|
||||||
logger.info('App state sync key arrived, triggering app state sync')
|
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
|
||||||
await doAppStateSync()
|
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'
|
||||||
|
)
|
||||||
|
)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user