fix(node): use semaphore to avoid db pool overflow

Author: M03ED

app/jobs/node_checker.py

@@ -126,6 +126,9 @@ async def initialize_nodes():
     async with GetDB() as db:
         db_nodes = await get_nodes(db=db, enabled=True)
 
+        # Semaphore to limit concurrent node startups to 3 at a time
+        semaphore = asyncio.Semaphore(3)
+
         async def start_node(node: Node):
             try:
                 await node_manager.update_node(node)
@@ -135,10 +138,14 @@ async def start_node(node: Node):
 
             await node_operator.connect_node(node_id=node.id)
 
+        async def start_node_with_limit(node: Node):
+            async with semaphore:
+                await start_node(node)
+
         if not db_nodes:
             logger.warning("Attention: You have no node, you need to have at least one node")
         else:
-            start_tasks = [start_node(node=db_node) for db_node in db_nodes]
+            start_tasks = [start_node_with_limit(node=db_node) for db_node in db_nodes]
             await asyncio.gather(*start_tasks)
             logger.info("All nodes' cores have been started.")
 

app/operation/node.py

@@ -195,7 +195,15 @@ async def restart_node(self, node_id: Node, admin: AdminDetails) -> None:
 
     async def restart_all_node(self, db: AsyncSession, admin: AdminDetails, core_id: int | None = None) -> None:
         nodes: list[Node] = await self.get_db_nodes(db, core_id)
-        await asyncio.gather(*[NodeOperation.connect_node(node.id) for node in nodes])
+
+        # Semaphore to limit concurrent node restarts to 3 at a time
+        semaphore = asyncio.Semaphore(3)
+
+        async def restart_with_limit(node_id: int):
+            async with semaphore:
+                await NodeOperation.connect_node(node_id)
+
+        await asyncio.gather(*[restart_with_limit(node.id) for node in nodes])
 
         logger.info(f'All nodes restarted by admin "{admin.username}"')