fix(inbound): align mutex key with processMessage chat scope
Addresses CodeRabbit Minor + Copilot review on PR #392: 1. Replace `msg.key.remoteJid || msg.key.id || 'unknown'` with `getChatId(msg.key) + jidNormalizedUser`. The previous derivation: - Diverged from processMessage's chat scoping (which calls `jidNormalizedUser(getChatId(message.key))`). For non-status broadcasts processMessage targets `participant`, not `remoteJid`, so the inner mutex would queue under a different key than the chat processMessage actually mutates — defeating per-chat ordering for broadcast traffic. - Used `msg.key.id` as a fallback. Per-message ids are unique, so malformed-key messages each got their own queue, defeating the mutex entirely. Replaced with the constant `'unknown'` bucket. 2. Drop hard-coded line numbers (`process-message.ts:650`, `messages-recv.ts:2416`) from comments — references rot. Comments now point at the function/handler instead (`APP_STATE_SYNC_KEY_SHARE handler`, `inbound caller`).
This commit is contained in:
+17
-11
@@ -54,7 +54,7 @@ import {
|
|||||||
resolveLidToPn
|
resolveLidToPn
|
||||||
} from '../Utils'
|
} from '../Utils'
|
||||||
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
|
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
|
||||||
import processMessage from '../Utils/process-message'
|
import processMessage, { getChatId } from '../Utils/process-message'
|
||||||
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
||||||
import {
|
import {
|
||||||
type BinaryNode,
|
type BinaryNode,
|
||||||
@@ -1437,15 +1437,15 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
if (isKeyShareDuringSync) {
|
if (isKeyShareDuringSync) {
|
||||||
// appStateSyncKeyShare path: processMessage persists the new app-state-sync
|
// appStateSyncKeyShare path: processMessage persists the new app-state-sync
|
||||||
// key via keyStore.transaction at process-message.ts:650. The follow-up
|
// key in its APP_STATE_SYNC_KEY_SHARE handler (via keyStore.transaction).
|
||||||
// doAppStateSync() needs that key to decrypt patches, so we MUST await
|
// The follow-up doAppStateSync() needs that key to decrypt patches, so we
|
||||||
// processMessage first.
|
// MUST await processMessage first.
|
||||||
//
|
//
|
||||||
// IMPORTANT: do NOT wrap in messageMutex here. The inbound caller in
|
// IMPORTANT: do NOT wrap in messageMutex here. The inbound caller in
|
||||||
// messages-recv.ts:2416 already holds messageMutex.mutex(chatId, ...) for
|
// messages-recv.ts already holds messageMutex.mutex(chatId, ...) for this
|
||||||
// this chat; re-acquiring the same keyed mutex while awaiting it would
|
// chat; re-acquiring the same keyed mutex while awaiting it would deadlock
|
||||||
// deadlock (async-mutex is not re-entrant). Running inline preserves
|
// (async-mutex is not re-entrant). Running inline preserves per-chat
|
||||||
// per-chat ordering via the caller's outer mutex.
|
// ordering via the caller's outer mutex.
|
||||||
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
|
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
|
||||||
try {
|
try {
|
||||||
await postUpsertTasks()
|
await postUpsertTasks()
|
||||||
@@ -1461,9 +1461,15 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
// ordering of processMessage side effects (chat.unreadCount, LID/PN mapping,
|
// ordering of processMessage side effects (chat.unreadCount, LID/PN mapping,
|
||||||
// messages.update, history downloads) is preserved across messages of the
|
// messages.update, history downloads) is preserved across messages of the
|
||||||
// same chat. Reentrancy is safe here because the outer mutex acquired by
|
// same chat. Reentrancy is safe here because the outer mutex acquired by
|
||||||
// messages-recv.ts releases as soon as upsertMessage returns (work is not
|
// the inbound caller releases as soon as upsertMessage returns (work is
|
||||||
// awaited), then this inner mutex enqueues per-chat.
|
// not awaited), then this inner mutex enqueues per-chat.
|
||||||
const postUpsertChatId = msg.key.remoteJid || msg.key.id || 'unknown'
|
//
|
||||||
|
// Use getChatId + jidNormalizedUser so the mutex key matches the chat-id
|
||||||
|
// scheme processMessage uses for chat updates (broadcasts target the
|
||||||
|
// participant). Falling back to a constant bucket avoids fragmenting the
|
||||||
|
// queue with per-message ids when the key is malformed.
|
||||||
|
const rawChatId = getChatId(msg.key)
|
||||||
|
const postUpsertChatId = rawChatId ? jidNormalizedUser(rawChatId) : 'unknown'
|
||||||
messageMutex.mutex(postUpsertChatId, postUpsertTasks).catch(err =>
|
messageMutex.mutex(postUpsertChatId, postUpsertTasks).catch(err =>
|
||||||
logger?.warn(
|
logger?.warn(
|
||||||
{
|
{
|
||||||
|
|||||||
Reference in New Issue
Block a user