fix(retry): resolve message delivery speed

fix(retry): resolve message delivery speed

src/Signal/libsignal.ts

@@ -506,16 +506,16 @@ export function makeLibSignalRepository(
 
 			// 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) {
+			if (userDevices === undefined) {
 				const result = await parsedKeys.get('device-list', [user])
-				userDevices = result[user]
+				userDevices = result[user] ?? []
-				if (userDevices) {
+				deviceListCache.set(user, userDevices)
-					deviceListCache.set(user, userDevices)
-				}
 			}
 
-			if (!userDevices) {
+			if (userDevices.length === 0) {
 				return { migrated: 0, skipped: 0, total: 0 }
 			}
 
@@ -534,6 +534,12 @@ export function makeLibSignalRepository(
 				return !migratedSessionCache.has(deviceKey)
 			})
 
+			// All devices already confirmed as migrated — skip DB lookups entirely
+			if (uncachedDevices.length === 0) {
+				logger.debug({ fromJid, totalDevices: userDevices.length }, 'bulk device migration - all devices already cached, skipping')
+				return { migrated: 0, skipped: 0, total: userDevices.length }
+			}
+
 			// Bulk check session existence only for uncached devices
 			const deviceSessionKeys = uncachedDevices.map(device => `${user}.${device}`)
 			const existingSessions = await parsedKeys.get('session', deviceSessionKeys)
@@ -565,6 +571,15 @@ export function makeLibSignalRepository(
 				'bulk device migration complete - all user devices processed'
 			)
 
+			// No PN-format sessions found: all devices are already migrated to LID addressing.
+			// Cache them now so subsequent messages skip redundant DB lookups entirely.
+			if (deviceJids.length === 0) {
+				for (const device of uncachedDevices) {
+					migratedSessionCache.set(`${user}.${device}`, true)
+				}
+				return { migrated: 0, skipped: 0, total: userDevices.length }
+			}
+
 			// Single transaction for all migrations
 			return parsedKeys.transaction(
 				async (): Promise<{ migrated: number; skipped: number; total: number }> => {
@@ -630,14 +645,12 @@ export function makeLibSignalRepository(
 					if (Object.keys(sessionUpdates).length > 0) {
 						await parsedKeys.set({ session: sessionUpdates })
 						logger.debug({ migratedSessions: migratedCount }, 'bulk session migration complete')
+					}
 
-						// Cache device-level migrations
+					// Cache ALL processed devices (migrated, skipped, or closed-session) to prevent
-						for (const op of migrationOps) {
+					// redundant DB lookups on subsequent messages from the same contact.
-							if (sessionUpdates[op.toAddr.toString()]) {
+					for (const op of migrationOps) {
-								const deviceKey = `${op.pnUser}.${op.deviceId}`
+						migratedSessionCache.set(`${op.pnUser}.${op.deviceId}`, true)
-								migratedSessionCache.set(deviceKey, true)
-							}
-						}
 					}
 
 					const skippedCount = totalOps - migratedCount

src/Socket/socket.ts

@@ -343,12 +343,22 @@ export const makeSocket = (config: SocketConfig) => {
 
 		const msgId = node.attrs.id
 
-		const result = await promiseTimeout<any>(timeoutMs, async (resolve, reject) => {
+		// Register the response listener BEFORE sending — avoids a race where the server
-			const result = waitForMessage(msgId, timeoutMs).catch(reject)
+		// responds before we start listening.  waitForMessage already handles its own
-			sendNode(node)
+		// timeout (returns undefined) and connection-close errors (throws), so we do NOT
-				.then(async () => resolve(await result))
+		// wrap it in a second promiseTimeout.  The outer wrapper caused a race condition
-				.catch(reject)
+		// where both timers fired at ~the same deadline: the outer one threw
-		})
+		// Boom('Timed Out') while sendNode was still pending, producing a spurious error
+		// 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)
+
+		// Await the send so that real sendNode failures (e.g. serialisation errors)
+		// are surfaced to the caller immediately rather than silently discarded.
+		// Socket-close errors are also propagated by waitForMessage via ws.on('close').
+		await sendNode(node)
+
+		const result = await responsePromise
 
 		if (result && 'tag' in result) {
 			assertNodeErrorFree(result)

src/Utils/circuit-breaker.ts

@@ -716,6 +716,10 @@ export function createConnectionCircuitBreaker(customOptions?: Partial<CircuitBr
 		resetTimeout: 60000,
 		successThreshold: 1,
 		shouldCountError: (error: Error) => {
+			// CircuitTimeoutError messages embed the circuit name (e.g. "socket-query"), which
+			// accidentally matches the "socket" pattern below and causes a self-reinforcing loop
+			// where the breaker's own timeouts keep tripping the breaker. Exclude them explicitly.
+			if (error instanceof CircuitTimeoutError) return false
 			const message = error.message.toLowerCase()
 			const code = (error as NodeJS.ErrnoException).code?.toLowerCase() || ''
 			return connectionErrorPatterns.some(pattern => message.includes(pattern) || code.includes(pattern))

src/Utils/message-retry-manager.ts

@@ -174,7 +174,13 @@ export class MessageRetryManager {
 		// (device suffix present/absent, LID vs PN, etc.)
 		const indexedKeyStr = this.messageKeyIndex.get(id)
 		if (indexedKeyStr) {
-			return this.recentMessagesMap.get(indexedKeyStr)
+			const message = this.recentMessagesMap.get(indexedKeyStr)
+			if (!message) {
+				// The LRU cache evicted this entry; clean up the stale index reference
+				// to prevent repeated futile lookups on subsequent retry receipts.
+				this.messageKeyIndex.delete(id)
+			}
+			return message
 		}
 
 		return undefined