Compare commits

..

17 Commits

Author SHA1 Message Date
Renato Alcara dc24e60e5b fix(inbound): avoid HOL blocking on malformed-JID fallback bucket
Addresses Copilot review on PR #392.

The previous fallback `normalizedChatId || 'unknown'` funneled every
message with a missing or malformed chat id into a single global queue
on `postUpsertMutex`, head-of-line blocking unrelated malformed inputs
behind a shared bucket.

Prefer `msg.key?.id` over the constant: it is unique per message so
unrelated malformed inputs no longer serialize together, while valid
messages continue to hit the normalized chat-id path — preserving the
per-chat ordering guarantee where it actually matters. The CodeRabbit
concern about `msg.key.id` fragmenting the queue applied to the primary
key derivation (where ordering matters); here it is purely a fallback
for a corruption edge case where there is no chat to order by anyway.
2026-04-26 16:03:07 -03:00
Renato Alcara 942f4ae2c9 fix(inbound): recover from doAppStateSync failures + route bg errors
Addresses two Copilot review comments on PR #392.

1. doAppStateSync stuck-state recovery (pre-existing bug, made more
   visible by Promise.allSettled silencing the propagation):
   If `resyncAppState` throws, the lines that set `syncState = Online`
   and call `ev.flush()` never execute — leaving the event buffer
   pinned and `syncState` stuck at `Syncing` until the buffer's own
   safety timeout expires.

   Wrap `resyncAppState` in try/catch that forces the state machine
   forward (`syncState = Online`, `ev.flush()`) before re-throwing, so
   live inbound events can flow even when app-state sync fails.
   Collections were already cleared, so blocked patches will be
   retried on the next creds.update tick.

2. Replace `void postUpsertWork` with a defensive
   `.catch(err => onUnexpectedError(err, ...))`. The previous `void`
   was clean but exposed the long-running socket to potential process
   termination from UnhandledPromiseRejection on Node ≥15. The new
   pattern matches existing background-work handling elsewhere in
   chats.ts (presence updates, init queries) and routes truly
   unexpected rejections through the centralised error handler
   instead of letting them surface as unhandled.
2026-04-26 15:50:47 -03:00
Renato Alcara 98d2de2984 fix(inbound): skip doAppStateSync when key-share persistence fails
Addresses Copilot review on PR #392.

Previous commit moved per-task error handling into `postUpsertTasks` via
per-task `.catch(log)` so the keyed mutex would only release after BOTH
tasks settle (preventing per-chat ordering races). Side effect: the
combined promise resolves *even if processMessage failed to persist the
new app-state-sync key*. The keyShare branch then ran `doAppStateSync()`
unconditionally, which hits `isMissingKeyError`, parks collections in
`blockedCollections`, and regresses the exact scenario this branch
was created to fix.

Restructure: `postUpsertTasks` now uses `Promise.allSettled` and returns
`{ processMessageOk: boolean }`. Per-task failures are still logged
inline. The keyShare branch reads `processMessageOk` and skips
`doAppStateSync()` when the key wasn't persisted — emitting an explicit
warning instead. The fire-and-forget branch is unaffected (still `void`).
2026-04-26 15:33:38 -03:00
Renato Alcara b0c34cb536 chore(inbound): drop dead outer catch on postUpsertWork
Addresses Copilot review on PR #392.

Both tasks inside `postUpsertTasks` already swallow their rejections via
`.catch`, so `Promise.all([taskA.catch, taskB.catch])` never rejects, and
`postUpsertMutex.mutex(... postUpsertTasks)` therefore never rejects in
practice. The outer `.catch` could only fire on AsyncMutex internal
corruption — an extremely rare event, and one whose log message would
be misleading ("background post-upsert work failed" pointing at the
mutex layer, not the work).

Replace with `void postUpsertWork`: marks the floating promise as
intentional, and lets a truly unexpected rejection surface as an
UnhandledPromiseRejection — a loud, attributable signal of a real bug
rather than a swallowed warn buried in logs.
2026-04-26 15:25:41 -03:00
Renato Alcara d169a6f755 fix(inbound): handle malformed JIDs in postUpsertChatId fallback
Addresses Copilot review on PR #392.

`jidNormalizedUser()` returns an empty string when `jidDecode()` fails
(e.g. a malformed JID missing the `@` separator). With the previous
ternary `rawChatId ? jidNormalizedUser(rawChatId) : 'unknown'`, a
truthy-but-malformed `rawChatId` produced `postUpsertChatId === ''`,
so the keyed mutex ran with an empty-string key instead of the
intended `'unknown'` fallback bucket — defeating the per-chat
serialization and silently lumping malformed-key messages into a
single anonymous queue.

