diff --git a/src/Signal/libsignal.ts b/src/Signal/libsignal.ts index b846f961..3ca460b4 100644 --- a/src/Signal/libsignal.ts +++ b/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] - if (userDevices) { - deviceListCache.set(user, userDevices) - } + userDevices = result[user] ?? [] + 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 - for (const op of migrationOps) { - if (sessionUpdates[op.toAddr.toString()]) { - const deviceKey = `${op.pnUser}.${op.deviceId}` - migratedSessionCache.set(deviceKey, true) - } - } + // Cache ALL processed devices (migrated, skipped, or closed-session) to prevent + // redundant DB lookups on subsequent messages from the same contact. + for (const op of migrationOps) { + migratedSessionCache.set(`${op.pnUser}.${op.deviceId}`, true) } const skippedCount = totalOps - migratedCount diff --git a/src/Socket/socket.ts b/src/Socket/socket.ts index f90d866a..896f2904 100644 --- a/src/Socket/socket.ts +++ b/src/Socket/socket.ts @@ -343,12 +343,22 @@ export const makeSocket = (config: SocketConfig) => { const msgId = node.attrs.id - const result = await promiseTimeout(timeoutMs, async (resolve, reject) => { - const result = waitForMessage(msgId, timeoutMs).catch(reject) - sendNode(node) - .then(async () => resolve(await result)) - .catch(reject) - }) + // Register the response listener BEFORE sending — avoids a race where the server + // responds before we start listening. waitForMessage already handles its own + // timeout (returns undefined) and connection-close errors (throws), so we do NOT + // wrap it in a second promiseTimeout. The outer wrapper caused a race condition + // 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(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) diff --git a/src/Utils/circuit-breaker.ts b/src/Utils/circuit-breaker.ts index ad9e01c0..d2d07d59 100644 --- a/src/Utils/circuit-breaker.ts +++ b/src/Utils/circuit-breaker.ts @@ -716,6 +716,10 @@ export function createConnectionCircuitBreaker(customOptions?: Partial { + // 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)) diff --git a/src/Utils/message-retry-manager.ts b/src/Utils/message-retry-manager.ts index 59164f07..b50d62bc 100644 --- a/src/Utils/message-retry-manager.ts +++ b/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