ruicesar/InfiniteAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(inbound): avoid messageMutex deadlock in appStateSyncKeyShare branch

Browse Source 
Addresses Copilot review on PR #392.

The previous commit wrapped postUpsertWork in messageMutex.mutex(chatId, ...)
for both branches and then awaited it in the appStateSyncKeyShare path. This
deadlocks when upsertMessage is invoked from messages-recv.ts:2416, which is
already holding messageMutex.mutex(chatId, ...) for the same chat:

  outer mutex (in caller)  → await upsertMessage(...)
   └─ upsertMessage work() → schedules inner mutex on same chatId (queued)
       └─ await postUpsertWork  ← inner cannot acquire while outer holds it
                                   outer cannot release while it awaits inner
                                   = eternal deadlock (async-mutex is not re-entrant)

Fire-and-forget path is safe because it does NOT await: outer releases as
soon as upsertMessage returns, then the queued inner acquires.

Fix: in the appStateSyncKeyShare branch, run postUpsertTasks() inline (no
inner messageMutex). Per-chat ordering is still preserved by the caller's
outer mutex while we await. Other paths keep the inner mutex wrapper for
the latency win + per-chat ordering of detached work.
This commit is contained in:
Renato Alcara
2026-04-26 14:14:16 -03:00
parent 72e78a12e0
commit 1e37448e05
1 changed files with 29 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Socket/chats.ts
+29 -22
View File
@@ -1411,20 +1411,12 @@ export const makeChatsSocket = (config: SocketConfig) => {
}
}
// Fire-and-forget: post-upsert work (history app-state sync + processMessage)
// 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.
//
// 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, () =>
// Post-upsert work: history app-state sync + processMessage side effects.
// Awaiting here keeps `messages.upsert` pinned in the event buffer
// (createBufferedFunction only schedules flush after work() resolves), so
// the hot path detaches this work to release the emit on the next debounce
// tick.
const postUpsertTasks = () =>
Promise.all([
shouldProcessHistoryMsg ? doAppStateSync() : Promise.resolve(),
processMessage(msg, {
@@ -1439,16 +1431,24 @@ export const makeChatsSocket = (config: SocketConfig) => {
getMessage
})
])
)
// 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) {
const isKeyShareDuringSync =
!!msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing
if (isKeyShareDuringSync) {
// appStateSyncKeyShare path: processMessage persists the new app-state-sync
// key via keyStore.transaction at process-message.ts:650. The follow-up
// doAppStateSync() needs that key to decrypt patches, so we MUST await
// processMessage first.
//
// IMPORTANT: do NOT wrap in messageMutex here. The inbound caller in
// messages-recv.ts:2416 already holds messageMutex.mutex(chatId, ...) for
// this chat; re-acquiring the same keyed mutex while awaiting it would
// deadlock (async-mutex is not re-entrant). Running inline preserves
// per-chat ordering via the caller's outer mutex.
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
try {
await postUpsertWork
await postUpsertTasks()
await doAppStateSync()
} catch (err) {
logger?.warn(
@@ -1457,7 +1457,14 @@ export const makeChatsSocket = (config: SocketConfig) => {
)
}
} else {
postUpsertWork.catch(err =>
// Fire-and-forget path: wrap in inner messageMutex(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. Reentrancy is safe here because the outer mutex acquired by
// messages-recv.ts releases as soon as upsertMessage returns (work is not
// awaited), then this inner mutex enqueues per-chat.
const postUpsertChatId = msg.key.remoteJid || msg.key.id || 'unknown'
messageMutex.mutex(postUpsertChatId, postUpsertTasks).catch(err =>
logger?.warn(
{
err,