Compare commits
5 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 9e3af35b54 | |||
| a631c8c3c3 | |||
| 4fe708445a | |||
| 336ed64a44 | |||
| 65cb09e4df |
@@ -1,7 +1,7 @@
|
||||
syntax = "proto3";
|
||||
package proto;
|
||||
|
||||
/// WhatsApp Version: 2.3000.1038024963
|
||||
/// WhatsApp Version: 2.3000.1038164556
|
||||
|
||||
message ADVDeviceIdentity {
|
||||
optional uint32 rawId = 1;
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"version":[2,3000,1038147544]}
|
||||
{"version":[2,3000,1038167900]}
|
||||
|
||||
+39
-29
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
+113
-22
@@ -40,7 +40,7 @@ import { getKeyAuthor, toNumber } from './generics'
|
||||
import { downloadAndProcessHistorySyncNotification } from './history'
|
||||
import type { ILogger } from './logger'
|
||||
import { metrics, recordHistorySyncMessages } from './prometheus-metrics.js'
|
||||
import { buildMergedTcTokenIndexWrite, resolveTcTokenJid } from './tc-token-utils'
|
||||
import { buildMergedTcTokenIndexWrite } from './tc-token-utils'
|
||||
|
||||
type ProcessMessageContext = {
|
||||
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
|
||||
* 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(
|
||||
chats: Chat[],
|
||||
signalRepository: SignalRepositoryWithLIDStore,
|
||||
keyStore: SignalKeyStoreWithTransaction,
|
||||
logger?: ILogger
|
||||
) {
|
||||
const getLIDForPN = signalRepository.lidMapping.getLIDForPN.bind(signalRepository.lidMapping)
|
||||
|
||||
const candidates: { storageJid: string; token: Buffer; ts: number; senderTs?: number }[] = []
|
||||
for (const chat of chats) {
|
||||
// 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 tokenChats = chats.filter(chat => {
|
||||
const ts = chat.tcTokenTimestamp ? toNumber(chat.tcTokenTimestamp) : 0
|
||||
if (chat.tcToken?.length && ts > 0) {
|
||||
const jid = jidNormalizedUser(chat.id!)
|
||||
const storageJid = await resolveTcTokenJid(jid, getLIDForPN)
|
||||
candidates.push({
|
||||
storageJid,
|
||||
token: Buffer.from(chat.tcToken),
|
||||
ts,
|
||||
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
||||
})
|
||||
return !!chat.tcToken?.length && ts > 0
|
||||
})
|
||||
|
||||
if (!tokenChats.length) {
|
||||
return
|
||||
}
|
||||
|
||||
// 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) {
|
||||
return
|
||||
}
|
||||
const candidates = normalized.map(({ chat, ts, jid }) => ({
|
||||
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 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
|
||||
// — 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,
|
||||
|
||||
Reference in New Issue
Block a user