Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f5c07163f8 | |||
| 0a43ad849d | |||
| ac7bb321d4 |
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
Reference in New Issue
Block a user