Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d28d8a4596 | |||
| 17e3bd5d36 |
@@ -31,6 +31,7 @@ import {
|
|||||||
aesDecryptCTR,
|
aesDecryptCTR,
|
||||||
aesEncryptGCM,
|
aesEncryptGCM,
|
||||||
cleanMessage,
|
cleanMessage,
|
||||||
|
cleanupCorruptedSession,
|
||||||
Curve,
|
Curve,
|
||||||
decodeMediaRetryNode,
|
decodeMediaRetryNode,
|
||||||
decodeMessageNode,
|
decodeMessageNode,
|
||||||
@@ -41,6 +42,7 @@ import {
|
|||||||
encodeSignedDeviceIdentity,
|
encodeSignedDeviceIdentity,
|
||||||
extractAddressingContext,
|
extractAddressingContext,
|
||||||
getCallStatusFromNode,
|
getCallStatusFromNode,
|
||||||
|
getDecryptionJid,
|
||||||
getHistoryMsg,
|
getHistoryMsg,
|
||||||
getNextPreKeys,
|
getNextPreKeys,
|
||||||
getStatusFromReceiptType,
|
getStatusFromReceiptType,
|
||||||
@@ -162,6 +164,12 @@ 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
|
||||||
|
|
||||||
@@ -1209,12 +1217,50 @@ 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
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1233,6 +1279,18 @@ 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
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1241,6 +1299,11 @@ 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
|
||||||
|
|
||||||
@@ -2171,7 +2234,6 @@ 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
|
||||||
|
|||||||
@@ -276,7 +276,6 @@ 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 {
|
||||||
@@ -422,34 +421,19 @@ export const decryptMessageNode = (
|
|||||||
if (isRetryExhausted) {
|
if (isRetryExhausted) {
|
||||||
logger.error(
|
logger.error(
|
||||||
errorContext,
|
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 {
|
} 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')
|
||||||
}
|
}
|
||||||
|
|
||||||
// Automatic cleanup of corrupted session (if enabled)
|
// Session cleanup is deferred to retry exhaustion (safety net).
|
||||||
// eslint-disable-next-line max-depth
|
// The Signal Protocol handles recovery naturally via retry+pkmsg:
|
||||||
if (autoCleanCorrupted) {
|
// Bad MAC -> retry receipt -> sender re-sends as pkmsg -> new session.
|
||||||
// eslint-disable-next-line max-depth
|
// Deleting sessions here (hot path) causes cascading failures when
|
||||||
try {
|
// multiple messages from the same contact arrive simultaneously.
|
||||||
const deletedCount = await cleanupCorruptedSession(decryptionJid, repository, logger)
|
// See: messages-recv.ts sendRetryRequest() for deferred cleanup.
|
||||||
|
|
||||||
// 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
|
||||||
@@ -504,10 +488,14 @@ 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.
|
||||||
*/
|
*/
|
||||||
async function cleanupCorruptedSession(
|
export async function cleanupCorruptedSession(
|
||||||
jid: string,
|
jid: string,
|
||||||
repository: SignalRepositoryWithLIDStore,
|
repository: SignalRepositoryWithLIDStore,
|
||||||
logger: ILogger
|
logger: ILogger
|
||||||
|
|||||||
@@ -211,15 +211,33 @@ export class MessageRetryManager {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// MAC errors require IMMEDIATE session recreation regardless of history
|
// MAC errors require session recreation, but rate-limited to prevent loops.
|
||||||
// This handles the case where contact reinstalled WhatsApp (identity key changed)
|
// 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)) {
|
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++
|
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, forcing immediate session recreation')
|
this.logger.warn({ jid, errorCode: reasonName }, 'MAC error detected, recreating session')
|
||||||
return {
|
return {
|
||||||
reason: `MAC error (${reasonName}) - contact may have reinstalled WhatsApp`,
|
reason: `MAC error (${reasonName}) - contact may have reinstalled WhatsApp`,
|
||||||
recreate: true
|
recreate: true
|
||||||
|
|||||||
Reference in New Issue
Block a user