Compare commits

..

3 Commits

Author SHA1 Message Date
Renato Alcara f5c07163f8 fix: address PR review — parallelize LID resolution, use helper consistently
- normalizeGroupMetadata: resolve participant LIDs with Promise.all
  instead of sequential loop (perf on large groups)
- handleCall: resolve participant JIDs with Promise.all
- handleBadAck: use normalizeKeyLidToPn() helper instead of manual
  resolveLidToPn + assignment (consistency with other code paths)
- CB:relay: resolve callCreator LID→PN before emitting

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-01 17:39:52 -03:00
Renato Alcara 0a43ad849d fix: normalize LID→PN in handleBadAck, media retry, and PDO recovery
Additional leak points found during final audit:
- handleBadAck: key.remoteJid from ack stanza could be LID
- messages.media-update: media retry key JIDs not normalized
- CTWA PDO recovery: webMessageInfo.key from phone response not normalized

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-01 17:29:22 -03:00
Renato Alcara ac7bb321d4 feat: centralized LID→PN normalization for all emitted events
WhatsApp's server increasingly uses LID (Linked ID) as the primary
addressing format. Consumers (ZPRO, etc.) expect phone numbers (PN),
not opaque LID identifiers. This adds comprehensive LID→PN resolution
across all ev.emit() paths so downstream consumers always receive PN.

Key changes:

- Add resolveLidToPn() and normalizeKeyLidToPn() centralized helpers
  in process-message.ts for consistent LID→PN resolution

- normalizeMessageJids() fast path: use alt JID directly from stanza
  attributes (zero I/O, eliminates race condition with LIDMappingStore)

- Await storeLIDPNMappings() before normalization in message receipt
  flow (was fire-and-forget, could race with subsequent getPNForLID)

- Normalize LID→PN in all event emission points:
  * messages.upsert (keys, nested reaction/poll keys, participantAlt)
  * presence.update (jid + participant)
  * message-receipt.update (key + userJid)
  * contacts.update (picture notifications)
  * blocklist.update (blocklist JIDs)
  * call events (chatId, from)
  * group metadata (participants, owner, subjectOwner)
  * group notifications (acting participant, add/remove/promote/demote)
  * newsletter notifications (author, user JIDs)
  * sync actions (mutation index normalization)

- Make handlePresenceUpdate and handleGroupNotification async to
  support await on LID resolution

- Add normalizeGroupMetadata() helper in groups.ts for all
  extractGroupMetadata call sites