Compute the normalized value first then apply the fallback only when
the result is empty.
2026-04-26 15:13:42 -03:00
Renato Alcara aef4973cac fix(inbound): use dedicated postUpsertMutex to preserve enqueue order
Addresses Copilot review on PR #392.

Sharing `messageMutex(chatId)` between the outer inbound caller and the
inner post-upsert task creates an enqueue-order race. The outer callback
awaits `decrypt()` before reaching the inner enqueue site, so a concurrently
arrived message N+1 can land on the same mutex queue *before* msg N's
post-upsert task gets enqueued. The queue ends up as [OuterN+1, InnerN, ...]
and msg N+1's processMessage runs before msg N's, breaking the per-chat
ordering of side effects (chat.unreadCount, LID/PN mapping, messages.update,
history downloads).

Introduce a dedicated `postUpsertMutex` (separate KeyedMutex instance)
keyed on chatId. Post-upsert tasks enqueue strictly in upsertMessage call
order — and that IS message arrival order because the outer messageMutex
serializes upserts per-chat. The two mutexes never collide, so no
reentrancy or deadlock concern in either branch.

The keyShare branch can now also use postUpsertMutex (instead of running
inline) since the deadlock is gone, which preserves ordering with any
fire-and-forget post-upsert tasks already queued for the same chat.
2026-04-26 15:05:00 -03:00
Renato Alcara b90b383811 fix(inbound): ensure mutex holds until both post-upsert tasks settle
Addresses Copilot review on PR #392.

A plain `Promise.all([doAppStateSync(), processMessage(...)])` rejects on the
first task that fails. When that happens inside the messageMutex callback,
the mutex releases as soon as the rejection settles even though the other
task is still running. The next message for the same chat can then overtake
the in-flight task and break the per-chat ordering guarantee that was the
whole point of the inner mutex.

Wrap each task in its own `.catch` that logs and resolves. The combined
Promise.all now only settles after BOTH operations finish (success or
failure), so the mutex is held for the full duration of the work and
ordering is preserved. Errors are still surfaced via the per-task warnings.
2026-04-26 14:50:14 -03:00
Renato Alcara 35461f96c5 fix(inbound): align mutex key with processMessage chat scope
Addresses CodeRabbit Minor + Copilot review on PR #392:

1. Replace `msg.key.remoteJid || msg.key.id || 'unknown'` with
   `getChatId(msg.key) + jidNormalizedUser`. The previous derivation:
   - Diverged from processMessage's chat scoping (which calls
     `jidNormalizedUser(getChatId(message.key))`). For non-status
     broadcasts processMessage targets `participant`, not `remoteJid`,
     so the inner mutex would queue under a different key than the chat
     processMessage actually mutates — defeating per-chat ordering for
     broadcast traffic.
   - Used `msg.key.id` as a fallback. Per-message ids are unique, so
     malformed-key messages each got their own queue, defeating the
     mutex entirely. Replaced with the constant `'unknown'` bucket.

2. Drop hard-coded line numbers (`process-message.ts:650`,
   `messages-recv.ts:2416`) from comments — references rot. Comments
   now point at the function/handler instead (`APP_STATE_SYNC_KEY_SHARE
   handler`, `inbound caller`).
2026-04-26 14:38:40 -03:00
Renato Alcara 1e37448e05 fix(inbound): avoid messageMutex deadlock in appStateSyncKeyShare branch
Addresses Copilot review on PR #392.

The previous commit wrapped postUpsertWork in messageMutex.mutex(chatId, ...)
for both branches and then awaited it in the appStateSyncKeyShare path. This
deadlocks when upsertMessage is invoked from messages-recv.ts:2416, which is
already holding messageMutex.mutex(chatId, ...) for the same chat:

  outer mutex (in caller)  → await upsertMessage(...)
   └─ upsertMessage work() → schedules inner mutex on same chatId (queued)
       └─ await postUpsertWork  ← inner cannot acquire while outer holds it
                                   outer cannot release while it awaits inner
                                   = eternal deadlock (async-mutex is not re-entrant)

Fire-and-forget path is safe because it does NOT await: outer releases as
soon as upsertMessage returns, then the queued inner acquires.

Fix: in the appStateSyncKeyShare branch, run postUpsertTasks() inline (no
inner messageMutex). Per-chat ordering is still preserved by the caller's
outer mutex while we await. Other paths keep the inner mutex wrapper for
the latency win + per-chat ordering of detached work.
2026-04-26 14:14:16 -03:00
Renato Alcara 72e78a12e0 fix(inbound): preserve per-chat ordering + key-share await in detached upsert
Addresses bot review on PR #392:

