Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d28d8a4596 | |||
| 17e3bd5d36 |
@@ -31,6 +31,7 @@ import {
|
||||
aesDecryptCTR,
|
||||
aesEncryptGCM,
|
||||
cleanMessage,
|
||||
cleanupCorruptedSession,
|
||||
Curve,
|
||||
decodeMediaRetryNode,
|
||||
decodeMessageNode,
|
||||
@@ -41,6 +42,7 @@ import {
|
||||
encodeSignedDeviceIdentity,
|
||||
extractAddressingContext,
|
||||
getCallStatusFromNode,
|
||||
getDecryptionJid,
|
||||
getHistoryMsg,
|
||||
getNextPreKeys,
|
||||
getStatusFromReceiptType,
|
||||
@@ -162,6 +164,12 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
const TC_TOKEN_PRUNE_TS_KEY = '__prune_ts'
|
||||
const tcTokenKnownJids = 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 lastTcTokenPruneTs = 0
|
||||
|
||||
@@ -1209,12 +1217,50 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
const { key: msgKey } = fullMessage
|
||||
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) {
|
||||
// Check if we've exceeded max retries using the new system
|
||||
if (messageRetryManager.hasExceededMaxRetries(msgId)) {
|
||||
logger.debug({ msgId }, 'reached retry limit with new retry manager, clearing')
|
||||
messageRetryManager.markRetryFailed(msgId)
|
||||
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
|
||||
}
|
||||
|
||||
@@ -1233,6 +1279,18 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
logger.debug({ retryCount, msgId }, 'reached retry limit, clearing')
|
||||
await msgRetryCache.del(key)
|
||||
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
|
||||
}
|
||||
|
||||
@@ -1241,6 +1299,11 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
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 retryCount = (await msgRetryCache.get<number>(key)) || 1
|
||||
|
||||
@@ -2171,7 +2234,6 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
authState.creds.me!.lid || '',
|
||||
signalRepository,
|
||||
logger,
|
||||
autoCleanCorrupted
|
||||
)
|
||||
|
||||
const alt = msg.key.participantAlt || msg.key.remoteJidAlt
|
||||
|
||||
@@ -276,7 +276,6 @@ export const decryptMessageNode = (
|
||||
meLid: string,
|
||||
repository: SignalRepositoryWithLIDStore,
|
||||
logger: ILogger,
|
||||
autoCleanCorrupted = true
|
||||
) => {
|
||||
const { fullMessage, author, sender } = decodeMessageNode(stanza, meId, meLid)
|
||||
return {
|
||||
@@ -422,34 +421,19 @@ export const decryptMessageNode = (
|
||||
if (isRetryExhausted) {
|
||||
logger.error(
|
||||
errorContext,
|
||||
`⚠️ Session corrupted and recovery failed after ${err.attempts} attempts. Auto-cleanup will attempt to recover.`
|
||||
`⚠️ Session corrupted after ${err.attempts} attempts. Retry+pkmsg flow will recover.`
|
||||
)
|
||||
} else {
|
||||
// First occurrence - log as warning since auto-recovery will attempt
|
||||
logger.warn(errorContext, '⚠️ Corrupted session detected - attempting auto-recovery')
|
||||
}
|
||||
|
||||
// Automatic cleanup of corrupted session (if enabled)
|
||||
// eslint-disable-next-line max-depth
|
||||
if (autoCleanCorrupted) {
|
||||
// eslint-disable-next-line max-depth
|
||||
try {
|
||||
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')
|
||||
}
|
||||
}
|
||||
// Session cleanup is deferred to retry exhaustion (safety net).
|
||||
// The Signal Protocol handles recovery naturally via retry+pkmsg:
|
||||
// Bad MAC -> retry receipt -> sender re-sends as pkmsg -> new session.
|
||||
// Deleting sessions here (hot path) causes cascading failures when
|
||||
// multiple messages from the same contact arrive simultaneously.
|
||||
// See: messages-recv.ts sendRetryRequest() for deferred cleanup.
|
||||
} else if (isSessionRecord) {
|
||||
// Session record errors are transient - retry should handle them
|
||||
// eslint-disable-next-line max-depth
|
||||
@@ -504,10 +488,14 @@ export function isCorruptedSessionError(error: any): boolean {
|
||||
}
|
||||
|
||||
/**
|
||||
* Clean up corrupted session by deleting all device sessions for a JID
|
||||
* Signal Protocol will automatically recreate the session on next message
|
||||
* Clean up corrupted session by deleting all device sessions for a JID.
|
||||
* 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.
|
||||
*/
|
||||
async function cleanupCorruptedSession(
|
||||
export async function cleanupCorruptedSession(
|
||||
jid: string,
|
||||
repository: SignalRepositoryWithLIDStore,
|
||||
logger: ILogger
|
||||
|
||||
@@ -211,15 +211,33 @@ export class MessageRetryManager {
|
||||
}
|
||||
}
|
||||
|
||||
// MAC errors require IMMEDIATE session recreation regardless of history
|
||||
// This handles the case where contact reinstalled WhatsApp (identity key changed)
|
||||
// MAC errors require session recreation, but rate-limited to prevent loops.
|
||||
// When multiple messages fail with Bad MAC simultaneously, only the first
|
||||
// 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)) {
|
||||
this.sessionRecreateHistory.set(jid, Date.now())
|
||||
const now = 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++
|
||||
const reasonName = RetryReason[errorCode] || `code_${errorCode}`
|
||||
metrics.signalMacErrors?.inc({ action: 'session_recreation' })
|
||||
metrics.signalSessionRecreations?.inc({ reason: 'mac_error' })
|
||||
this.logger.warn({ jid, errorCode: reasonName }, 'MAC error detected, forcing immediate session recreation')
|
||||
this.logger.warn({ jid, errorCode: reasonName }, 'MAC error detected, recreating session')
|
||||
return {
|
||||
reason: `MAC error (${reasonName}) - contact may have reinstalled WhatsApp`,
|
||||
recreate: true
|
||||
|
||||
Reference in New Issue
Block a user