fix(tctoken): fix 3 bugs found during code review
fix(tctoken): fix 3 bugs found during code review
This commit is contained in:
@@ -1085,6 +1085,12 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Don't store timestamp-less tokens — they expire immediately and would
|
||||||
|
// corrupt a valid existing entry if one is already present
|
||||||
|
if(!incomingTs) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
await authState.keys.set({
|
await authState.keys.set({
|
||||||
tctoken: {
|
tctoken: {
|
||||||
[storageJid]: {
|
[storageJid]: {
|
||||||
@@ -1878,22 +1884,25 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
if(attrs.error) {
|
if(attrs.error) {
|
||||||
if(attrs.error === SERVER_ERROR_CODES.MissingTcToken) {
|
if(attrs.error === SERVER_ERROR_CODES.MissingTcToken) {
|
||||||
const msgId = attrs.id
|
const msgId = attrs.id
|
||||||
const jid = attrs.from
|
const jid = jidNormalizedUser(attrs.from)
|
||||||
logTcToken('error_463', { jid, msgId })
|
logTcToken('error_463', { jid, msgId })
|
||||||
|
|
||||||
// Single-retry: wait 1.5s for the server's tctoken notification to arrive,
|
// Single-retry: wait 1.5s for the server's tctoken notification to arrive,
|
||||||
// then resend. A Set prevents infinite retry loops.
|
// then resend. A Set prevents infinite retry loops.
|
||||||
if(msgId && jid && !tcTokenRetriedMsgIds.has(msgId)) {
|
// Composite key (jid:msgId) ensures retries are isolated per destination.
|
||||||
tcTokenRetriedMsgIds.add(msgId)
|
const retryKey = `${jid}:${msgId}`
|
||||||
// Safety cap — prevent unbounded memory growth
|
if(msgId && jid && !tcTokenRetriedMsgIds.has(retryKey)) {
|
||||||
if(tcTokenRetriedMsgIds.size > 500) {
|
tcTokenRetriedMsgIds.add(retryKey)
|
||||||
tcTokenRetriedMsgIds.clear()
|
// Each entry auto-expires after 60s — naturally bounded under normal use
|
||||||
}
|
setTimeout(() => tcTokenRetriedMsgIds.delete(retryKey), 60_000)
|
||||||
|
|
||||||
;(async () => {
|
;(async () => {
|
||||||
try {
|
try {
|
||||||
await delay(1500)
|
await delay(1500)
|
||||||
const msg = await getMessage(key)
|
const msg =
|
||||||
|
(await getMessage(key)) ??
|
||||||
|
// Fallback: ack can arrive <30ms after send, before store persists
|
||||||
|
messageRetryManager?.getRecentMessage(jid, msgId)?.message
|
||||||
if(msg) {
|
if(msg) {
|
||||||
await relayMessage(jid, msg, { messageId: msgId, useUserDevicesCache: true })
|
await relayMessage(jid, msg, { messageId: msgId, useUserDevicesCache: true })
|
||||||
logTcToken('retry_463_ok', { jid, msgId })
|
logTcToken('retry_463_ok', { jid, msgId })
|
||||||
@@ -2078,7 +2087,25 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
ev.on('connection.update', ({ isOnline }) => {
|
ev.on('connection.update', ({ isOnline, connection }) => {
|
||||||
|
// Flush pending tctoken index save on disconnect to avoid writing after close
|
||||||
|
if(connection === 'close' && tcTokenIndexSaveTimer) {
|
||||||
|
clearTimeout(tcTokenIndexSaveTimer)
|
||||||
|
tcTokenIndexSaveTimer = undefined
|
||||||
|
// Await index load first — prevents overwriting a more complete persisted index
|
||||||
|
// if the connection closes before the initial load finishes.
|
||||||
|
tcTokenIndexLoaded.then(() => {
|
||||||
|
Promise.resolve(authState.keys.set({
|
||||||
|
tctoken: {
|
||||||
|
[TC_TOKEN_INDEX_KEY]: {
|
||||||
|
token: Buffer.from(JSON.stringify([...tcTokenKnownJids]), 'utf8'),
|
||||||
|
timestamp: unixTimestampSeconds().toString()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})).catch(() => { /* non-critical */ })
|
||||||
|
}).catch(() => { /* non-critical */ })
|
||||||
|
}
|
||||||
|
|
||||||
if(typeof isOnline !== 'undefined') {
|
if(typeof isOnline !== 'undefined') {
|
||||||
sendActiveReceipts = isOnline
|
sendActiveReceipts = isOnline
|
||||||
logger.trace(`sendActiveReceipts set to "${sendActiveReceipts}"`)
|
logger.trace(`sendActiveReceipts set to "${sendActiveReceipts}"`)
|
||||||
|
|||||||
+12
-27
@@ -1521,31 +1521,8 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
result: fetchResult,
|
result: fetchResult,
|
||||||
fallbackJid: destinationJid,
|
fallbackJid: destinationJid,
|
||||||
keys: authState.keys,
|
keys: authState.keys,
|
||||||
getLIDForPN: signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping),
|
getLIDForPN: signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
||||||
onNewJidStored: (storedJid) => {
|
// onNewJidStored not passed — the pruning index lives in messages-recv (higher layer)
|
||||||
// Fire-and-forget: persist JID into tctoken index for pruning
|
|
||||||
(async () => {
|
|
||||||
try {
|
|
||||||
const TC_IDX = '__index'
|
|
||||||
const idxData = await authState.keys.get('tctoken', [TC_IDX])
|
|
||||||
const idxEntry = idxData[TC_IDX]
|
|
||||||
const existing: string[] = idxEntry?.token
|
|
||||||
? JSON.parse(Buffer.from(idxEntry.token).toString('utf8'))
|
|
||||||
: []
|
|
||||||
if(!existing.includes(storedJid)) {
|
|
||||||
existing.push(storedJid)
|
|
||||||
await authState.keys.set({
|
|
||||||
tctoken: {
|
|
||||||
[TC_IDX]: {
|
|
||||||
token: Buffer.from(JSON.stringify(existing), 'utf8'),
|
|
||||||
timestamp: unixTimestampSeconds().toString()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
})
|
|
||||||
}
|
|
||||||
} catch { /* non-critical — index rebuilt on next pruning cycle */ }
|
|
||||||
})()
|
|
||||||
}
|
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
.catch(err => {
|
.catch(err => {
|
||||||
@@ -1654,7 +1631,15 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
// This ensures failed issuance allows re-issuance on the next message
|
// This ensures failed issuance allows re-issuance on the next message
|
||||||
// rather than blocking it for up to 7 days (one bucket duration).
|
// rather than blocking it for up to 7 days (one bucket duration).
|
||||||
getPrivacyTokens([destinationJid], issueTimestamp)
|
getPrivacyTokens([destinationJid], issueTimestamp)
|
||||||
.then(async () => {
|
.then(async result => {
|
||||||
|
// Store any tokens received in the IQ response
|
||||||
|
await storeTcTokensFromIqResult({
|
||||||
|
result,
|
||||||
|
fallbackJid: tcTokenJid,
|
||||||
|
keys: authState.keys,
|
||||||
|
getLIDForPN: signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
||||||
|
})
|
||||||
|
|
||||||
// Re-read entry to avoid overwriting concurrent notification handler updates
|
// Re-read entry to avoid overwriting concurrent notification handler updates
|
||||||
const currentData = await authState.keys.get('tctoken', [tcTokenJid])
|
const currentData = await authState.keys.get('tctoken', [tcTokenJid])
|
||||||
const currentEntry = currentData[tcTokenJid]
|
const currentEntry = currentData[tcTokenJid]
|
||||||
@@ -1701,7 +1686,7 @@ export const makeMessagesSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
// Add message to retry cache if enabled
|
// Add message to retry cache if enabled
|
||||||
if (messageRetryManager && !participant) {
|
if (messageRetryManager && !participant) {
|
||||||
messageRetryManager.addRecentMessage(destinationJid, msgId, message)
|
messageRetryManager.addRecentMessage(jidNormalizedUser(destinationJid), msgId, message)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Track session activity for cleanup (all target JIDs)
|
// Track session activity for cleanup (all target JIDs)
|
||||||
|
|||||||
@@ -148,6 +148,12 @@ export async function storeTcTokensFromIqResult({
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Don't store timestamp-less tokens at all — isTcTokenExpired treats them
|
||||||
|
// as immediately expired regardless of whether an existing entry is present
|
||||||
|
if(!incomingTs) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
await keys.set({
|
await keys.set({
|
||||||
tctoken: {
|
tctoken: {
|
||||||
[storageJid]: {
|
[storageJid]: {
|
||||||
|
|||||||
Reference in New Issue
Block a user