fix Memory leak in makeMutex - Promise never gets garbage collected (#2151)

* Memory leak in makeMutex - Promise never gets garbage collected

Hey, I've been debugging a memory leak in my application and traced it back to the makeMutex implementation.

The current implementation chains promises indefinitely without ever breaking the chain:

Every call to mutex() creates a new Promise that awaits the previous task, then becomes the new task. The problem is the old promises never get released because each one holds a reference to the previous through the closure.

What I found
Took a heap snapshot after running for a while and found hundreds of Promises from make-mutex.js holding ~15MB and growing. The retainer graph shows a long chain of Promises all pointing back to each other.

Since processingMutex handles every incoming message/notification, this chain grows constantly and never shrinks.

This keeps the same mutex behavior but lets the GC clean up old promises every 50 tasks instead of holding them forever.

* Refactor makeMutex to use AsyncMutex directly

* lint

* revert

src/Socket/messages-recv.ts

@@ -153,7 +153,7 @@ export const makeMessagesRecvSocket = (config: SocketConfig) => {
 
 		await delay(5000)
 
-		if (!await placeholderResendCache.get(messageKey?.id!)) {
+		if (!(await placeholderResendCache.get(messageKey?.id!))) {
 			logger.debug({ messageKey }, 'message received while resend requested')
 			return 'RESOLVED'
 		}

src/Utils/make-mutex.ts

@@ -1,29 +1,11 @@
-export const makeMutex = () => {
-	// eslint-disable-next-line @typescript-eslint/no-explicit-any
-	let task = Promise.resolve() as Promise<any>
+import { Mutex as AsyncMutex } from 'async-mutex'
 
-	let taskTimeout: NodeJS.Timeout | undefined
+export const makeMutex = () => {
+	const mutex = new AsyncMutex()
 
 	return {
 		mutex<T>(code: () => Promise<T> | T): Promise<T> {
-			task = (async () => {
-				// wait for the previous task to complete
-				// if there is an error, we swallow so as to not block the queue
-				try {
-					await task
-				} catch {}
-
-				try {
-					// execute the current task
-					const result = await code()
-					return result
-				} finally {
-					clearTimeout(taskTimeout)
-				}
-			})()
-			// we replace the existing task, appending the new piece of execution to it
-			// so the next task will have to wait for this one to finish
-			return task
+			return mutex.runExclusive(code)
 		}
 	}
 }
@@ -31,15 +13,30 @@ export const makeMutex = () => {
 export type Mutex = ReturnType<typeof makeMutex>
 
 export const makeKeyedMutex = () => {
-	const map: { [id: string]: Mutex } = {}
+	const map = new Map<string, { mutex: AsyncMutex; refCount: number }>()
 
 	return {
-		mutex<T>(key: string, task: () => Promise<T> | T): Promise<T> {
-			if (!map[key]) {
-				map[key] = makeMutex()
+		async mutex<T>(key: string, task: () => Promise<T> | T): Promise<T> {
+			let entry = map.get(key)
+
+			if (!entry) {
+				entry = { mutex: new AsyncMutex(), refCount: 0 }
+				map.set(key, entry)
 			}
 
-			return map[key].mutex(task)
+			entry.refCount++
+
+			try {
+				return await entry.mutex.runExclusive(task)
+			} finally {
+				entry.refCount--
+				// only delete it if this is still the current entry
+				if (entry.refCount === 0 && map.get(key) === entry) {
+					map.delete(key)
+				}
+			}
 		}
 	}
 }
+
+export type KeyedMutex = ReturnType<typeof makeKeyedMutex>