1. Codex P1 + CodeRabbit Major: removing the `await` on Promise.all broke
   the per-chat ordering guarantee enforced by `messageMutex` in
   messages-recv.ts:2416 (`await messageMutex.mutex(... await upsertMessage(...))`).
   The outer mutex unlocked before the detached processMessage finished,
   so a follow-up message from the same chat could overtake state mutations
   (chat.unreadCount, LID/PN mapping, messages.update emits, history downloads).

   Fix: wrap the fire-and-forget Promise.all in `messageMutex.mutex(chatId, ...)`
   inside chats.ts. The outer mutex (in messages-recv) releases as soon as
   upsertMessage returns (fast — work is detached), then this inner mutex
   enqueues per-chat. Reentrancy is safe because outer always releases before
   inner acquires.

2. CodeRabbit Critical: appStateSyncKeyShare race. processMessage persists
   the new app-state-sync key via keyStore.transaction at process-message.ts:650.
   Previously the awaited Promise.all guaranteed the key was in the store
   before the follow-up `await doAppStateSync()` (which decodes patches via
   that key) ran. After detaching, doAppStateSync could hit isMissingKeyError
   and park collections in blockedCollections, requiring an extra retry cycle.

   Fix: for the rare appStateSyncKeyShare + Syncing branch, await the
   postUpsertWork before triggering doAppStateSync. All other messages stay
   fire-and-forget — the latency win is preserved for the hot path.

3. Copilot nit: enrich the warn payload with messageId, remoteJid and
   shouldProcessHistoryMsg so background failures can be correlated.

4. CodeRabbit nit: replace the IIFE around the conditional doAppStateSync
   with a ternary expression.
2026-04-26 13:47:14 -03:00
Renato Alcara a5b7ce7ef9 perf(inbound): detach post-upsert work from buffered function
Removes the await around Promise.all([doAppStateSync, processMessage])
inside upsertMessage. createBufferedFunction (event-buffer.ts:819) only
schedules the buffer flush AFTER its work() callback resolves, so awaiting
post-upsert work pinned the messages.upsert emit inside the buffer for the
full duration of processMessage (signal cleanup, store writes, tcToken
maintenance, history mappings).

After this change the buffered function returns as soon as the synchronous
state-machine bookkeeping completes, the debounced flush releases
messages.upsert to the consumer, and processMessage / doAppStateSync run
in the background. Failures are caught and warn-logged so unhandled
rejections cannot crash the socket.

