fix(critical): resolve 4 race conditions in transaction capability and session management
fix(critical): resolve 4 race conditions in transaction capability and session management
This commit is contained in:
@@ -465,7 +465,10 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
if (shouldRecreateSession) {
|
if (shouldRecreateSession) {
|
||||||
logger.debug({ fromJid, retryCount, reason: recreateReason, errorCode }, 'recreating session for retry')
|
logger.debug({ fromJid, retryCount, reason: recreateReason, errorCode }, 'recreating session for retry')
|
||||||
// Delete existing session to force recreation
|
// Delete existing session to force recreation
|
||||||
await authState.keys.set({ session: { [sessionId]: null } })
|
// CRITICAL: Wrap in transaction to prevent race with other session operations
|
||||||
|
await authState.keys.transaction(async () => {
|
||||||
|
await authState.keys.set({ session: { [sessionId]: null } })
|
||||||
|
}, `delete-session-${sessionId}`)
|
||||||
forceIncludeKeys = true
|
forceIncludeKeys = true
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
@@ -1023,7 +1026,10 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
|
|||||||
|
|
||||||
if (shouldRecreateSession) {
|
if (shouldRecreateSession) {
|
||||||
logger.debug({ participant, retryCount, reason: recreateReason, errorCode }, 'recreating session for outgoing retry')
|
logger.debug({ participant, retryCount, reason: recreateReason, errorCode }, 'recreating session for outgoing retry')
|
||||||
await authState.keys.set({ session: { [sessionId]: null } })
|
// CRITICAL: Wrap in transaction to prevent race with other session operations
|
||||||
|
await authState.keys.transaction(async () => {
|
||||||
|
await authState.keys.set({ session: { [sessionId]: null } })
|
||||||
|
}, `delete-session-${sessionId}`)
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
logger.warn({ error, participant }, 'failed to check session recreation for outgoing retry')
|
logger.warn({ error, participant }, 'failed to check session recreation for outgoing retry')
|
||||||
|
|||||||
@@ -766,12 +766,12 @@ export const makeSocket = (config: SocketConfig) => {
|
|||||||
logger.error({ error }, '🔑 PreKey auto-sync failed')
|
logger.error({ error }, '🔑 PreKey auto-sync failed')
|
||||||
} finally {
|
} finally {
|
||||||
isRunning = false
|
isRunning = false
|
||||||
}
|
|
||||||
|
|
||||||
// PROTECTION 3: Prevent timer accumulation and post-cleanup rescheduling
|
// PROTECTION 3: Prevent timer accumulation and post-cleanup rescheduling
|
||||||
// Check cleanedUp flag atomically to prevent race condition
|
// Check cleanedUp flag atomically INSIDE finally to minimize race window
|
||||||
if (!closed && !cleanedUp && ws.isOpen) {
|
if (!closed && !cleanedUp && ws.isOpen) {
|
||||||
syncTimer = setTimeout(syncLoop, SYNC_INTERVAL)
|
syncTimer = setTimeout(syncLoop, SYNC_INTERVAL)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+38
-20
@@ -129,6 +129,9 @@ export const addTransactionCapability = (
|
|||||||
// Pre-key manager for specialized operations
|
// Pre-key manager for specialized operations
|
||||||
const preKeyManager = new PreKeyManager(state, logger)
|
const preKeyManager = new PreKeyManager(state, logger)
|
||||||
|
|
||||||
|
// Destroyed flag to prevent operations after cleanup
|
||||||
|
let destroyed = false
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Get or create a queue for a specific key type
|
* Get or create a queue for a specific key type
|
||||||
*/
|
*/
|
||||||
@@ -300,6 +303,11 @@ export const addTransactionCapability = (
|
|||||||
isInTransaction,
|
isInTransaction,
|
||||||
|
|
||||||
transaction: async (work, key) => {
|
transaction: async (work, key) => {
|
||||||
|
// CRITICAL: Prevent transactions after destroy to avoid use-after-free
|
||||||
|
if (destroyed) {
|
||||||
|
throw new Error('Transaction capability destroyed - socket closed')
|
||||||
|
}
|
||||||
|
|
||||||
const existing = txStorage.getStore()
|
const existing = txStorage.getStore()
|
||||||
|
|
||||||
// Nested transaction - reuse existing context
|
// Nested transaction - reuse existing context
|
||||||
@@ -346,7 +354,36 @@ export const addTransactionCapability = (
|
|||||||
* Should be called during connection cleanup
|
* Should be called during connection cleanup
|
||||||
*/
|
*/
|
||||||
destroy: () => {
|
destroy: () => {
|
||||||
|
// CRITICAL: Set destroyed flag FIRST to prevent new transactions
|
||||||
|
destroyed = true
|
||||||
logger.debug('🗑️ Cleaning up transaction capability resources')
|
logger.debug('🗑️ Cleaning up transaction capability resources')
|
||||||
|
|
||||||
|
// Count locked mutexes BEFORE destroying resources
|
||||||
|
let clearedCount = 0
|
||||||
|
let lockedCount = 0
|
||||||
|
const lockedKeys: string[] = []
|
||||||
|
|
||||||
|
txMutexes.forEach((mutex, key) => {
|
||||||
|
if (!mutex.isLocked()) {
|
||||||
|
txMutexes.delete(key)
|
||||||
|
txMutexRefCounts.delete(key)
|
||||||
|
clearedCount++
|
||||||
|
} else {
|
||||||
|
lockedCount++
|
||||||
|
lockedKeys.push(key)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
|
||||||
|
// If there are locked mutexes, log error and skip resource destruction
|
||||||
|
if (lockedCount > 0) {
|
||||||
|
logger.error(
|
||||||
|
{ lockedCount, lockedKeys },
|
||||||
|
'❌ Cannot destroy resources - transactions still active! Resources will be cleaned up by GC.'
|
||||||
|
)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Safe to destroy resources (no active transactions)
|
||||||
preKeyManager.destroy()
|
preKeyManager.destroy()
|
||||||
|
|
||||||
// Clear all key queues
|
// Clear all key queues
|
||||||
@@ -357,26 +394,7 @@ export const addTransactionCapability = (
|
|||||||
})
|
})
|
||||||
keyQueues.clear()
|
keyQueues.clear()
|
||||||
|
|
||||||
// Clear transaction mutexes and reference counts
|
logger.debug({ clearedCount }, 'Transaction capability cleanup completed')
|
||||||
// CRITICAL: Only delete unlocked mutexes to prevent corrupting active transactions
|
|
||||||
let clearedCount = 0
|
|
||||||
let lockedCount = 0
|
|
||||||
txMutexes.forEach((mutex, key) => {
|
|
||||||
if (!mutex.isLocked()) {
|
|
||||||
txMutexes.delete(key)
|
|
||||||
txMutexRefCounts.delete(key)
|
|
||||||
clearedCount++
|
|
||||||
} else {
|
|
||||||
lockedCount++
|
|
||||||
logger.warn(
|
|
||||||
{ key },
|
|
||||||
'Transaction mutex still locked during cleanup - transaction may be in progress'
|
|
||||||
)
|
|
||||||
}
|
|
||||||
})
|
|
||||||
logger.debug({ clearedCount, lockedCount }, 'Transaction mutexes cleanup completed')
|
|
||||||
|
|
||||||
logger.debug('Transaction capability cleanup completed')
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user