perf(signal): eliminate 3 post-restart latency sources in migrateSession

perf(signal): eliminate 3 post-restart latency sources in migrateSession

src/Signal/libsignal.ts

@@ -281,7 +281,7 @@ export function makeLibSignalRepository(
 
 	const parsedKeys = auth.keys as SignalKeyStoreWithTransaction
 	const migratedSessionCache = new LRUCache<string, true>({
-		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<string, Promise<{ migrated: number; skipped: number; total: number }>>()
+
 	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
 		}
 	}
 

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<any>(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