fix: sync pre-keys on socket start and add specific pre-key error approach (#1663)
* fix: sync prekeys when storage mismatch * fix: verify if current pre-key exists instead of matching * add: pre-keys error specific approach * fix: lint * fix: lint
This commit is contained in:
@@ -111,6 +111,9 @@ export const MIN_PREKEY_COUNT = 5
|
|||||||
|
|
||||||
export const INITIAL_PREKEY_COUNT = 30
|
export const INITIAL_PREKEY_COUNT = 30
|
||||||
|
|
||||||
|
export const UPLOAD_TIMEOUT = 30000 // 30 seconds
|
||||||
|
export const MIN_UPLOAD_INTERVAL = 5000 // 5 seconds minimum between uploads
|
||||||
|
|
||||||
export const DEFAULT_CACHE_TTLS = {
|
export const DEFAULT_CACHE_TTLS = {
|
||||||
SIGNAL_STORE: 5 * 60, // 5 minutes
|
SIGNAL_STORE: 5 * 60, // 5 minutes
|
||||||
MSG_RETRY: 60 * 60, // 1 hour
|
MSG_RETRY: 60 * 60, // 1 hour
|
||||||
|
|||||||
@@ -711,8 +711,11 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
updateSendMessageAgainCount(ids[0], key.participant)
|
updateSendMessageAgainCount(ids[0], key.participant)
|
||||||
logger.debug({ attrs, key }, 'recv retry request')
|
logger.debug({ attrs, key }, 'recv retry request')
|
||||||
await sendMessagesAgain(key, ids, retryNode!)
|
await sendMessagesAgain(key, ids, retryNode!)
|
||||||
} catch (error: any) {
|
} catch (error: unknown) {
|
||||||
logger.error({ key, ids, trace: error.stack }, 'error in sending message again')
|
logger.error(
|
||||||
|
{ key, ids, trace: error instanceof Error ? error.stack : 'Unknown error' },
|
||||||
|
'error in sending message again'
|
||||||
|
)
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
logger.info({ attrs, key }, 'recv retry for not fromMe message')
|
logger.info({ attrs, key }, 'recv retry for not fromMe message')
|
||||||
@@ -823,19 +826,52 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
return sendMessageAck(node, NACK_REASONS.ParsingError)
|
return sendMessageAck(node, NACK_REASONS.ParsingError)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const errorMessage = msg?.messageStubParameters?.[0] || ''
|
||||||
|
const isPreKeyError = errorMessage.includes('PreKey')
|
||||||
|
|
||||||
|
console.debug(`[handleMessage] Attempting retry request for failed decryption`)
|
||||||
|
|
||||||
|
// Handle both pre-key and normal retries in single mutex
|
||||||
retryMutex.mutex(async () => {
|
retryMutex.mutex(async () => {
|
||||||
if (ws.isOpen) {
|
try {
|
||||||
if (getBinaryNodeChild(node, 'unavailable')) {
|
if (!ws.isOpen) {
|
||||||
|
logger.debug({ node }, 'Connection closed, skipping retry')
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (getBinaryNodeChild(node, 'unavailable')) {
|
||||||
|
logger.debug('Message unavailable, 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')
|
const encNode = getBinaryNodeChild(node, 'enc')
|
||||||
await sendRetryRequest(node, !encNode)
|
await sendRetryRequest(node, !encNode)
|
||||||
if (retryRequestDelayMs) {
|
if (retryRequestDelayMs) {
|
||||||
await delay(retryRequestDelayMs)
|
await delay(retryRequestDelayMs)
|
||||||
}
|
}
|
||||||
} else {
|
} catch (err) {
|
||||||
logger.debug({ node }, 'connection closed, ignoring retry req')
|
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 {
|
} else {
|
||||||
|
|||||||
+103
-13
@@ -8,7 +8,9 @@ import {
|
|||||||
DEF_TAG_PREFIX,
|
DEF_TAG_PREFIX,
|
||||||
INITIAL_PREKEY_COUNT,
|
INITIAL_PREKEY_COUNT,
|
||||||
MIN_PREKEY_COUNT,
|
MIN_PREKEY_COUNT,
|
||||||
NOISE_WA_HEADER
|
MIN_UPLOAD_INTERVAL,
|
||||||
|
NOISE_WA_HEADER,
|
||||||
|
UPLOAD_TIMEOUT
|
||||||
} from '../Defaults'
|
} from '../Defaults'
|
||||||
import { cleanupQueues } from '../Signal/Group/queue-job'
|
import { cleanupQueues } from '../Signal/Group/queue-job'
|
||||||
import type { SocketConfig } from '../Types'
|
import type { SocketConfig } from '../Types'
|
||||||
@@ -298,24 +300,112 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
return +countChild.attrs.value!
|
return +countChild.attrs.value!
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Pre-key upload state management
|
||||||
|
let uploadPreKeysPromise: Promise<void> | null = null
|
||||||
|
let lastUploadTime = 0
|
||||||
|
|
||||||
/** generates and uploads a set of pre-keys to the server */
|
/** generates and uploads a set of pre-keys to the server */
|
||||||
const uploadPreKeys = async (count = INITIAL_PREKEY_COUNT) => {
|
const uploadPreKeys = async (count = INITIAL_PREKEY_COUNT, retryCount = 0) => {
|
||||||
await keys.transaction(async () => {
|
// Check minimum interval (except for retries)
|
||||||
logger.info({ count }, 'uploading pre-keys')
|
if (retryCount === 0) {
|
||||||
const { update, node } = await getNextPreKeysNode({ creds, keys }, count)
|
const timeSinceLastUpload = Date.now() - lastUploadTime
|
||||||
|
if (timeSinceLastUpload < MIN_UPLOAD_INTERVAL) {
|
||||||
|
logger.debug(`Skipping upload, only ${timeSinceLastUpload}ms since last upload`)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
await query(node)
|
// Prevent multiple concurrent uploads
|
||||||
ev.emit('creds.update', update)
|
if (uploadPreKeysPromise) {
|
||||||
|
logger.debug('Pre-key upload already in progress, waiting for completion')
|
||||||
|
return uploadPreKeysPromise
|
||||||
|
}
|
||||||
|
|
||||||
logger.info({ count }, 'uploaded pre-keys')
|
const uploadLogic = async () => {
|
||||||
})
|
logger.info({ count, retryCount }, 'uploading pre-keys')
|
||||||
|
|
||||||
|
// Generate and save pre-keys atomically (prevents ID collisions on retry)
|
||||||
|
const node = await keys.transaction(async () => {
|
||||||
|
logger.debug({ requestedCount: count }, 'generating pre-keys with requested count')
|
||||||
|
const { update, node } = await getNextPreKeysNode({ creds, keys }, count)
|
||||||
|
// Update credentials immediately to prevent duplicate IDs on retry
|
||||||
|
ev.emit('creds.update', update)
|
||||||
|
return node // Only return node since update is already used
|
||||||
|
})
|
||||||
|
|
||||||
|
// Upload to server (outside transaction, can fail without affecting local keys)
|
||||||
|
try {
|
||||||
|
await query(node)
|
||||||
|
logger.info({ count }, 'uploaded pre-keys successfully')
|
||||||
|
lastUploadTime = Date.now()
|
||||||
|
} catch (uploadError) {
|
||||||
|
logger.error({ uploadError, count }, 'Failed to upload pre-keys to server')
|
||||||
|
|
||||||
|
// Exponential backoff retry (max 3 retries)
|
||||||
|
if (retryCount < 3) {
|
||||||
|
const backoffDelay = Math.min(1000 * Math.pow(2, retryCount), 10000)
|
||||||
|
logger.info(`Retrying pre-key upload in ${backoffDelay}ms`)
|
||||||
|
await new Promise(resolve => setTimeout(resolve, backoffDelay))
|
||||||
|
return uploadPreKeys(count, retryCount + 1)
|
||||||
|
}
|
||||||
|
|
||||||
|
throw uploadError
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Add timeout protection
|
||||||
|
uploadPreKeysPromise = Promise.race([
|
||||||
|
uploadLogic(),
|
||||||
|
new Promise<void>((_, reject) =>
|
||||||
|
setTimeout(() => reject(new Boom('Pre-key upload timeout', { statusCode: 408 })), UPLOAD_TIMEOUT)
|
||||||
|
)
|
||||||
|
])
|
||||||
|
|
||||||
|
try {
|
||||||
|
await uploadPreKeysPromise
|
||||||
|
} finally {
|
||||||
|
uploadPreKeysPromise = null
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
const verifyCurrentPreKeyExists = async () => {
|
||||||
|
const currentPreKeyId = creds.nextPreKeyId - 1
|
||||||
|
if (currentPreKeyId <= 0) {
|
||||||
|
return { exists: false, currentPreKeyId: 0 }
|
||||||
|
}
|
||||||
|
|
||||||
|
const preKeys = await keys.get('pre-key', [currentPreKeyId.toString()])
|
||||||
|
const exists = !!preKeys[currentPreKeyId.toString()]
|
||||||
|
|
||||||
|
return { exists, currentPreKeyId }
|
||||||
}
|
}
|
||||||
|
|
||||||
const uploadPreKeysToServerIfRequired = async () => {
|
const uploadPreKeysToServerIfRequired = async () => {
|
||||||
const preKeyCount = await getAvailablePreKeysOnServer()
|
try {
|
||||||
logger.info(`${preKeyCount} pre-keys found on server`)
|
const preKeyCount = await getAvailablePreKeysOnServer()
|
||||||
if (preKeyCount <= MIN_PREKEY_COUNT) {
|
const { exists: currentPreKeyExists, currentPreKeyId } = await verifyCurrentPreKeyExists()
|
||||||
await uploadPreKeys()
|
|
||||||
|
logger.info(`${preKeyCount} pre-keys found on server`)
|
||||||
|
logger.info(`Current prekey ID: ${currentPreKeyId}, exists in storage: ${currentPreKeyExists}`)
|
||||||
|
|
||||||
|
const lowServerCount = preKeyCount <= MIN_PREKEY_COUNT
|
||||||
|
const missingCurrentPreKey = !currentPreKeyExists && currentPreKeyId > 0
|
||||||
|
|
||||||
|
const shouldUpload = lowServerCount || missingCurrentPreKey
|
||||||
|
|
||||||
|
if (shouldUpload) {
|
||||||
|
const reasons = []
|
||||||
|
if (lowServerCount) reasons.push(`server count low (${preKeyCount})`)
|
||||||
|
if (missingCurrentPreKey) reasons.push(`current prekey ${currentPreKeyId} missing from storage`)
|
||||||
|
|
||||||
|
logger.info(`Uploading PreKeys due to: ${reasons.join(', ')}`)
|
||||||
|
await uploadPreKeys()
|
||||||
|
} else {
|
||||||
|
logger.info(`PreKey validation passed - Server: ${preKeyCount}, Current prekey ${currentPreKeyId} exists`)
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
logger.error({ error }, 'Failed to check/upload pre-keys during initialization')
|
||||||
|
// Don't throw - allow connection to continue even if pre-key check fails
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user