ruicesar/InfiniteAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(history-sync): true batch + partial-failure resilience for tcToken sync

Browse Source 
Addresses all 4 PR #387 review findings (Copilot + CodeRabbit):

#1 (Copilot, Major) — Promise.all still issued one getLIDForPN call per
chat. Replaced with a single getLIDsForPNs batch over deduped PN inputs:
turns O(N) round-trips into 1, AND shares USync retry across PNs that
miss cache. LID inputs (and @hosted.lid) skip the lookup entirely.

#4 (CodeRabbit, Major) — Promise.all is all-or-nothing: a single
resolveTcTokenJid rejection (transient DB error) would reject the whole
storeTcTokensFromHistorySync call AND abort the surrounding
HISTORY_SYNC_NOTIFICATION handler — meaning messaging-history.set
never fires and EVERY tctoken in the chunk is lost. Wrapped the batch
call in try/catch with per-chat fallback (storage under unresolved jid),
matching resolveTcTokenJid's null-LID branch.

#2 (Copilot/CodeRabbit, Minor) — `if (!candidates.length) return` was
dead code: after the early return for empty tokenChats, candidates is a
1:1 map of tokenChats and can never be empty. Removed.

#3 (CodeRabbit, Minor) — `as { ... }[]` type assertion was redundant
(TypeScript already infers the shape from the async map callback).
Removed — inferred typing now catches future shape drift.

Bonus cleanup: dropped unused `resolveTcTokenJid` import.

Tests: 35/35 suites, 824/824 passing. Customizations untouched.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Renato Alcara
2026-04-25 23:43:48 -03:00
parent ae333b2910
commit b5910c591a
1 changed files with 41 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Utils/process-message.ts
+41 -26
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
@@ -88,8 +88,6 @@ async function storeTcTokensFromHistorySync(
keyStore: SignalKeyStoreWithTransaction,
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 tokenChats = chats.filter(chat => {
@@ -101,32 +99,49 @@ async function storeTcTokensFromHistorySync(
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 }[]
// 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!)
}))
if (!candidates.length) {
return
// 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)
if (mappings) {
for (const { pn, lid } of mappings) {
if (pn && lid) 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')
}
}
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)
const entries: Record<string, { token: Buffer; timestamp?: string; senderTimestamp?: number }> = {}