messages-recv: simplify retry logic + re-do acks
This commit is contained in:
Rajeh Taher
+82
-82
@@ -1187,91 +1187,91 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
||||
}
|
||||
|
||||
try {
|
||||
await Promise.all([
|
||||
processingMutex.mutex(async () => {
|
||||
await decrypt()
|
||||
// message failed to decrypt
|
||||
if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT) {
|
||||
if (msg?.messageStubParameters?.[0] === MISSING_KEYS_ERROR_TEXT) {
|
||||
return sendMessageAck(node, NACK_REASONS.ParsingError)
|
||||
}
|
||||
|
||||
const errorMessage = msg?.messageStubParameters?.[0] || ''
|
||||
const isPreKeyError = errorMessage.includes('PreKey')
|
||||
|
||||
logger.debug(`[handleMessage] Attempting retry request for failed decryption`)
|
||||
|
||||
// Handle both pre-key and normal retries in single mutex
|
||||
retryMutex.mutex(async () => {
|
||||
try {
|
||||
if (!ws.isOpen) {
|
||||
logger.debug({ node }, 'Connection closed, skipping retry')
|
||||
return
|
||||
}
|
||||
|
||||
// Handle pre-key errors with upload and delay
|
||||
if (isPreKeyError) {
|
||||
logger.info({ error: errorMessage }, 'PreKey error detected, uploading and retrying')
|
||||
|
||||
try {
|
||||
logger.debug('Uploading pre-keys for error recovery')
|
||||
await uploadPreKeys(5)
|
||||
logger.debug('Waiting for server to process new pre-keys')
|
||||
await delay(1000)
|
||||
} catch (uploadErr) {
|
||||
logger.error({ uploadErr }, 'Pre-key upload failed, proceeding with retry anyway')
|
||||
}
|
||||
}
|
||||
|
||||
const encNode = getBinaryNodeChild(node, 'enc')
|
||||
await sendRetryRequest(node, !encNode)
|
||||
if (retryRequestDelayMs) {
|
||||
await delay(retryRequestDelayMs)
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error({ err, isPreKeyError }, 'Failed to handle retry, attempting basic retry')
|
||||
// Still attempt retry even if pre-key upload failed
|
||||
try {
|
||||
const encNode = getBinaryNodeChild(node, 'enc')
|
||||
await sendRetryRequest(node, !encNode)
|
||||
} catch (retryErr) {
|
||||
logger.error({ retryErr }, 'Failed to send retry after error handling')
|
||||
}
|
||||
}
|
||||
})
|
||||
} else {
|
||||
// no type in the receipt => message delivered
|
||||
let type: MessageReceiptType = undefined
|
||||
let participant = msg.key.participant
|
||||
if (category === 'peer') {
|
||||
// special peer message
|
||||
type = 'peer_msg'
|
||||
} else if (msg.key.fromMe) {
|
||||
// message was sent by us from a different device
|
||||
type = 'sender'
|
||||
// need to specially handle this case
|
||||
if (isLidUser(msg.key.remoteJid!) || isLidUser(msg.key.remoteJidAlt)) {
|
||||
participant = author // TODO: investigate sending receipts to LIDs and not PNs
|
||||
}
|
||||
} else if (!sendActiveReceipts) {
|
||||
type = 'inactive'
|
||||
}
|
||||
|
||||
await sendReceipt(msg.key.remoteJid!, participant!, [msg.key.id!], type)
|
||||
|
||||
// send ack for history message
|
||||
const isAnyHistoryMsg = getHistoryMsg(msg.message!)
|
||||
if (isAnyHistoryMsg) {
|
||||
const jid = jidNormalizedUser(msg.key.remoteJid!)
|
||||
await sendReceipt(jid, undefined, [msg.key.id!], 'hist_sync')
|
||||
}
|
||||
await processingMutex.mutex(async () => {
|
||||
await decrypt()
|
||||
// message failed to decrypt
|
||||
if (msg.messageStubType === proto.WebMessageInfo.StubType.CIPHERTEXT) {
|
||||
if (msg?.messageStubParameters?.[0] === MISSING_KEYS_ERROR_TEXT) {
|
||||
return sendMessageAck(node, NACK_REASONS.ParsingError)
|
||||
}
|
||||
|
||||
cleanMessage(msg, authState.creds.me!.id, authState.creds.me!.lid!)
|
||||
const errorMessage = msg?.messageStubParameters?.[0] || ''
|
||||
const isPreKeyError = errorMessage.includes('PreKey')
|
||||
|
||||
await upsertMessage(msg, node.attrs.offline ? 'append' : 'notify')
|
||||
})
|
||||
])
|
||||
logger.debug(`[handleMessage] Attempting retry request for failed decryption`)
|
||||
|
||||
// Handle both pre-key and normal retries in single mutex
|
||||
retryMutex.mutex(async () => {
|
||||
try {
|
||||
if (!ws.isOpen) {
|
||||
logger.debug({ node }, 'Connection closed, skipping retry')
|
||||
return
|
||||
}
|
||||
|
||||
// Handle pre-key errors with upload and delay
|
||||
if (isPreKeyError) {
|
||||
logger.info({ error: errorMessage }, 'PreKey error detected, uploading and retrying')
|
||||
|
||||
try {
|
||||
logger.debug('Uploading pre-keys for error recovery')
|
||||
await uploadPreKeys(5)
|
||||
logger.debug('Waiting for server to process new pre-keys')
|
||||
await delay(1000)
|
||||
} catch (uploadErr) {
|
||||
logger.error({ uploadErr }, 'Pre-key upload failed, proceeding with retry anyway')
|
||||
}
|
||||
}
|
||||
|
||||
const encNode = getBinaryNodeChild(node, 'enc')
|
||||
await sendRetryRequest(node, !encNode)
|
||||
if (retryRequestDelayMs) {
|
||||
await delay(retryRequestDelayMs)
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error({ err, isPreKeyError }, 'Failed to handle retry, attempting basic retry')
|
||||
// Still attempt retry even if pre-key upload failed
|
||||
try {
|
||||
const encNode = getBinaryNodeChild(node, 'enc')
|
||||
await sendRetryRequest(node, !encNode)
|
||||
} catch (retryErr) {
|
||||
logger.error({ retryErr }, 'Failed to send retry after error handling')
|
||||
}
|
||||
}
|
||||
|
||||
await sendMessageAck(node, NACK_REASONS.UnhandledError)
|
||||
})
|
||||
} else {
|
||||
// no type in the receipt => message delivered
|
||||
let type: MessageReceiptType = undefined
|
||||
let participant = msg.key.participant
|
||||
if (category === 'peer') {
|
||||
// special peer message
|
||||
type = 'peer_msg'
|
||||
} else if (msg.key.fromMe) {
|
||||
// message was sent by us from a different device
|
||||
type = 'sender'
|
||||
// need to specially handle this case
|
||||
if (isLidUser(msg.key.remoteJid!) || isLidUser(msg.key.remoteJidAlt)) {
|
||||
participant = author // TODO: investigate sending receipts to LIDs and not PNs
|
||||
}
|
||||
} else if (!sendActiveReceipts) {
|
||||
type = 'inactive'
|
||||
}
|
||||
|
||||
await sendReceipt(msg.key.remoteJid!, participant!, [msg.key.id!], type)
|
||||
|
||||
// send ack for history message
|
||||
const isAnyHistoryMsg = getHistoryMsg(msg.message!)
|
||||
if (isAnyHistoryMsg) {
|
||||
const jid = jidNormalizedUser(msg.key.remoteJid!)
|
||||
await sendReceipt(jid, undefined, [msg.key.id!], 'hist_sync')
|
||||
}
|
||||
}
|
||||
|
||||
cleanMessage(msg, authState.creds.me!.id, authState.creds.me!.lid!)
|
||||
|
||||
await upsertMessage(msg, node.attrs.offline ? 'append' : 'notify')
|
||||
})
|
||||
} catch (error) {
|
||||
logger.error({ error, node: binaryNodeToString(node) }, 'error in handling message')
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user