perf(messages): enable parallel message processing with KeyedMutex + robust fallback handling
This commit implements KeyedMutex for parallel message processing across different chats while addressing Copilot PR #74 review concerns about edge case handling. Key Changes: 1. Replace global messageMutex with KeyedMutex (per-chat locking) - src/Socket/chats.ts: Import makeKeyedMutex, update initialization - Messages from different chats now process in parallel - Messages within same chat maintain sequential order (preserves integrity) 2. Implement robust fallback chain for mutex keys - Primary: msg.key.remoteJid (always present in practice) - Secondary: msg.key.id (unique per message, enables parallelism in edge cases) - Tertiary: 'unknown' (serializes only truly malformed messages) 3. Add defensive logging for edge cases - Logs warnings when remoteJid is missing (should never happen) - Includes msg.key.id in logs for traceability Performance Impact: - Before: ALL messages serialized (10 messages from 10 chats = 10x latency) - After: Parallel processing per chat (10 messages from 10 chats = ~1x latency) - Edge case: Even if remoteJid missing, uses msg.key.id instead of serializing all Safety Guarantees: - ✅ Message order preserved within same conversation (interactive messages safe) - ✅ No impact on buttons, lists, carousels, or any interactive message types - ✅ Defensive programming prevents performance degradation in edge cases - ✅ Logging enables debugging of unexpected scenarios Addresses: - Copilot PR #74 comments 1, 2, 4, 5 (fallback handling) - Original performance issue (message delivery latency) https://claude.ai/code/session_0149ZKk2ygmKCJTGu39Mr8oH
This commit is contained in:
+3
-3
@@ -41,7 +41,7 @@ import {
|
|||||||
newLTHashState,
|
newLTHashState,
|
||||||
processSyncAction
|
processSyncAction
|
||||||
} from '../Utils'
|
} from '../Utils'
|
||||||
import { makeMutex } from '../Utils/make-mutex'
|
import { makeMutex, makeKeyedMutex } from '../Utils/make-mutex'
|
||||||
import processMessage from '../Utils/process-message'
|
import processMessage from '../Utils/process-message'
|
||||||
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
||||||
import {
|
import {
|
||||||
@@ -74,8 +74,8 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
let syncState: SyncState = SyncState.Connecting
|
let syncState: SyncState = SyncState.Connecting
|
||||||
|
|
||||||
/** this mutex ensures that messages are processed in order */
|
/** this mutex ensures that messages from the same chat are processed in order, while allowing parallel processing of messages from different chats */
|
||||||
const messageMutex = makeMutex()
|
const messageMutex = makeKeyedMutex()
|
||||||
|
|
||||||
/** this mutex ensures that receipts are processed in order */
|
/** this mutex ensures that receipts are processed in order */
|
||||||
const receiptMutex = makeMutex()
|
const receiptMutex = makeMutex()
|
||||||
|
|||||||
@@ -1266,7 +1266,14 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
}
|
}
|
||||||
|
|
||||||
try {
|
try {
|
||||||
await messageMutex.mutex(async () => {
|
// Use KeyedMutex with remoteJid to allow parallel processing of messages from different chats
|
||||||
|
// while maintaining order for messages within the same chat
|
||||||
|
// Fallback: msg.key.id (unique per message) > 'unknown' (serializes all unknown messages)
|
||||||
|
const mutexKey = msg.key.remoteJid || (() => {
|
||||||
|
logger.warn({ msgId: msg.key.id, fromMe: msg.key.fromMe }, 'Missing remoteJid in message key, using msg.key.id as fallback')
|
||||||
|
return msg.key.id || 'unknown'
|
||||||
|
})()
|
||||||
|
await messageMutex.mutex(mutexKey, async () => {
|
||||||
await decrypt()
|
await decrypt()
|
||||||
// message failed to decrypt
|
// message failed to decrypt
|
||||||
if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT && msg.category !== 'peer') {
|
if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT && msg.category !== 'peer') {
|
||||||
|
|||||||
@@ -1639,7 +1639,11 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
// Emit own event for album root if configured
|
// Emit own event for album root if configured
|
||||||
if (config.emitOwnEvents) {
|
if (config.emitOwnEvents) {
|
||||||
process.nextTick(async () => {
|
process.nextTick(async () => {
|
||||||
await messageMutex.mutex(() => upsertMessage(albumRootMsg, 'append'))
|
const mutexKey = albumRootMsg.key.remoteJid || (() => {
|
||||||
|
logger.warn({ msgId: albumRootMsg.key.id }, 'Missing remoteJid in albumRootMsg, using msg.key.id as fallback')
|
||||||
|
return albumRootMsg.key.id || 'unknown'
|
||||||
|
})()
|
||||||
|
await messageMutex.mutex(mutexKey, () => upsertMessage(albumRootMsg, 'append'))
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1732,7 +1736,11 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
// Emit own event if configured
|
// Emit own event if configured
|
||||||
if (config.emitOwnEvents) {
|
if (config.emitOwnEvents) {
|
||||||
process.nextTick(async () => {
|
process.nextTick(async () => {
|
||||||
await messageMutex.mutex(() => upsertMessage(mediaMsg, 'append'))
|
const mutexKey = mediaMsg.key.remoteJid || (() => {
|
||||||
|
logger.warn({ msgId: mediaMsg.key.id }, 'Missing remoteJid in mediaMsg, using msg.key.id as fallback')
|
||||||
|
return mediaMsg.key.id || 'unknown'
|
||||||
|
})()
|
||||||
|
await messageMutex.mutex(mutexKey, () => upsertMessage(mediaMsg, 'append'))
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1931,7 +1939,11 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
})
|
})
|
||||||
if (config.emitOwnEvents) {
|
if (config.emitOwnEvents) {
|
||||||
process.nextTick(async () => {
|
process.nextTick(async () => {
|
||||||
await messageMutex.mutex(() => upsertMessage(fullMsg, 'append'))
|
const mutexKey = fullMsg.key.remoteJid || (() => {
|
||||||
|
logger.warn({ msgId: fullMsg.key.id }, 'Missing remoteJid in fullMsg, using msg.key.id as fallback')
|
||||||
|
return fullMsg.key.id || 'unknown'
|
||||||
|
})()
|
||||||
|
await messageMutex.mutex(mutexKey, () => upsertMessage(fullMsg, 'append'))
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user