perf(history-sync): parallelise tcToken JID resolution per chunk
PR #386 added storeTcTokensFromHistorySync which iterates chats in a sync chunk and calls resolveTcTokenJid (an async getLIDForPN lookup) per candidate. The original shape used sequential `await` inside a for-of, so each chunk blocked for O(N) round-trips before messaging-history.set could fire downstream. In production this surfaced as systemic message-delivery latency: under heavy history sync (re-scan / multi-device pairing) the event buffer backed up, the adaptive flush mode escalated to "aggressive", and outbound sends competed with the sync queue for socket time. QR re-scans made it worse because a fresh history sync triggered another round of sequential resolutions. Fix: 1. Pre-filter chats with `tcToken && tcTokenTimestamp > 0` so we never spin up promises for the empty cases (most chats in a chunk). 2. Parallelise the resolveTcTokenJid lookups via Promise.all — same semantics, but all N getLIDForPN calls run concurrently instead of serially. Downstream (loop that builds `entries`) is untouched: it remains synchronous and keeps the same-batch dedup guard, so monotonicity guarantees stay intact. Customizations untouched: zero diff in carousel/buttons/lists/ LID-PN normalization/Bad MAC/proto. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -90,21 +90,39 @@ async function storeTcTokensFromHistorySync(
|
||||
) {
|
||||
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
|
||||
}
|
||||
|
||||
// PARALLEL resolveTcTokenJid: each call hits getLIDForPN which can be a DB lookup.
|
||||
// Sequential await inside a for-of (the previous shape) blocked history-sync chunks
|
||||
// for O(N) round-trips per chunk, which under heavy sync stalled the event buffer
|
||||
// and caused user-visible message-delivery latency. Promise.all parallelises the
|
||||
// resolution while keeping the same semantics for the subsequent candidates loop
|
||||
// (which is purely synchronous and does its own same-batch dedup).
|
||||
const candidates = (
|
||||
await Promise.all(
|
||||
tokenChats.map(async chat => {
|
||||
const ts = toNumber(chat.tcTokenTimestamp!)
|
||||
const jid = jidNormalizedUser(chat.id!)
|
||||
const storageJid = await resolveTcTokenJid(jid, getLIDForPN)
|
||||
return {
|
||||
storageJid,
|
||||
token: Buffer.from(chat.tcToken!),
|
||||
ts,
|
||||
senderTs: chat.tcTokenSenderTimestamp ? toNumber(chat.tcTokenSenderTimestamp) : undefined
|
||||
}
|
||||
})
|
||||
)
|
||||
) as { storageJid: string; token: Buffer; ts: number; senderTs?: number }[]
|
||||
|
||||
if (!candidates.length) {
|
||||
return
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user