feat: Optimize Offline Node Processing with Batching and Event Loop Yielding

src/Socket/messages-recv.ts

@@ -151,7 +151,7 @@
 
 		await delay(5000)
 
 		if (!placeholderResendCache.get(messageKey?.id!)) {
 			logger.debug({ messageKey }, 'message received while resend requested')
 			return 'RESOLVED'
 		}
@@ -183,7 +183,7 @@
 			return
 		}
 
 		let data: any
 		try {
 			data = JSON.parse(mexNode.content.toString())
 		} catch (error) {
@@ -279,7 +279,7 @@
 			case 'update':
 				const settingsNode = getBinaryNodeChild(child, 'settings')
 				if (settingsNode) {
 					const update: Record<string, any> = {}
 					const nameNode = getBinaryNodeChild(settingsNode, 'name')
 					if (nameNode?.content) update.name = nameNode.content.toString()
 
@@ -1433,6 +1433,11 @@
 		node: BinaryNode
 	}
 
+	/** Yields control to the event loop to prevent blocking */
+	const yieldToEventLoop = (): Promise<void> => {
+		return new Promise(resolve => setImmediate(resolve))
+	}
+
 	const makeOfflineNodeProcessor = () => {
 		const nodeProcessorMap: Map<MessageType, (node: BinaryNode) => Promise<void>> = new Map([
 			['message', handleMessage],
@@ -1443,6 +1448,9 @@
 		const nodes: OfflineNode[] = []
 		let isProcessing = false
 
+		// Number of nodes to process before yielding to event loop
+		const BATCH_SIZE = 10
+
 		const enqueue = (type: MessageType, node: BinaryNode) => {
 			nodes.push({ type, node })
 
@@ -1453,6 +1461,8 @@
 			isProcessing = true
 
 			const promise = async () => {
+				let processedInBatch = 0
+
 				while (nodes.length && ws.isOpen) {
 					const { type, node } = nodes.shift()!
 
@@ -1464,6 +1474,14 @@
 					}
 
 					await nodeProcessor(node)
+					processedInBatch++
+
+					// Yield to event loop after processing a batch
+					// This prevents blocking the event loop for too long when there are many offline nodes
+					if (processedInBatch >= BATCH_SIZE) {
+						processedInBatch = 0
+						await yieldToEventLoop()
+					}
 				}
 
 				isProcessing = false