This is the last remaining sync hop on the inbound hot path that was still
adding latency on top of PR #390 (fire-and-forget LID mapping + tcToken
history sync).
2026-04-26 13:21:52 -03:00
Renato Alcara df1acc8f0c chore(logs): reduce decrypt-error noise (~75% fewer lines per recover… (#391)
chore(logs): reduce decrypt-error noise (~75% fewer lines per recover… (#391)
2026-04-26 11:36:21 -03:00
Renato Alcara 1dffe3b311 perf(inbound-latency): restore async LID mapping + fire-and-forget tc… (#390)
* perf(inbound-latency): restore async LID mapping + fire-and-forget tctoken history sync

PRODUCTION ISSUE: Inbound messages from smartphone to ZPRO frontend were
arriving with seconds of delay. Outbound (ZPRO → smartphone) was instant.
Started after PR #386 (tctoken lifecycle) deploy.

ROOT CAUSE: Three compounding factors:

1. The historical fix d73cd28d39 (2026-02-03, "fix inbound latency by making
   LID mapping async") was partially reverted the same day by c3fc792351
   ("hybrid approach") due to a valid race-condition concern with decrypt().
   The reversion was over-protective: storeLIDPNMappings does NOT need to be
   sync — only migrateSession does. The hybrid kept all 3 awaits sync.

2. PR #386 added `await storeTcTokensFromHistorySync(...)` BEFORE the
   `messaging-history.set` emit. Per chunk this drains the event buffer with
   2-4 store ops, which compounds when many chunks arrive at once (restart,
   QR scan, multi-device login).

3. Each pre-check `await getPNForLID(alt)` / `getLIDForPN(alt)` before
   storeLIDPNMappings was redundant — the store has its own LRU cache + dedup.

Combined under production load (multi-instance store contention, post-PR #386
extra ops per send) the per-message hot-path penalty became user-visible delay.

THIS FIX:

#1+#3: messages-recv.ts ~line 2332 — `storeLIDPNMappings` becomes
fire-and-forget, pre-check `getPNForLID/getLIDForPN` removed. `migrateSession`
stays SYNC (REQUIRED for decrypt — see Codex/Copilot review on PR #72 / commit
c3fc792351). 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.

#2: process-message.ts ~line 451 — `storeTcTokensFromHistorySync` becomes
fire-and-forget. Trade-off: a listener firing an outbound send IMMEDIATELY
after the emit may race the background persistence and hit error 463 on that
specific send. Existing 463 handler in messages-recv.ts triggers
getPrivacyTokens() refetch that auto-recovers in seconds. Net UX is much
better than per-chunk stalls.

INVARIANTS PRESERVED:
- migrateSession remains SYNC — decrypt() depends on it (race condition guard)
- normalizeMessageJids remains SYNC — events need correct JIDs before emit
- messageMutex remains SYNC — per-chat ordering preserved
- All 824 tests still pass
2026-04-26 10:30:09 -03:00
Renato Alcara 4fe708445a chore: update WhatsApp Web version to v2.3000.1038167900 (#389)
Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
2026-04-26 06:18:52 -03:00
github-actions[bot] 336ed64a44 chore: update proto/version to v2.3000.1038164556 (#388)
Co-authored-by: rsalcara <rsalcara@users.noreply.github.com>
2026-04-26 01:07:43 -03:00
Renato Alcara 65cb09e4df perf(history-sync): parallelise tcToken JID resolution per chunk (#387)
perf(history-sync): parallelise tcToken JID resolution per chunk (#387)
2026-04-25 23:52:37 -03:00
Renato Alcara 9ff21db749 feat(tctoken): complete lifecycle (TIER 1 + 2 + 3 of upstream PR)
feat(tctoken): complete lifecycle (TIER 1 + 2 + 3 of upstream PR)
2026-04-25 18:52:45 -03:00
8 changed files with 350 additions and 82 deletions
+1 -1
View File
@@ -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
View File
@@ -1 +1 @@
{"version":[2,3000,1038147544]}
{"version":[2,3000,1038167900]}
+150 -24
View File
@@ -54,7 +54,7 @@ import {
resolveLidToPn
} from '../Utils'
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 {
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 */
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
let awaitingSyncTimeout: NodeJS.Timeout | undefined
@@ -1399,7 +1411,19 @@ export const makeChatsSocket = (config: SocketConfig) => {
blockedCollections.clear()
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
syncState = SyncState.Online
@@ -1411,29 +1435,131 @@ export const makeChatsSocket = (config: SocketConfig) => {
}
}
await Promise.all([
(async () => {
if (shouldProcessHistoryMsg) {
await doAppStateSync()
}
})(),
processMessage(msg, {
signalRepository,
shouldProcessHistoryMsg,
placeholderResendCache,
ev,
creds: authState.creds,
keyStore: authState.keys,
logger,
options: config.options,
getMessage
})
])
// Post-upsert work: history app-state sync + processMessage side effects.
// Awaiting here keeps `messages.upsert` pinned in the event buffer
// (createBufferedFunction only schedules flush after work() resolves), so
// the hot path detaches this work to release the emit on the next debounce
// tick.
//
// Use Promise.allSettled so the combined promise only settles after BOTH
// tasks finish. With a plain Promise.all, an early rejection from one task
// would release the keyed mutex while the other task is still mutating
// chat state — letting the next message of the same chat overtake it and
// break per-chat ordering.
//
// Returns the per-task settle status so the keyShare branch can know
// whether processMessage actually persisted the new app-state-sync key
// before triggering doAppStateSync (otherwise the sync would hit
// 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 (msg.message?.protocolMessage?.appStateSyncKeyShare && syncState === SyncState.Syncing) {
logger.info('App state sync key arrived, triggering app state sync')
await doAppStateSync()
if (historyResult.status === 'rejected') {
logger?.warn(
{ err: historyResult.reason, messageId: msg.key?.id, remoteJid: msg.key?.remoteJid },
'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
View File
@@ -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)
}
}
+12 -1
View File
@@ -341,7 +341,18 @@ export const addTransactionCapability = (
return result
} 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
}
})
+23 -2
View File
@@ -58,7 +58,11 @@ export const BAD_MAC_ERROR_TEXT = 'Bad MAC'
export const DECRYPTION_RETRY_CONFIG = {
maxRetries: 3,
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]
}
@@ -421,9 +425,26 @@ export const decryptMessageNode = (
const isCorrupted = isCorruptedSessionError(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 = {
key: fullMessage.key,
err: originalError,
err: isRecoverableCategory ? slimErr : originalError,
messageType: tag === 'plaintext' ? 'plaintext' : attrs.type,
sender,
author,
+113 -22
View File
@@ -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,
+11 -2
View File
@@ -28,7 +28,16 @@ console.info = function (...args: unknown[]) {
// Track errors by type + JID to avoid duplicates (using Map for better performance)
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[]) {
if (args.length > 0 && typeof args[0] === 'string') {
@@ -70,7 +79,7 @@ console.error = function (...args: unknown[]) {
const lastTime = _errorTimestamps.get(dedupeKey)
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)