Compare commits
16 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| dc24e60e5b | |||
| 942f4ae2c9 | |||
| 98d2de2984 | |||
| b0c34cb536 | |||
| d169a6f755 | |||
| aef4973cac | |||
| b90b383811 | |||
| 35461f96c5 | |||
| 1e37448e05 | |||
| 72e78a12e0 | |||
| a5b7ce7ef9 | |||
| df1acc8f0c | |||
| 1dffe3b311 | |||
| 4fe708445a | |||
| 336ed64a44 | |||
| 65cb09e4df |
@@ -1,7 +1,7 @@
|
|||||||
syntax = "proto3";
|
syntax = "proto3";
|
||||||
package proto;
|
package proto;
|
||||||
|
|
||||||
/// WhatsApp Version: 2.3000.1038024963
|
/// WhatsApp Version: 2.3000.1038164556
|
||||||
|
|
||||||
message ADVDeviceIdentity {
|
message ADVDeviceIdentity {
|
||||||
optional uint32 rawId = 1;
|
optional uint32 rawId = 1;
|
||||||
|
|||||||
@@ -1 +1 @@
|
|||||||
{"version":[2,3000,1038147544]}
|
{"version":[2,3000,1038167900]}
|
||||||
|
|||||||
+150
-24
@@ -54,7 +54,7 @@ import {
|
|||||||
resolveLidToPn
|
resolveLidToPn
|
||||||
} from '../Utils'
|
} from '../Utils'
|
||||||
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
|
import { makeKeyedMutex, makeMutex } from '../Utils/make-mutex'
|
||||||
import processMessage from '../Utils/process-message'
|
import processMessage, { getChatId } from '../Utils/process-message'
|
||||||
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
import { buildTcTokenFromJid } from '../Utils/tc-token-utils'
|
||||||
import {
|
import {
|
||||||
type BinaryNode,
|
type BinaryNode,
|
||||||
@@ -130,6 +130,18 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
/** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */
|
/** this mutex ensures that notifications from the same chat are processed in order, while allowing parallel processing across chats */
|
||||||
const notificationMutex = makeKeyedMutex()
|
const notificationMutex = makeKeyedMutex()
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Per-chat mutex dedicated to post-upsert work (history app-state sync +
|
||||||
|
* processMessage side effects). Kept separate from `messageMutex` because
|
||||||
|
* the inbound caller already holds `messageMutex(chatId)` while running
|
||||||
|
* decrypt + upsertMessage; sharing the same mutex would let a concurrently-
|
||||||
|
* arrived message N+1 enqueue *between* msg N's outer callback and msg N's
|
||||||
|
* post-upsert task, so msg N+1's processMessage could run before msg N's
|
||||||
|
* (breaking per-chat ordering of side effects). With a separate mutex,
|
||||||
|
* post-upsert tasks enqueue strictly in upsertMessage call order.
|
||||||
|
*/
|
||||||
|
const postUpsertMutex = makeKeyedMutex()
|
||||||
|
|
||||||
// Timeout for AwaitingInitialSync state
|
// Timeout for AwaitingInitialSync state
|
||||||
let awaitingSyncTimeout: NodeJS.Timeout | undefined
|
let awaitingSyncTimeout: NodeJS.Timeout | undefined
|
||||||
|
|
||||||
@@ -1399,7 +1411,19 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
blockedCollections.clear()
|
blockedCollections.clear()
|
||||||
|
|
||||||
logger.info('Doing app state sync')
|
logger.info('Doing app state sync')
|
||||||
await resyncAppState(ALL_WA_PATCH_NAMES, true)
|
try {
|
||||||
|
await resyncAppState(ALL_WA_PATCH_NAMES, true)
|
||||||
|
} catch (err) {
|
||||||
|
// Failure recovery: without this, syncState would stay at Syncing
|
||||||
|
// and ev.flush() would never run, leaving the event buffer pinned
|
||||||
|
// until the buffer's own safety timeout expires. Force the state
|
||||||
|
// machine forward so live inbound events can flow even if the
|
||||||
|
// app-state resync failed (collections are already cleared, so
|
||||||
|
// blocked patches will be retried on the next creds.update tick).
|
||||||
|
syncState = SyncState.Online
|
||||||
|
ev.flush()
|
||||||
|
throw err
|
||||||
|
}
|
||||||
|
|
||||||
// Sync is complete, go online and flush everything
|
// Sync is complete, go online and flush everything
|
||||||
syncState = SyncState.Online
|
syncState = SyncState.Online
|
||||||
@@ -1411,29 +1435,131 @@ export const makeChatsSocket = (config: SocketConfig) => {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
await Promise.all([
|
// Post-upsert work: history app-state sync + processMessage side effects.
|
||||||
(async () => {
|
// Awaiting here keeps `messages.upsert` pinned in the event buffer
|
||||||
if (shouldProcessHistoryMsg) {
|
// (createBufferedFunction only schedules flush after work() resolves), so
|
||||||
await doAppStateSync()
|
// the hot path detaches this work to release the emit on the next debounce
|
||||||
}
|
// tick.
|
||||||
})(),
|
//
|
||||||
processMessage(msg, {
|
// Use Promise.allSettled so the combined promise only settles after BOTH
|
||||||
signalRepository,
|
// tasks finish. With a plain Promise.all, an early rejection from one task
|
||||||
shouldProcessHistoryMsg,
|
// would release the keyed mutex while the other task is still mutating
|
||||||
placeholderResendCache,
|
// chat state — letting the next message of the same chat overtake it and
|
||||||
ev,
|
// break per-chat ordering.
|
||||||
creds: authState.creds,
|
//
|
||||||
keyStore: authState.keys,
|
// Returns the per-task settle status so the keyShare branch can know
|
||||||
logger,
|
// whether processMessage actually persisted the new app-state-sync key
|
||||||
options: config.options,
|
// before triggering doAppStateSync (otherwise the sync would hit
|
||||||
getMessage
|
// isMissingKeyError and park collections in blockedCollections).
|
||||||
})
|
const postUpsertTasks = async (): Promise<{ processMessageOk: boolean }> => {
|
||||||
])
|
const [historyResult, processResult] = await Promise.allSettled([
|
||||||
|
shouldProcessHistoryMsg ? doAppStateSync() : Promise.resolve(),
|
||||||
|
processMessage(msg, {
|
||||||
|
signalRepository,
|
||||||
|
shouldProcessHistoryMsg,
|
||||||
|
placeholderResendCache,
|
||||||
|
ev,
|
||||||
|
creds: authState.creds,
|
||||||
|
keyStore: authState.keys,
|
||||||
|
logger,
|
||||||
|
options: config.options,
|
||||||
|
getMessage
|
||||||
|
})
|
||||||
|
])
|
||||||
|
|
||||||
// If the app state key arrives and we are waiting to sync, trigger the sync now.
|
if (historyResult.status === 'rejected') {
|
||||||
if (msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing) {
|
logger?.warn(
|
||||||
logger.info('App state sync key arrived, triggering app state sync')
|
{ err: historyResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
||||||
await doAppStateSync()
|
'history doAppStateSync failed'
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
if (processResult.status === 'rejected') {
|
||||||
|
logger?.warn(
|
||||||
|
{ err: processResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
||||||
|
'processMessage failed'
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
return { processMessageOk: processResult.status === 'fulfilled' }
|
||||||
|
}
|
||||||
|
|
||||||
|
// Use getChatId + jidNormalizedUser so the mutex key matches the chat-id
|
||||||
|
// scheme processMessage uses for chat updates (broadcasts target the
|
||||||
|
// participant). When getChatId/jidNormalizedUser yields nothing usable
|
||||||
|
// (missing or malformed JID), prefer a message-derived fallback over a
|
||||||
|
// single global 'unknown' bucket — that bucket would head-of-line block
|
||||||
|
// every malformed message behind a shared queue. msg.key.id is unique
|
||||||
|
// per message so unrelated malformed inputs no longer serialize together;
|
||||||
|
// for valid messages we still hit the normalized chat-id path, so the
|
||||||
|
// per-chat ordering guarantee is unchanged where it matters.
|
||||||
|
const rawChatId = getChatId(msg.key)
|
||||||
|
const normalizedChatId = rawChatId ? jidNormalizedUser(rawChatId) : ''
|
||||||
|
const postUpsertChatId = normalizedChatId || msg.key?.id || 'unknown'
|
||||||
|
|
||||||
|
// Wrap in `postUpsertMutex(chatId)` (a SEPARATE keyed mutex from the outer
|
||||||
|
// `messageMutex` held by the inbound caller) so per-chat ordering of
|
||||||
|
// processMessage side effects (chat.unreadCount, LID/PN mapping,
|
||||||
|
// messages.update, history downloads) is preserved across messages of the
|
||||||
|
// same chat.
|
||||||
|
//
|
||||||
|
// Why a separate mutex: if we re-used messageMutex, a concurrently-arrived
|
||||||
|
// message N+1 could enqueue on the outer mutex BEFORE msg N's post-upsert
|
||||||
|
// task gets enqueued (because N's outer callback yields on `await decrypt()`
|
||||||
|
// before reaching the inner enqueue site). The queue would then be
|
||||||
|
// [OuterN+1, InnerN, ...], so InnerN+1 would beat InnerN to processMessage.
|
||||||
|
// With its own mutex, post-upsert tasks enqueue strictly in upsertMessage
|
||||||
|
// call order (which IS message arrival order because the outer
|
||||||
|
// messageMutex serializes the upserts per-chat).
|
||||||
|
const postUpsertWork = postUpsertMutex.mutex(postUpsertChatId, postUpsertTasks)
|
||||||
|
|
||||||
|
const isKeyShareDuringSync =
|
||||||
|
!!msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing
|
||||||
|
|
||||||
|
if (isKeyShareDuringSync) {
|
||||||
|
// appStateSyncKeyShare path: processMessage persists the new app-state-sync
|
||||||
|
// key in its APP_STATE_SYNC_KEY_SHARE handler (via keyStore.transaction).
|
||||||
|
// The follow-up doAppStateSync() needs that key to decrypt patches, so
|
||||||
|
// we MUST wait for processMessage to actually succeed before kicking off
|
||||||
|
// the sync — otherwise it would hit isMissingKeyError and park
|
||||||
|
// collections in blockedCollections, regressing the very issue this
|
||||||
|
// branch was added to fix.
|
||||||
|
//
|
||||||
|
// No deadlock with the inbound caller's messageMutex because
|
||||||
|
// postUpsertMutex is a different mutex instance.
|
||||||
|
logger.info('App state sync key arrived, awaiting persistence before triggering sync')
|
||||||
|
const { processMessageOk } = await postUpsertWork
|
||||||
|
|
||||||
|
if (!processMessageOk) {
|
||||||
|
logger?.warn(
|
||||||
|
{ messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
||||||
|
'processMessage failed during key-share — skipping doAppStateSync to avoid isMissingKeyError'
|
||||||
|
)
|
||||||
|
} else {
|
||||||
|
try {
|
||||||
|
await doAppStateSync()
|
||||||
|
} catch (err) {
|
||||||
|
logger?.warn(
|
||||||
|
{ err, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
|
||||||
|
'doAppStateSync failed after key-share persistence'
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
// `postUpsertWork` is not expected to reject — `Promise.allSettled`
|
||||||
|
// inside `postUpsertTasks` never rejects, and per-task failures are
|
||||||
|
// already logged inline. The defensive catch routes any truly
|
||||||
|
// unexpected rejection (e.g. `postUpsertMutex` internal corruption,
|
||||||
|
// future synchronous throws inside processMessage) through
|
||||||
|
// `onUnexpectedError` instead of letting it surface as an
|
||||||
|
// UnhandledPromiseRejection — which on Node ≥15 can terminate the
|
||||||
|
// long-running socket process.
|
||||||
|
postUpsertWork.catch(err =>
|
||||||
|
onUnexpectedError(
|
||||||
|
err,
|
||||||
|
`processing post-upsert work for message ${msg.key?.id || 'unknown'} on ${msg.key?.remoteJid || 'unknown chat'}`
|
||||||
|
)
|
||||||
|
)
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
+39
-29
@@ -2330,44 +2330,54 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
)
|
)
|
||||||
|
|
||||||
const alt = msg.key.participantAlt || msg.key.remoteJidAlt
|
const alt = msg.key.participantAlt || msg.key.remoteJidAlt
|
||||||
// Handle LID/PN mappings with hybrid approach:
|
// Handle LID/PN mappings with optimized hot-path:
|
||||||
// - Store mapping operation runs in background (non-critical for decrypt)
|
// - storeLIDPNMappings is fire-and-forget (background) — does NOT block decrypt
|
||||||
// - Session migration MUST complete before decrypt() to avoid "No session record" errors
|
// - migrateSession is SYNC (await) — REQUIRED for decrypt to find session
|
||||||
// This addresses Codex/Copilot review concerns about race conditions with decrypt()
|
//
|
||||||
|
// SAFETY: normalizeMessageJids has a fast-path that uses key.*Alt directly without
|
||||||
|
// hitting the store, so the just-arrived message normalizes correctly even before
|
||||||
|
// the background store completes. Subsequent messages in the same chat hit the
|
||||||
|
// store after the background write is done (ms later).
|
||||||
|
//
|
||||||
|
// Pre-check (getPNForLID/getLIDForPN) was removed — storeLIDPNMappings has internal
|
||||||
|
// LRU cache + dedup, the pre-check was a redundant store round-trip per inbound
|
||||||
|
// message that added latency under load.
|
||||||
|
//
|
||||||
|
// HISTORICAL: this restores the intent of d73cd28d39 (2026-02-03) which was
|
||||||
|
// partially reverted by c3fc792351 the same day due to a race-condition concern
|
||||||
|
// with migrateSession (kept sync here). storeLIDPNMappings was over-protected:
|
||||||
|
// it persists a mapping that downstream consumers can re-derive from key.*Alt,
|
||||||
|
// while migrateSession actually moves the Signal session record that decrypt()
|
||||||
|
// will load microseconds later — those two have very different criticality.
|
||||||
|
//
|
||||||
|
// DO NOT make migrateSession async — decrypt() depends on the session being at
|
||||||
|
// the correct identifier (LID vs PN) when it runs. Other code paths (USync
|
||||||
|
// device lookup in messages-send.ts) create LID/PN mappings without migrating
|
||||||
|
// the session, so we cannot skip migration even when the mapping already exists.
|
||||||
if (!!alt) {
|
if (!!alt) {
|
||||||
const altServer = jidDecode(alt)?.server
|
const altServer = jidDecode(alt)?.server
|
||||||
const primaryJid = msg.key.participant || msg.key.remoteJid!
|
const primaryJid = msg.key.participant || msg.key.remoteJid!
|
||||||
|
|
||||||
if (altServer === 'lid') {
|
if (altServer === 'lid') {
|
||||||
// Check if mapping already exists to avoid unnecessary storage operations
|
// Fire-and-forget: storeLIDPNMappings has internal cache+dedup,
|
||||||
const existingMapping = await signalRepository.lidMapping.getPNForLID(alt)
|
// pre-check (getPNForLID) was redundant.
|
||||||
if (!existingMapping) {
|
signalRepository.lidMapping
|
||||||
// MUST await: normalizeMessageJids() runs after this and needs the mapping
|
.storeLIDPNMappings([{ lid: alt, pn: primaryJid }])
|
||||||
// in the LIDMappingStore to resolve LID→PN for events delivered to consumers
|
.catch(error => logger.warn({ error, alt, primaryJid }, 'background LID mapping store failed'))
|
||||||
await signalRepository.lidMapping
|
|
||||||
.storeLIDPNMappings([{ lid: alt, pn: primaryJid }])
|
|
||||||
.catch(error => logger.warn({ error, alt, primaryJid }, 'LID mapping storage failed'))
|
|
||||||
}
|
|
||||||
|
|
||||||
// CRITICAL: ALWAYS migrate session, even if mapping exists
|
// CRITICAL: ALWAYS migrate session SYNC, even if mapping exists.
|
||||||
// Other code paths (e.g., USync device lookup in messages-send.ts:310-319)
|
// Other code paths (e.g., USync device lookup in messages-send.ts) may create
|
||||||
// may create mappings via storeLIDPNMappings() without calling migrateSession()
|
// mappings via storeLIDPNMappings() without calling migrateSession(). This
|
||||||
// This leaves sessions under PN format while decrypt() expects LID format
|
// leaves sessions under PN format while decrypt() expects LID format.
|
||||||
// Skipping migration based on mapping existence causes "No session record" errors
|
// Skipping migration based on mapping existence causes "No session record" errors.
|
||||||
await signalRepository.migrateSession(primaryJid, alt)
|
await signalRepository.migrateSession(primaryJid, alt)
|
||||||
} else {
|
} else {
|
||||||
// Check if reverse mapping exists
|
// Fire-and-forget: same rationale as above.
|
||||||
const existingMapping = await signalRepository.lidMapping.getLIDForPN(alt)
|
signalRepository.lidMapping
|
||||||
if (!existingMapping) {
|
.storeLIDPNMappings([{ lid: primaryJid, pn: alt }])
|
||||||
// MUST await: normalizeMessageJids() runs after this and needs the mapping
|
.catch(error => logger.warn({ error, alt, primaryJid }, 'background LID mapping store failed'))
|
||||||
// in the LIDMappingStore to resolve LID→PN for events delivered to consumers
|
|
||||||
await signalRepository.lidMapping
|
|
||||||
.storeLIDPNMappings([{ lid: primaryJid, pn: alt }])
|
|
||||||
.catch(error => logger.warn({ error, alt, primaryJid }, 'LID mapping storage failed'))
|
|
||||||
}
|
|
||||||
|
|
||||||
// CRITICAL: ALWAYS migrate session, even if mapping exists
|
// CRITICAL: ALWAYS migrate session SYNC.
|
||||||
// Same reasoning as above - mapping existence doesn't guarantee session migration
|
|
||||||
await signalRepository.migrateSession(alt, primaryJid)
|
await signalRepository.migrateSession(alt, primaryJid)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+12
-1
@@ -341,7 +341,18 @@ export const addTransactionCapability = (
|
|||||||
|
|
||||||
return result
|
return result
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
logger.error({ error }, 'transaction failed, rolling back')
|
// SessionError is part of the normal Bad MAC recovery flow
|
||||||
|
// (retry receipt → sender resends as pkmsg → new session within ~1.3s).
|
||||||
|
// Logging it as ERROR creates 2 noise lines per recoverable Bad MAC cycle.
|
||||||
|
// Downgrade to debug for SessionError; keep ERROR for everything else.
|
||||||
|
// The error is still re-thrown — recovery behavior is unchanged.
|
||||||
|
const errName = (error as { name?: string })?.name
|
||||||
|
if (errName === 'SessionError') {
|
||||||
|
logger.debug({ error }, 'transaction failed (SessionError — recoverable via retry receipt)')
|
||||||
|
} else {
|
||||||
|
logger.error({ error }, 'transaction failed, rolling back')
|
||||||
|
}
|
||||||
|
|
||||||
throw error
|
throw error
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -58,7 +58,11 @@ export const BAD_MAC_ERROR_TEXT = 'Bad MAC'
|
|||||||
export const DECRYPTION_RETRY_CONFIG = {
|
export const DECRYPTION_RETRY_CONFIG = {
|
||||||
maxRetries: 3,
|
maxRetries: 3,
|
||||||
baseDelayMs: 100,
|
baseDelayMs: 100,
|
||||||
sessionRecordErrors: ['No session record', 'SessionError: No session record'],
|
// 'No matching sessions found' is the libsignal error when decryptWithSessions exhausts
|
||||||
|
// all stored sessions for a JID. Same recovery flow (retry receipt → pkmsg → new session)
|
||||||
|
// — categorise it as session-record so the caller logs DEBUG on retry, ERROR only when
|
||||||
|
// retries are exhausted (instead of dumping the full stack as an unknown error).
|
||||||
|
sessionRecordErrors: ['No session record', 'SessionError: No session record', 'No matching sessions found'],
|
||||||
corruptedSessionErrors: ['Bad MAC', 'MessageCounterError', MISSING_KEYS_ERROR_TEXT]
|
corruptedSessionErrors: ['Bad MAC', 'MessageCounterError', MISSING_KEYS_ERROR_TEXT]
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -421,9 +425,26 @@ export const decryptMessageNode = (
|
|||||||
const isCorrupted = isCorruptedSessionError(originalError)
|
const isCorrupted = isCorruptedSessionError(originalError)
|
||||||
const isSessionRecord = isSessionRecordError(originalError)
|
const isSessionRecord = isSessionRecordError(originalError)
|
||||||
|
|
||||||
|
// Slim error projection — keep name/message/type for diagnosis,
|
||||||
|
// drop `stack` which adds 4-5 lines of node_modules paths per log
|
||||||
|
// for known-recoverable libsignal errors.
|
||||||
|
//
|
||||||
|
// CRITICAL: only slim for KNOWN-RECOVERABLE categories (corrupted /
|
||||||
|
// session-record). The unknown-error branch keeps the full Error so
|
||||||
|
// protobuf/parsing/runtime bugs still emit a stack trace where it
|
||||||
|
// matters most. Catches Copilot/Codex P2 review on PR #391.
|
||||||
|
const slimErr = originalError
|
||||||
|
? {
|
||||||
|
name: (originalError as { name?: string }).name,
|
||||||
|
message: (originalError as { message?: string }).message,
|
||||||
|
type: (originalError as { type?: string }).type
|
||||||
|
}
|
||||||
|
: undefined
|
||||||
|
const isRecoverableCategory = isCorrupted || isSessionRecord
|
||||||
|
|
||||||
const errorContext = {
|
const errorContext = {
|
||||||
key: fullMessage.key,
|
key: fullMessage.key,
|
||||||
err: originalError,
|
err: isRecoverableCategory ? slimErr : originalError,
|
||||||
messageType: tag === 'plaintext' ? 'plaintext' : attrs.type,
|
messageType: tag === 'plaintext' ? 'plaintext' : attrs.type,
|
||||||
sender,
|
sender,
|
||||||
author,
|
author,
|
||||||
|
|||||||
+113
-22
@@ -40,7 +40,7 @@ import { getKeyAuthor, toNumber } from './generics'
|
|||||||
import { downloadAndProcessHistorySyncNotification } from './history'
|
import { downloadAndProcessHistorySyncNotification } from './history'
|
||||||
import type { ILogger } from './logger'
|
import type { ILogger } from './logger'
|
||||||
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
||||||
import { buildMergedTcTokenIndexWrite, resolveTcTokenJid } from './tc-token-utils'
|
import { buildMergedTcTokenIndexWrite } from './tc-token-utils'
|
||||||
|
|
||||||
type ProcessMessageContext = {
|
type ProcessMessageContext = {
|
||||||
shouldProcessHistoryMsg: boolean
|
shouldProcessHistoryMsg: boolean
|
||||||
@@ -82,32 +82,106 @@ const REAL_MSG_REQ_ME_STUB_TYPES = new Set([WAMessageStubType.GROUP_PARTICIPANT_
|
|||||||
* (TC_TOKEN_INDEX_KEY) via buildMergedTcTokenIndexWrite, so the 24h prune sweep in
|
* (TC_TOKEN_INDEX_KEY) via buildMergedTcTokenIndexWrite, so the 24h prune sweep in
|
||||||
* messages-recv picks them up across sessions.
|
* messages-recv picks them up across sessions.
|
||||||
*/
|
*/
|
||||||
|
/**
|
||||||
|
* Single-concurrency queue for `storeTcTokensFromHistorySync` calls.
|
||||||
|
*
|
||||||
|
* Why: the function does read-then-write merges (`keyStore.get('tctoken', ...)` →
|
||||||
|
* compute → `keyStore.set(...)`) which are NOT atomic at the store level. If two
|
||||||
|
* history-sync chunks invoke this concurrently (common during reconnect / QR
|
||||||
|
* scan), an older chunk that started first can `keyStore.set` AFTER a newer
|
||||||
|
* chunk, overwriting the newer entry — and worse, the merged `__index` write
|
||||||
|
* can drop JIDs the other chunk just added. Result: stale tcTokens / repeat 463
|
||||||
|
* sends until the next opportunistic refetch.
|
||||||
|
*
|
||||||
|
* Serialising via a chained Promise keeps the runs ordered while still freeing
|
||||||
|
* the calling `processMessage` to emit `messaging-history.set` immediately
|
||||||
|
* (the chain is fire-and-forget at the call site). Errors don't break the chain
|
||||||
|
* — each `catch` resets it to `Promise.resolve()` so a single failure can't
|
||||||
|
* stall future runs.
|
||||||
|
*
|
||||||
|
* The chain is module-scoped (one per Node process). Multiple Baileys instances
|
||||||
|
* sharing this module will serialise across instances too, but their writes
|
||||||
|
* target different keyStores so there's no correctness gain — only a tiny loss
|
||||||
|
* of inter-instance parallelism for tcToken syncs, which is acceptable given
|
||||||
|
* how rarely this runs vs. how rare cross-instance contention is.
|
||||||
|
*/
|
||||||
|
let historyTcTokenChain: Promise<void> = Promise.resolve()
|
||||||
|
|
||||||
|
function scheduleHistoryTcTokenSync(
|
||||||
|
chats: Chat[],
|
||||||
|
signalRepository: SignalRepositoryWithLIDStore,
|
||||||
|
keyStore: SignalKeyStoreWithTransaction,
|
||||||
|
logger?: ILogger
|
||||||
|
): void {
|
||||||
|
historyTcTokenChain = historyTcTokenChain
|
||||||
|
.catch(() => {
|
||||||
|
/* swallow prior error so chain stays alive */
|
||||||
|
})
|
||||||
|
.then(() => storeTcTokensFromHistorySync(chats, signalRepository, keyStore, logger))
|
||||||
|
.catch(err => {
|
||||||
|
logger?.warn({ err }, 'background tctoken history-sync persistence failed')
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
async function storeTcTokensFromHistorySync(
|
async function storeTcTokensFromHistorySync(
|
||||||
chats: Chat[],
|
chats: Chat[],
|
||||||
signalRepository: SignalRepositoryWithLIDStore,
|
signalRepository: SignalRepositoryWithLIDStore,
|
||||||
keyStore: SignalKeyStoreWithTransaction,
|
keyStore: SignalKeyStoreWithTransaction,
|
||||||
logger?: ILogger
|
logger?: ILogger
|
||||||
) {
|
) {
|
||||||
const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
// Cheap filter first — most chats in a sync chunk don't carry tcToken at all,
|
||||||
|
// and we want to avoid spinning up promises for them.
|
||||||
const candidates: { storageJid: string; token: Buffer; ts: number; senderTs?: number }[] = []
|
const tokenChats = chats.filter(chat => {
|
||||||
for (const chat of chats) {
|
|
||||||
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
|
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
|
||||||
if (chat.tcToken?.length && ts > 0) {
|
return !!chat.tcToken?.length && ts > 0
|
||||||
const jid = jidNormalizedUser(chat.id!)
|
})
|
||||||
const storageJid = await resolveTcTokenJid(jid, getLIDForPN)
|
|
||||||
candidates.push({
|
if (!tokenChats.length) {
|
||||||
storageJid,
|
return
|
||||||
token: Buffer.from(chat.tcToken),
|
}
|
||||||
ts,
|
|
||||||
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
// Pre-normalize so the rest of the pipeline is a synchronous join.
|
||||||
})
|
const normalized = tokenChats.map(chat => ({
|
||||||
|
chat,
|
||||||
|
ts: toNumber(chat.tcTokenTimestamp!),
|
||||||
|
jid: jidNormalizedUser(chat.id!)
|
||||||
|
}))
|
||||||
|
|
||||||
|
// BATCHED LID resolution. The previous shape called getLIDForPN once per
|
||||||
|
// chat (sequential await inside a for-of), which became the bottleneck
|
||||||
|
// during heavy history sync — every cold-cache hit was a DB round-trip,
|
||||||
|
// stalling messaging-history.set and spilling into the event-buffer.
|
||||||
|
// `getLIDsForPNs` resolves a deduped list in ONE batched query (and shares
|
||||||
|
// USync retry across PNs that miss cache), turning O(N) round-trips into 1.
|
||||||
|
//
|
||||||
|
// LID inputs (and `@hosted.lid`) skip the lookup entirely — they're already
|
||||||
|
// the storage form. Failures degrade gracefully: a missing mapping just
|
||||||
|
// stores under the original jid, matching `resolveTcTokenJid`'s null branch.
|
||||||
|
const pnsToResolve = [...new Set(normalized.filter(({ jid }) => !isLidUser(jid)).map(({ jid }) => jid))]
|
||||||
|
const pnToLid = new Map<string, string>()
|
||||||
|
|
||||||
|
if (pnsToResolve.length) {
|
||||||
|
try {
|
||||||
|
const mappings = await signalRepository.lidMapping.getLIDsForPNs(pnsToResolve)
|
||||||
|
// Flat loop (continue-on-skip) keeps max nesting depth at 4 for lint.
|
||||||
|
for (const { pn, lid } of mappings ?? []) {
|
||||||
|
if (!pn || !lid) continue
|
||||||
|
pnToLid.set(jidNormalizedUser(pn), lid)
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
// Per-chat fallback below (storageJid := jid). Don't abort the chunk —
|
||||||
|
// CodeRabbit noted that all-or-nothing rejection here would drop every
|
||||||
|
// tctoken in the batch AND prevent messaging-history.set from firing.
|
||||||
|
logger?.warn({ err }, 'storeTcTokensFromHistorySync: getLIDsForPNs batch failed; falling back to per-chat jid')
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!candidates.length) {
|
const candidates = normalized.map(({ chat, ts, jid }) => ({
|
||||||
return
|
storageJid: pnToLid.get(jid) ?? jid,
|
||||||
}
|
token: Buffer.from(chat.tcToken!),
|
||||||
|
ts,
|
||||||
|
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
||||||
|
}))
|
||||||
|
|
||||||
const jids = candidates.map(c => c.storageJid)
|
const jids = candidates.map(c => c.storageJid)
|
||||||
const existing = await keyStore.get('tctoken', jids)
|
const existing = await keyStore.get('tctoken', jids)
|
||||||
@@ -532,11 +606,28 @@ const processMessage = async (
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Persist tctokens carried by history-sync chats BEFORE emitting messaging-history.set
|
// Persist tctokens carried by history-sync chats in BACKGROUND, serialised.
|
||||||
// — listeners may immediately fire outbound sends that need the tctoken, and the store
|
//
|
||||||
// has to be populated first to avoid an error 463 on the first multi-device send.
|
// Originally awaited (PR #386) to avoid 463 on first multi-device send, but in
|
||||||
// Runs AFTER storeLIDPNMappings (see comment above) so LID resolution works.
|
// production this drained the event buffer per-chunk and added visible delivery
|
||||||
await storeTcTokensFromHistorySync(data.chats, signalRepository, keyStore, logger)
|
// latency (especially after restart / QR scan when many chunks arrived at once).
|
||||||
|
//
|
||||||
|
// `scheduleHistoryTcTokenSync` enqueues onto a single-concurrency promise chain
|
||||||
|
// (see definition above) — chunks persist sequentially in the order they were
|
||||||
|
// emitted, preserving timestamp monotonicity AND keeping the `__index` write
|
||||||
|
// safe from concurrent merge clobbers. The call returns immediately so the
|
||||||
|
// `messaging-history.set` emit is not blocked.
|
||||||
|
//
|
||||||
|
// TRADE-OFF: a listener that fires an outbound send IMMEDIATELY after the emit
|
||||||
|
// may race the still-pending persistence and get a 463 on that specific send.
|
||||||
|
// The existing 463 handler in messages-recv.ts triggers a getPrivacyTokens()
|
||||||
|
// refetch that auto-recovers within seconds. Net result is much better UX than
|
||||||
|
// per-chunk stalls.
|
||||||
|
//
|
||||||
|
// DO NOT add `await` back here without re-evaluating production latency, AND
|
||||||
|
// DO NOT call storeTcTokensFromHistorySync directly — it must go through the
|
||||||
|
// chain to preserve write ordering across overlapping chunks.
|
||||||
|
scheduleHistoryTcTokenSync(data.chats, signalRepository, keyStore, logger)
|
||||||
|
|
||||||
ev.emit('messaging-history.set', {
|
ev.emit('messaging-history.set', {
|
||||||
...data,
|
...data,
|
||||||
|
|||||||
+11
-2
@@ -28,7 +28,16 @@ console.info = function (...args: unknown[]) {
|
|||||||
|
|
||||||
// Track errors by type + JID to avoid duplicates (using Map for better performance)
|
// Track errors by type + JID to avoid duplicates (using Map for better performance)
|
||||||
const _errorTimestamps = new Map<string, number>()
|
const _errorTimestamps = new Map<string, number>()
|
||||||
const DEDUP_WINDOW_MS = 150
|
// Dedup window for repeated decrypt-error console lines (Bad MAC / Counter / etc).
|
||||||
|
// Was 150ms, but retry attempts of the SAME message are typically ~300-1000ms apart,
|
||||||
|
// so the second attempt fell outside the window and double-printed.
|
||||||
|
//
|
||||||
|
// TRADE-OFF: dedup key is `errorType + JID` (no message-id). With 5s, a burst of
|
||||||
|
// errors for the SAME JID — even of slightly different categories or different
|
||||||
|
// messages — collapses to one log line every 5s. This is intentional for a noisy
|
||||||
|
// production stream; if you need per-message visibility, set BAILEYS_LOG_LEVEL=debug
|
||||||
|
// to bypass this console-side dedup and see the structured pino logs in full.
|
||||||
|
const DEDUP_WINDOW_MS = 5000
|
||||||
|
|
||||||
console.error = function (...args: unknown[]) {
|
console.error = function (...args: unknown[]) {
|
||||||
if (args.length > 0 && typeof args[0] === 'string') {
|
if (args.length > 0 && typeof args[0] === 'string') {
|
||||||
@@ -70,7 +79,7 @@ console.error = function (...args: unknown[]) {
|
|||||||
const lastTime = _errorTimestamps.get(dedupeKey)
|
const lastTime = _errorTimestamps.get(dedupeKey)
|
||||||
|
|
||||||
if (lastTime && now - lastTime < DEDUP_WINDOW_MS) {
|
if (lastTime && now - lastTime < DEDUP_WINDOW_MS) {
|
||||||
return // Skip duplicate within 150ms window
|
return // Skip duplicate within DEDUP_WINDOW_MS window
|
||||||
}
|
}
|
||||||
|
|
||||||
_errorTimestamps.set(dedupeKey, now)
|
_errorTimestamps.set(dedupeKey, now)
|
||||||
|
|||||||
Reference in New Issue
Block a user