Performance: ~0.01ms per message (LRU cache hit). No impact on
message sending. First-contact resolution ~2-5ms (one-time per contact).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-01 17:25:14 -03:00
3 changed files with 31 additions and 99 deletions
+1 -63
View File
@@ -31,7 +31,6 @@ import {
aesDecryptCTR, aesDecryptCTR,
aesEncryptGCM, aesEncryptGCM,
cleanMessage, cleanMessage,
cleanupCorruptedSession,
Curve, Curve,
decodeMediaRetryNode, decodeMediaRetryNode,
decodeMessageNode, decodeMessageNode,
@@ -42,7 +41,6 @@ import {
encodeSignedDeviceIdentity, encodeSignedDeviceIdentity,
extractAddressingContext, extractAddressingContext,
getCallStatusFromNode, getCallStatusFromNode,
getDecryptionJid,
getHistoryMsg, getHistoryMsg,
getNextPreKeys, getNextPreKeys,
getStatusFromReceiptType, getStatusFromReceiptType,
@@ -164,12 +162,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
const TC_TOKEN_PRUNE_TS_KEY = '__prune_ts' const TC_TOKEN_PRUNE_TS_KEY = '__prune_ts'
const tcTokenKnownJids = new Set<string>() const tcTokenKnownJids = new Set<string>()
const tcTokenRetriedMsgIds = new Set<string>() const tcTokenRetriedMsgIds = new Set<string>()
// Deduplicates retry requests per JID within a short window.
// When a burst of Bad MAC errors arrives for the same contact,
// only the first retry request is sent — the peer resends everything
// with a single pkmsg, avoiding the close-session cascade.
const retryRequestActiveJids = new Set<string>()
let tcTokenIndexSaveTimer: ReturnType<typeof setTimeout> | undefined let tcTokenIndexSaveTimer: ReturnType<typeof setTimeout> | undefined
let lastTcTokenPruneTs = 0 let lastTcTokenPruneTs = 0
@@ -1217,50 +1209,12 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
const { key: msgKey } = fullMessage const { key: msgKey } = fullMessage
const msgId = msgKey.id! const msgId = msgKey.id!
// Per-JID deduplication: when multiple messages from the same contact
// fail with Bad MAC simultaneously, only send ONE retry request.
// The peer will resend all failed messages when it receives the retry receipt.
// For group messages, scope by participant (each participant has its own Signal session).
const retryDedupeJid = msgKey.participant
? jidNormalizedUser(msgKey.participant)
: jidNormalizedUser(node.attrs.from!)
if (retryRequestActiveJids.has(retryDedupeJid)) {
logger.debug(
{ fromJid: retryDedupeJid, msgId },
'Skipping duplicate retry request — already in-flight for this JID'
)
return
}
if (messageRetryManager) { if (messageRetryManager) {
// Check if we've exceeded max retries using the new system // Check if we've exceeded max retries using the new system
if (messageRetryManager.hasExceededMaxRetries(msgId)) { if (messageRetryManager.hasExceededMaxRetries(msgId)) {
logger.debug({ msgId }, 'reached retry limit with new retry manager, clearing') logger.debug({ msgId }, 'reached retry limit with new retry manager, clearing')
messageRetryManager.markRetryFailed(msgId) messageRetryManager.markRetryFailed(msgId)
recordMessageFailure('retry', 'max_retries_reached') recordMessageFailure('retry', 'max_retries_reached')
// Safety net: clean up corrupted sessions only after all retries exhausted.
// This avoids the cascading delete loop that occurs when cleanup runs
// on every Bad MAC in the hot path (decode-wa-message.ts).
// The Signal Protocol recovers naturally via retry+pkmsg for most cases;
// this cleanup only runs as a last resort.
// For group messages, use participant JID (Signal sessions are per-participant, not per-group).
if (autoCleanCorrupted) {
const senderJid = msgKey.participant ? jidNormalizedUser(msgKey.participant) : jidNormalizedUser(node.attrs.from!)
try {
const decryptionJid = await getDecryptionJid(senderJid, signalRepository)
const deletedCount = await cleanupCorruptedSession(decryptionJid, signalRepository, logger)
if (deletedCount > 0) {
logger.info(
{ msgId, jid: decryptionJid, targetedDevices: deletedCount },
`🔄 Session cleanup (retry exhausted) | Targeted: ${deletedCount} devices`
)
}
} catch (cleanupErr) {
logger.warn({ msgId, senderJid, err: cleanupErr }, 'Failed to cleanup session after retry exhaustion')
}
}
return return
} }
@@ -1279,18 +1233,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
logger.debug({ retryCount, msgId }, 'reached retry limit, clearing') logger.debug({ retryCount, msgId }, 'reached retry limit, clearing')
await msgRetryCache.del(key) await msgRetryCache.del(key)
recordMessageFailure('retry', 'max_retries_reached') recordMessageFailure('retry', 'max_retries_reached')
// Safety net cleanup (same as new system above)
if (autoCleanCorrupted) {
const senderJid = msgKey.participant ? jidNormalizedUser(msgKey.participant) : jidNormalizedUser(node.attrs.from!)
try {
const decryptionJid = await getDecryptionJid(senderJid, signalRepository)
await cleanupCorruptedSession(decryptionJid, signalRepository, logger)
} catch (cleanupErr) {
logger.warn({ msgId, senderJid, err: cleanupErr }, 'Failed to cleanup session after retry exhaustion')
}
}
return return
} }
@@ -1299,11 +1241,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
recordMessageRetry('retry') recordMessageRetry('retry')
} }
// Register dedup AFTER early-return checks so that max-retries paths
// don't block subsequent messages from the same JID for 5 seconds.
retryRequestActiveJids.add(retryDedupeJid)
setTimeout(() => retryRequestActiveJids.delete(retryDedupeJid), 5_000)
const key = `${msgId}:${msgKey?.participant}` const key = `${msgId}:${msgKey?.participant}`
const retryCount = (await msgRetryCache.get<number>(key)) || 1 const retryCount = (await msgRetryCache.get<number>(key)) || 1
@@ -2234,6 +2171,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
authState.creds.me!.lid || '', authState.creds.me!.lid || '',
signalRepository, signalRepository,
logger, logger,
autoCleanCorrupted
) )
const alt = msg.key.participantAlt || msg.key.remoteJidAlt const alt = msg.key.participantAlt || msg.key.remoteJidAlt
+26 -14
View File
@@ -276,6 +276,7 @@ export const decryptMessageNode = (
meLid: string, meLid: string,
repository: SignalRepositoryWithLIDStore, repository: SignalRepositoryWithLIDStore,
logger: ILogger, logger: ILogger,
autoCleanCorrupted = true
) => { ) => {
const { fullMessage, author, sender } = decodeMessageNode(stanza, meId, meLid) const { fullMessage, author, sender } = decodeMessageNode(stanza, meId, meLid)
return { return {
@@ -421,19 +422,34 @@ export const decryptMessageNode = (
if (isRetryExhausted) { if (isRetryExhausted) {
logger.error( logger.error(
errorContext, errorContext,
`⚠️ Session corrupted after ${err.attempts} attempts. Retry+pkmsg flow will recover.` `⚠️ Session corrupted and recovery failed after ${err.attempts} attempts. Auto-cleanup will attempt to recover.`
) )
} else { } else {
// First occurrence - log as warning since auto-recovery will attempt // First occurrence - log as warning since auto-recovery will attempt
logger.warn(errorContext, '⚠️ Corrupted session detected - attempting auto-recovery') logger.warn(errorContext, '⚠️ Corrupted session detected - attempting auto-recovery')
} }
// Session cleanup is deferred to retry exhaustion (safety net). // Automatic cleanup of corrupted session (if enabled)
// The Signal Protocol handles recovery naturally via retry+pkmsg: // eslint-disable-next-line max-depth
// Bad MAC -> retry receipt -> sender re-sends as pkmsg -> new session. if (autoCleanCorrupted) {
// Deleting sessions here (hot path) causes cascading failures when // eslint-disable-next-line max-depth
// multiple messages from the same contact arrive simultaneously. try {
// See: messages-recv.ts sendRetryRequest() for deferred cleanup. const deletedCount = await cleanupCorruptedSession(decryptionJid, repository, logger)
// Mask only user portion of JID for privacy (preserve domain info)
const { user, server } = jidDecode(decryptionJid) || {}
const maskedUser =
user && user.length > 8 ? `${user.substring(0, 4)}****${user.substring(user.length - 4)}` : user
const maskedJid = maskedUser && server ? `${maskedUser}@${server}` : decryptionJid
logger.info(
{ jid: decryptionJid, maskedJid, targetedDevices: deletedCount },
`🔄 Session Reset | JID: ${maskedJid} | Targeted: ${deletedCount} devices | Will recreate on next message`
)
} catch (cleanupErr) {
logger.error({ decryptionJid, err: cleanupErr }, '❌ Failed to cleanup corrupted session')
}
}
} else if (isSessionRecord) { } else if (isSessionRecord) {
// Session record errors are transient - retry should handle them // Session record errors are transient - retry should handle them
// eslint-disable-next-line max-depth // eslint-disable-next-line max-depth
@@ -488,14 +504,10 @@ export function isCorruptedSessionError(error: any): boolean {
} }
/** /**
* Clean up corrupted session by deleting all device sessions for a JID. * Clean up corrupted session by deleting all device sessions for a JID
* Signal Protocol will automatically recreate the session on next message. * Signal Protocol will automatically recreate the session on next message
*
* NOTE: This should NOT be called on every Bad MAC error (hot path).
* Instead, let the retry+pkmsg flow handle recovery naturally (like WhatsApp does).
* Only call this as a safety net when retries are exhausted.
*/ */
export async function cleanupCorruptedSession( async function cleanupCorruptedSession(
jid: string, jid: string,
repository: SignalRepositoryWithLIDStore, repository: SignalRepositoryWithLIDStore,
logger: ILogger logger: ILogger
+4 -22
View File
@@ -211,33 +211,15 @@ export class MessageRetryManager {
} }
} }
// MAC errors require session recreation, but rate-limited to prevent loops. // MAC errors require IMMEDIATE session recreation regardless of history
// When multiple messages fail with Bad MAC simultaneously, only the first // This handles the case where contact reinstalled WhatsApp (identity key changed)
// should trigger recreation; subsequent ones within the cooldown window
// piggyback on the same new session established by the retry+pkmsg flow.
if (errorCode !== undefined && MAC_ERROR_CODES.has(errorCode)) { if (errorCode !== undefined && MAC_ERROR_CODES.has(errorCode)) {
const now = Date.now() this.sessionRecreateHistory.set(jid, Date.now())
const prevTime = this.sessionRecreateHistory.get(jid)
const MAC_ERROR_COOLDOWN_MS = 10_000 // 10 seconds
if (prevTime && now - prevTime < MAC_ERROR_COOLDOWN_MS) {
const reasonName = RetryReason[errorCode] || `code_${errorCode}`
this.logger.debug(
{ jid, errorCode: reasonName, msSinceLast: now - prevTime },
'MAC error session recreation skipped — cooldown active'
)
return {
reason: '',
recreate: false
}
}
this.sessionRecreateHistory.set(jid, now)
this.statistics.sessionRecreations++ this.statistics.sessionRecreations++
const reasonName = RetryReason[errorCode] || `code_${errorCode}` const reasonName = RetryReason[errorCode] || `code_${errorCode}`
metrics.signalMacErrors?.inc({ action: 'session_recreation' }) metrics.signalMacErrors?.inc({ action: 'session_recreation' })
metrics.signalSessionRecreations?.inc({ reason: 'mac_error' }) metrics.signalSessionRecreations?.inc({ reason: 'mac_error' })
this.logger.warn({ jid, errorCode: reasonName }, 'MAC error detected, recreating session') this.logger.warn({ jid, errorCode: reasonName }, 'MAC error detected, forcing immediate session recreation')
return { return {
reason: `MAC error (${reasonName}) - contact may have reinstalled WhatsApp`, reason: `MAC error (${reasonName}) - contact may have reinstalled WhatsApp`,
recreate: true recreate: true