From 73682d40c67784fe390c3a09fd4655fa24ae373d Mon Sep 17 00:00:00 2001 From: Renato Alcara Date: Fri, 27 Feb 2026 00:12:29 -0300 Subject: [PATCH] perf(signal): eliminate 3 post-restart latency sources in migrateSession perf(signal): eliminate 3 post-restart latency sources in migrateSession --- src/Signal/libsignal.ts | 33 ++++++++++++++++++++++++++++----- src/Socket/socket.ts | 18 +++++++++++++++--- 2 files changed, 43 insertions(+), 8 deletions(-) diff --git a/src/Signal/libsignal.ts b/src/Signal/libsignal.ts index 3ca460b4..e2c30b85 100644 --- a/src/Signal/libsignal.ts +++ b/src/Signal/libsignal.ts @@ -281,7 +281,7 @@ export function makeLibSignalRepository( const parsedKeys = auth.keys as SignalKeyStoreWithTransaction const migratedSessionCache = new LRUCache({ - ttl: 3 * 24 * 60 * 60 * 1000, // 7 days + ttl: 3 * 24 * 60 * 60 * 1000, // 3 days ttlAutopurge: true, updateAgeOnGet: true }) @@ -297,6 +297,11 @@ export function makeLibSignalRepository( ttlAutopurge: true }) + // In-flight deduplication map for migrateSession: if N concurrent callers arrive + // for the same PN user before the first migration completes, they all share one + // Promise instead of each spawning their own DB transactions. + const migrationInFlight = new Map>() + const repository: SignalRepositoryWithLIDStore = { decryptGroupMessage({ group, authorJid, msg }) { const senderName = jidToSignalSenderKeyName(group, authorJid) @@ -502,17 +507,28 @@ export function makeLibSignalRepository( const { user } = decoded1 - logger.debug({ fromJid }, 'bulk device migration - loading all user devices') + // In-flight deduplication: if a migration is already running for this PN user, + // return the existing Promise. The check+set is synchronous (before any await) + // so it is safe in Node.js's single-threaded model. + const inFlight = migrationInFlight.get(user) + if (inFlight) { + logger.trace({ fromJid }, 'migrateSession: reusing in-flight migration for same user') + return inFlight + } + const migrationPromise = (async (): Promise<{ migrated: number; skipped: number; total: number }> => { // Get user's device list — use in-memory cache to avoid a DB round-trip on // every incoming LID message. Cache is invalidated after 5 minutes. // We use undefined to mean "not yet checked" and [] to mean "checked, no devices // found" so that DB misses are also cached and don't cause a per-message lookup. let userDevices: string[] | undefined = deviceListCache.get(user) if (userDevices === undefined) { + logger.debug({ fromJid }, 'bulk device migration - loading all user devices from DB') const result = await parsedKeys.get('device-list', [user]) userDevices = result[user] ?? [] deviceListCache.set(user, userDevices) + } else { + logger.trace({ fromJid, deviceCount: userDevices.length }, 'bulk device migration - device list from cache') } if (userDevices.length === 0) { @@ -616,9 +632,11 @@ export function makeLibSignalRepository( const totalOps = migrationOps.length let migratedCount = 0 - // Bulk fetch PN sessions - already exist (verified during device discovery) - const pnAddrStrings = Array.from(new Set(migrationOps.map(op => op.fromAddr.toString()))) - const pnSessions = await parsedKeys.get('session', pnAddrStrings) + // Reuse existingSessions fetched above — for PN users on s.whatsapp.net the + // signal-address format (user.device) is identical to deviceSessionKeys, so + // existingSessions already contains every session needed here. + // Avoids a redundant storage round-trip inside the transaction. + const pnSessions = existingSessions // Prepare bulk session updates (PN → LID migration + deletion) const sessionUpdates: { [key: string]: Uint8Array | null } = {} @@ -658,6 +676,11 @@ export function makeLibSignalRepository( }, `migrate-${deviceJids.length}-sessions-${jidDecode(toJid)?.user}` ) + })() + + migrationInFlight.set(user, migrationPromise) + migrationPromise.finally(() => migrationInFlight.delete(user)) + return migrationPromise } } diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index 896f2904..8381418d 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -352,6 +352,11 @@ export const makeSocket = (config: SocketConfig) => { // whose message contained "socket-query" → matched the circuit-breaker's // "socket" pattern → incorrectly tripped the breaker after 5 timeouts. const responsePromise = waitForMessage(msgId, timeoutMs) + // Prevent unhandled-rejection if sendNode throws before we reach + // `await responsePromise` below. The error from sendNode still propagates + // to the caller; this only silences the secondary rejection from + // responsePromise (which waitForMessage will emit when ws closes). + responsePromise.catch(() => {}) // Await the send so that real sendNode failures (e.g. serialisation errors) // are surfaced to the caller immediately rather than silently discarded. @@ -1548,10 +1553,17 @@ export const makeSocket = (config: SocketConfig) => { recordConnectionAttempt('success') incrementActiveConnections() - // Start session cleanup scheduler - sessionCleanup.start() + // Defer session cleanup start by 5 s to avoid DB contention during the + // initial message flood right after CB:success. The heavyweight + // getAllSessionKeys() scan would otherwise compete with migrateSession() + // for the same storage locks while the offline-message backlog is draining. + // start() is idempotent (guarded by cleanupInterval check) so deferring is safe. + const _cleanupStartTimer = setTimeout(() => sessionCleanup.start(), 5_000) + if (typeof (_cleanupStartTimer as NodeJS.Timeout).unref === 'function') { + ;(_cleanupStartTimer as NodeJS.Timeout).unref() + } - // Start session activity tracker + // Start session activity tracker immediately (lightweight, no DB scan) sessionActivityTracker.start() // Update server time offset from success node