diff --git a/src/Socket/messages-recv.ts b/src/Socket/messages-recv.ts index 04665b2d..970f511c 100644 --- a/src/Socket/messages-recv.ts +++ b/src/Socket/messages-recv.ts @@ -2330,44 +2330,54 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => { ) const alt = msg.key.participantAlt || msg.key.remoteJidAlt - // Handle LID/PN mappings with hybrid approach: - // - Store mapping operation runs in background (non-critical for decrypt) - // - Session migration MUST complete before decrypt() to avoid "No session record" errors - // This addresses Codex/Copilot review concerns about race conditions with decrypt() + // Handle LID/PN mappings with optimized hot-path: + // - storeLIDPNMappings is fire-and-forget (background) — does NOT block decrypt + // - migrateSession is SYNC (await) — REQUIRED for decrypt to find session + // + // 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) { const altServer = jidDecode(alt)?.server const primaryJid = msg.key.participant || msg.key.remoteJid! if (altServer === 'lid') { - // Check if mapping already exists to avoid unnecessary storage operations - const existingMapping = await signalRepository.lidMapping.getPNForLID(alt) - if (!existingMapping) { - // MUST await: normalizeMessageJids() runs after this and needs the mapping - // in the LIDMappingStore to resolve LID→PN for events delivered to consumers - await signalRepository.lidMapping - .storeLIDPNMappings([{ lid: alt, pn: primaryJid }]) - .catch(error => logger.warn({ error, alt, primaryJid }, 'LID mapping storage failed')) - } + // Fire-and-forget: storeLIDPNMappings has internal cache+dedup, + // pre-check (getPNForLID) was redundant. + signalRepository.lidMapping + .storeLIDPNMappings([{ lid: alt, pn: primaryJid }]) + .catch(error => logger.warn({ error, alt, primaryJid }, 'background LID mapping store failed')) - // CRITICAL: ALWAYS migrate session, even if mapping exists - // Other code paths (e.g., USync device lookup in messages-send.ts:310-319) - // may create mappings via storeLIDPNMappings() without calling migrateSession() - // This leaves sessions under PN format while decrypt() expects LID format - // Skipping migration based on mapping existence causes "No session record" errors + // CRITICAL: ALWAYS migrate session SYNC, even if mapping exists. + // Other code paths (e.g., USync device lookup in messages-send.ts) may create + // mappings via storeLIDPNMappings() without calling migrateSession(). This + // leaves sessions under PN format while decrypt() expects LID format. + // Skipping migration based on mapping existence causes "No session record" errors. await signalRepository.migrateSession(primaryJid, alt) } else { - // Check if reverse mapping exists - const existingMapping = await signalRepository.lidMapping.getLIDForPN(alt) - if (!existingMapping) { - // MUST await: normalizeMessageJids() runs after this and needs the mapping - // 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')) - } + // Fire-and-forget: same rationale as above. + signalRepository.lidMapping + .storeLIDPNMappings([{ lid: primaryJid, pn: alt }]) + .catch(error => logger.warn({ error, alt, primaryJid }, 'background LID mapping store failed')) - // CRITICAL: ALWAYS migrate session, even if mapping exists - // Same reasoning as above - mapping existence doesn't guarantee session migration + // CRITICAL: ALWAYS migrate session SYNC. await signalRepository.migrateSession(alt, primaryJid) } } diff --git a/src/Utils/process-message.ts b/src/Utils/process-message.ts index df77a801..846441eb 100644 --- a/src/Utils/process-message.ts +++ b/src/Utils/process-message.ts @@ -82,6 +82,47 @@ const REAL_MSG_REQ_ME_STUB_TYPES = new Set([WAMessageStubType.GROUP_PARTICIPANT_ * (TC_TOKEN_INDEX_KEY) via buildMergedTcTokenIndexWrite, so the 24h prune sweep in * 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 = 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( chats: Chat[], signalRepository: SignalRepositoryWithLIDStore, @@ -565,11 +606,28 @@ const processMessage = async ( } } - // Persist tctokens carried by history-sync chats BEFORE emitting messaging-history.set - // — 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. - // Runs AFTER storeLIDPNMappings (see comment above) so LID resolution works. - await storeTcTokensFromHistorySync(data.chats, signalRepository, keyStore, logger) + // Persist tctokens carried by history-sync chats in BACKGROUND, serialised. + // + // Originally awaited (PR #386) to avoid 463 on first multi-device send, but in + // production this drained the event buffer per-chunk and added visible delivery + // 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', { ...data,