fix: tctoken 463/479 — peer guard + retry logic
* feat: apply PR #2339 missing commits (ced3305 + fe5a37c) Apply two Feb-19 commits from WhiskeySockets/Baileys PR #2339 that were missing from the initial integration: ced3305 — Prevent tctoken attachment to peer (AppStateSync) messages - Add isPeerMessage guard: additionalAttributes?.['category'] === 'peer' - Exclude peer messages from is1on1Send to fix error 479 (SmaxInvalid) which was causing hundreds of rejections on multi-device sync JIDs fe5a37c — Error 463 retry logic + blocking→non-blocking refactor - Replace blocking await getPrivacyTokens() with fire-and-forget .then/.catch so the send path is never delayed by a token fetch - Add tcTokenRetriedMsgIds Set in messages-recv.ts to track retried msgIds - On error 463 (MissingTcToken): wait 1.5s then relay the message once, allowing the server's tctoken notification to arrive before resend - Add retry_463_ok log event to baileys-logger for observability
This commit is contained in:
@@ -155,6 +155,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
const TC_TOKEN_INDEX_KEY = '__index'
|
const TC_TOKEN_INDEX_KEY = '__index'
|
||||||
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>()
|
||||||
let tcTokenIndexSaveTimer: ReturnType<typeof setTimeout> | undefined
|
let tcTokenIndexSaveTimer: ReturnType<typeof setTimeout> | undefined
|
||||||
let lastTcTokenPruneTs = 0
|
let lastTcTokenPruneTs = 0
|
||||||
|
|
||||||
@@ -1876,7 +1877,37 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
// device could not display the message
|
// device could not display the message
|
||||||
if(attrs.error) {
|
if(attrs.error) {
|
||||||
if(attrs.error === SERVER_ERROR_CODES.MissingTcToken) {
|
if(attrs.error === SERVER_ERROR_CODES.MissingTcToken) {
|
||||||
logTcToken('error_463', { jid: attrs.from, msgId: attrs.id })
|
const msgId = attrs.id
|
||||||
|
const jid = attrs.from
|
||||||
|
logTcToken('error_463', { jid, msgId })
|
||||||
|
|
||||||
|
// Single-retry: wait 1.5s for the server's tctoken notification to arrive,
|
||||||
|
// then resend. A Set prevents infinite retry loops.
|
||||||
|
if(msgId && jid && !tcTokenRetriedMsgIds.has(msgId)) {
|
||||||
|
tcTokenRetriedMsgIds.add(msgId)
|
||||||
|
// Safety cap — prevent unbounded memory growth
|
||||||
|
if(tcTokenRetriedMsgIds.size > 500) {
|
||||||
|
tcTokenRetriedMsgIds.clear()
|
||||||
|
}
|
||||||
|
|
||||||
|
;(async () => {
|
||||||
|
try {
|
||||||
|
await delay(1500)
|
||||||
|
const msg = await getMessage(key)
|
||||||
|
if(msg) {
|
||||||
|
await relayMessage(jid, msg, { messageId: msgId, useUserDevicesCache: true })
|
||||||
|
logTcToken('retry_463_ok', { jid, msgId })
|
||||||
|
} else {
|
||||||
|
logger.warn({ jid, msgId }, '463 retry: message not found in store')
|
||||||
|
ev.emit('messages.update', [{ key, update: { status: WAMessageStatus.ERROR, messageStubParameters: ['463'] } }])
|
||||||
|
}
|
||||||
|
} catch(err: any) {
|
||||||
|
logger.warn({ jid, msgId, err: err?.message }, '463 retry failed')
|
||||||
|
ev.emit('messages.update', [{ key, update: { status: WAMessageStatus.ERROR, messageStubParameters: ['463'] } }])
|
||||||
|
}
|
||||||
|
})()
|
||||||
|
return
|
||||||
|
}
|
||||||
} else if(attrs.error === SERVER_ERROR_CODES.SmaxInvalid) {
|
} else if(attrs.error === SERVER_ERROR_CODES.SmaxInvalid) {
|
||||||
logTcToken('error_479', { jid: attrs.from, msgId: attrs.id })
|
logTcToken('error_479', { jid: attrs.from, msgId: attrs.id })
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
+48
-63
@@ -119,6 +119,9 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
// Prevent race conditions in Signal session encryption by user
|
// Prevent race conditions in Signal session encryption by user
|
||||||
const encryptionMutex = makeKeyedMutex()
|
const encryptionMutex = makeKeyedMutex()
|
||||||
|
|
||||||
|
// Tracks JIDs with an in-flight getPrivacyTokens IQ to avoid duplicate concurrent fetches
|
||||||
|
const tcTokenFetchingJids = new Set<string>()
|
||||||
|
|
||||||
let mediaConn: Promise<MediaConnInfo>
|
let mediaConn: Promise<MediaConnInfo>
|
||||||
const refreshMediaConn = async (forceGet = false) => {
|
const refreshMediaConn = async (forceGet = false) => {
|
||||||
const media = await mediaConn
|
const media = await mediaConn
|
||||||
@@ -1482,8 +1485,9 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// tctoken lifecycle: fetch, validate expiry, proactive re-fetch if missing/expired
|
// tctoken lifecycle: fetch, validate expiry, proactive re-fetch if missing/expired
|
||||||
const is1on1Send = !isGroup && !isRetryResend && !isStatus && !isNewsletter
|
// WA Web never attaches tctoken to peer (AppStateSync) messages — server
|
||||||
let didFetchTcToken = false
|
const isPeerMessage = additionalAttributes?.['category'] === 'peer'
|
||||||
|
const is1on1Send = !isGroup && !isRetryResend && !isStatus && !isNewsletter && !isPeerMessage
|
||||||
|
|
||||||
// Resolve destination to LID for tctoken storage — matches Signal session key pattern
|
// Resolve destination to LID for tctoken storage — matches Signal session key pattern
|
||||||
const tcTokenJid = is1on1Send
|
const tcTokenJid = is1on1Send
|
||||||
@@ -1506,69 +1510,50 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
} catch { /* ignore cleanup errors */ }
|
} catch { /* ignore cleanup errors */ }
|
||||||
}
|
}
|
||||||
|
|
||||||
// If tctoken is missing or expired for a 1:1 send, proactively fetch it from the server
|
// If tctoken is missing for a 1:1 send, fire-and-forget fetch so the
|
||||||
if(!tcTokenBuffer?.length && is1on1Send) {
|
// retry path (error 463 → handleBadAck) can pick it up on resend
|
||||||
try {
|
if(!tcTokenBuffer?.length && is1on1Send && !tcTokenFetchingJids.has(tcTokenJid)) {
|
||||||
logTcToken('fetch', { jid: destinationJid })
|
tcTokenFetchingJids.add(tcTokenJid)
|
||||||
didFetchTcToken = true
|
logTcToken('fetch', { jid: destinationJid })
|
||||||
const fetchResult = await getPrivacyTokens([destinationJid])
|
getPrivacyTokens([destinationJid])
|
||||||
|
.then(async fetchResult => {
|
||||||
// Parse inline tokens from IQ result using the shared parser
|
await storeTcTokensFromIqResult({
|
||||||
// (includes monotonicity guard)
|
result: fetchResult,
|
||||||
await storeTcTokensFromIqResult({
|
fallbackJid: destinationJid,
|
||||||
result: fetchResult,
|
keys: authState.keys,
|
||||||
fallbackJid: destinationJid,
|
getLIDForPN: signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping),
|
||||||
keys: authState.keys,
|
onNewJidStored: (storedJid) => {
|
||||||
getLIDForPN: signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping),
|
// Fire-and-forget: persist JID into tctoken index for pruning
|
||||||
onNewJidStored: (storedJid) => {
|
(async () => {
|
||||||
// Fire-and-forget: persist JID into tctoken index for pruning
|
try {
|
||||||
(async () => {
|
const TC_IDX = '__index'
|
||||||
try {
|
const idxData = await authState.keys.get('tctoken', [TC_IDX])
|
||||||
const TC_IDX = '__index'
|
const idxEntry = idxData[TC_IDX]
|
||||||
const idxData = await authState.keys.get('tctoken', [TC_IDX])
|
const existing: string[] = idxEntry?.token
|
||||||
const idxEntry = idxData[TC_IDX]
|
? JSON.parse(Buffer.from(idxEntry.token).toString('utf8'))
|
||||||
const existing: string[] = idxEntry?.token
|
: []
|
||||||
? JSON.parse(Buffer.from(idxEntry.token).toString('utf8'))
|
if(!existing.includes(storedJid)) {
|
||||||
: []
|
existing.push(storedJid)
|
||||||
if(!existing.includes(storedJid)) {
|
await authState.keys.set({
|
||||||
existing.push(storedJid)
|
tctoken: {
|
||||||
await authState.keys.set({
|
[TC_IDX]: {
|
||||||
tctoken: {
|
token: Buffer.from(JSON.stringify(existing), 'utf8'),
|
||||||
[TC_IDX]: {
|
timestamp: unixTimestampSeconds().toString()
|
||||||
token: Buffer.from(JSON.stringify(existing), 'utf8'),
|
}
|
||||||
timestamp: unixTimestampSeconds().toString()
|
|
||||||
}
|
}
|
||||||
}
|
})
|
||||||
})
|
}
|
||||||
}
|
|
||||||
} catch { /* non-critical — index rebuilt on next pruning cycle */ }
|
} catch { /* non-critical — index rebuilt on next pruning cycle */ }
|
||||||
})()
|
})()
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
// Re-read from key store — the notification handler or inline
|
|
||||||
// parsing above may have stored the token
|
|
||||||
const refreshed = await authState.keys.get('tctoken', [tcTokenJid])
|
|
||||||
const refreshedEntry = refreshed[tcTokenJid]
|
|
||||||
tcTokenBuffer = refreshedEntry?.token
|
|
||||||
|
|
||||||
// The getPrivacyTokens IQ (type='set') also acts as issuance,
|
|
||||||
// so record senderTimestamp to prevent redundant fire-and-forget
|
|
||||||
// on the next message to this contact.
|
|
||||||
if(refreshedEntry?.token?.length) {
|
|
||||||
logTcToken('fetched', { jid: destinationJid })
|
|
||||||
await authState.keys.set({
|
|
||||||
tctoken: {
|
|
||||||
[tcTokenJid]: {
|
|
||||||
...refreshedEntry,
|
|
||||||
senderTimestamp: unixTimestampSeconds()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
})
|
||||||
} catch(err: any) {
|
.catch(err => {
|
||||||
logger.warn({ jid: destinationJid, trace: err?.stack }, 'failed to fetch privacy token before send')
|
logger.debug({ jid: destinationJid, err: err?.message }, 'fire-and-forget tctoken fetch failed')
|
||||||
}
|
})
|
||||||
|
.finally(() => {
|
||||||
|
tcTokenFetchingJids.delete(tcTokenJid)
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
if(tcTokenBuffer?.length) {
|
if(tcTokenBuffer?.length) {
|
||||||
@@ -1660,8 +1645,8 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
await sendNode(stanza)
|
await sendNode(stanza)
|
||||||
|
|
||||||
// Fire-and-forget: issue our token to the contact (like WA Web's sendTcToken)
|
// Fire-and-forget: issue our token to the contact (like WA Web's sendTcToken)
|
||||||
// Only for 1:1 sends where we didn't already fetch, and only when bucket boundary crossed
|
// Only for 1:1 sends where we already have a token, and when bucket boundary crossed
|
||||||
if(is1on1Send && !didFetchTcToken && shouldSendNewTcToken(existingTokenEntry?.senderTimestamp)) {
|
if(is1on1Send && tcTokenBuffer?.length && shouldSendNewTcToken(existingTokenEntry?.senderTimestamp)) {
|
||||||
const issueTimestamp = unixTimestampSeconds()
|
const issueTimestamp = unixTimestampSeconds()
|
||||||
logTcToken('reissue', { jid: destinationJid })
|
logTcToken('reissue', { jid: destinationJid })
|
||||||
// WA Web writes senderTimestamp only AFTER the IQ succeeds
|
// WA Web writes senderTimestamp only AFTER the IQ succeeds
|
||||||
|
|||||||
@@ -958,7 +958,7 @@ export function logLidMapping(
|
|||||||
* // Output: [BAILEYS] 🔑 TcToken expired → 5511999999999@s.whatsapp.net { age: 32d }
|
* // Output: [BAILEYS] 🔑 TcToken expired → 5511999999999@s.whatsapp.net { age: 32d }
|
||||||
*/
|
*/
|
||||||
export function logTcToken(
|
export function logTcToken(
|
||||||
event: 'stored' | 'expired' | 'fetch' | 'fetched' | 'reissue' | 'reissue_ok' | 'reissue_fail' | 'prune' | 'error_463' | 'error_479' | 'attached',
|
event: 'stored' | 'expired' | 'fetch' | 'fetched' | 'reissue' | 'reissue_ok' | 'reissue_fail' | 'prune' | 'error_463' | 'error_479' | 'attached' | 'retry_463_ok',
|
||||||
data?: Record<string, unknown>,
|
data?: Record<string, unknown>,
|
||||||
sessionName?: string
|
sessionName?: string
|
||||||
): void {
|
): void {
|
||||||
@@ -1004,6 +1004,9 @@ export function logTcToken(
|
|||||||
case 'error_479':
|
case 'error_479':
|
||||||
console.log(`${prefix} ⚠️ TcToken smax-invalid (479)${jid}${extraStr}`)
|
console.log(`${prefix} ⚠️ TcToken smax-invalid (479)${jid}${extraStr}`)
|
||||||
break
|
break
|
||||||
|
case 'retry_463_ok':
|
||||||
|
console.log(`${prefix} 🔄 TcToken retry 463 OK${jid}${extraStr}`)
|
||||||
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user