chore: update pasarguard-node-bridge to version 0.4.1 and adjust related dependencies

x0sina · x0sina · commit 439796fe2878 · 2026-01-08T14:36:27.000+03:30
diff --git a/app/jobs/record_usages.py b/app/jobs/record_usages.py
@@ -94,31 +94,6 @@ async def _cleanup_thread_pool():
             logger.info("ThreadPoolExecutor shut down successfully")
 
 
-# Helper functions for threading (lightweight operations that release GIL)
-def _process_node_chunk(chunk_data: tuple) -> dict:
-    """
-    Process a chunk of node data - lightweight CPU operation.
-    Uses simple arithmetic and dict operations that release GIL, perfect for threads.
-    """
-    node_id, params, coeff = chunk_data
-    users_usage = defaultdict(int)
-    for param in params:
-        uid = int(param["uid"])
-        value = int(param["value"] * coeff)
-        users_usage[uid] += value
-    return dict(users_usage)
-
-
-def _merge_usage_dicts(dicts: list[dict]) -> dict:
-    """
-    Merge multiple usage dictionaries.
-    Dict operations release GIL, perfect for ThreadPoolExecutor.
-    """
-    merged = defaultdict(int)
-    for d in dicts:
-        for uid, value in d.items():
-            merged[uid] += value
-    return dict(merged)
 
 
 async def get_dialect() -> str:
@@ -567,76 +542,6 @@ async def calculate_admin_usage(users_usage: list) -> tuple[dict, set[int]]:
     return admin_usage, set(user_admin_map.keys())
 
 
-async def calculate_users_usage(api_params: dict, usage_coefficient: dict) -> list:
-    """Calculate aggregated user usage across all nodes with coefficients applied.
-    
-    Uses ThreadPoolExecutor for lightweight operations (dict/arithmetic that release GIL).
-    ThreadPoolExecutor is faster than ProcessPoolExecutor for these operations due to less overhead.
-    """
-    if not api_params:
-        return []
-
-    def _process_usage_sync(chunks_data: list[tuple[int, list[dict], float]]):
-        """Synchronous fallback used for small batches or on executor failures."""
-        users_usage = defaultdict(int)
-        for _, params, coeff in chunks_data:
-            for param in params:
-                uid = int(param["uid"])
-                value = int(param["value"] * coeff)
-                users_usage[uid] += value
-        return [{"uid": uid, "value": value} for uid, value in users_usage.items()]
-
-    # Prepare chunks for parallel processing
-    chunks = [
-        (node_id, params, usage_coefficient.get(node_id, 1))
-        for node_id, params in api_params.items()
-        if params  # Skip empty params
-    ]
-
-    if not chunks:
-        return []
-
-    # For small datasets, process synchronously to avoid overhead
-    total_params = sum(len(params) for _, params, _ in chunks)
-    if total_params < 1000:
-        return _process_usage_sync(chunks)
-
-    # Large dataset - use ThreadPoolExecutor (faster for lightweight operations)
-    loop = asyncio.get_running_loop()
-    try:
-        thread_pool = await _get_thread_pool()
-    except Exception:
-        logger.exception("Falling back to synchronous user usage calculation: failed to init thread pool")
-        return _process_usage_sync(chunks)
-
-    try:
-        # Process chunks in parallel using threads (less overhead than processes)
-        tasks = [loop.run_in_executor(thread_pool, _process_node_chunk, chunk) for chunk in chunks]
-        chunk_results = await asyncio.gather(*tasks)
-
-        # Merge results - also lightweight, use threads
-        if len(chunk_results) > 4:
-            # Split merge operation into smaller chunks
-            chunk_size = max(1, len(chunk_results) // 4)
-            merge_chunks = [
-                chunk_results[i : i + chunk_size]
-                for i in range(0, len(chunk_results), chunk_size)
-            ]
-            merge_tasks = [
-                loop.run_in_executor(thread_pool, _merge_usage_dicts, merge_chunk)
-                for merge_chunk in merge_chunks
-            ]
-            partial_results = await asyncio.gather(*merge_tasks)
-            final_result = _merge_usage_dicts(partial_results)
-        else:
-            final_result = _merge_usage_dicts(chunk_results)
-
-        return [{"uid": uid, "value": value} for uid, value in final_result.items()]
-    except Exception:
-        logger.exception("Falling back to synchronous user usage calculation: executor merge failed")
-        return _process_usage_sync(chunks)
-
-
 async def _record_user_usages_impl():
     """
     Internal implementation of record_user_usages.
@@ -673,7 +578,18 @@ async def _record_user_usages_impl():
             else:
                 api_params[node_id] = result
 
-        users_usage = await calculate_users_usage(api_params, usage_coefficient)
+        # Aggregate user usage across all nodes with coefficients applied
+        users_usage_dict = defaultdict(int)
+        for node_id, params in api_params.items():
+            if not params:
+                continue
+            coeff = usage_coefficient.get(node_id, 1.0)
+            for param in params:
+                uid = int(param["uid"])
+                value = int(param["value"] * coeff)
+                users_usage_dict[uid] += value
+        
+        users_usage = [{"uid": uid, "value": value} for uid, value in users_usage_dict.items()]
         if not users_usage:
             logger.debug("No user usage to record")
             return
diff --git a/app/node/__init__.py b/app/node/__init__.py
@@ -16,25 +16,15 @@
 
 def calculate_max_message_size(active_users_count: int) -> int:
     """
-    Calculate max_message_size based on active users count.
-    
-    Formula: Base 5MB + (active_users_count * 1KB per user)
-    This ensures sufficient buffer for large node configurations with many users.
+    max_message_size to 64MB.
     
     Args:
         active_users_count: Number of active users in the system
         
     Returns:
-        int: Max message size in bytes
+        int: Max message size in bytes (64MB)
     """
-    base_size = 5 * 1024 * 1024  # 5MB base
-    per_user_size = 1024  # 1KB per user
-    calculated_size = base_size + (active_users_count * per_user_size)
-    
-    # Cap at 100MB to prevent excessive memory usage
-    max_cap = 100 * 1024 * 1024  # 100MB
-    
-    return min(calculated_size, max_cap)
+    return 64 * 1024 * 1024  # 64MB
 
 
 class NodeManager:
diff --git a/dashboard/package.json b/dashboard/package.json
@@ -1,7 +1,7 @@
 {
   "name": "pasarguard-dashboard",
   "private": true,
-  "version": "1.10.0",
+  "version": "1.10.2",
   "scripts": {
     "dev": "vite dev",
     "build": "vite build",
diff --git a/pyproject.toml b/pyproject.toml
@@ -40,7 +40,7 @@ dependencies = [
     "uvloop>=0.22.1; sys_platform != 'win32'",
     "aiorwlock>=1.5.0",
     "typer>=0.20.0",
-    "pasarguard-node-bridge>=0.4.0",
+    "pasarguard-node-bridge>=0.4.1",
     "pip-system-certs>=5.3",
 ]
 
diff --git a/uv.lock b/uv.lock

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,7 @@`
`1`	`1`	`{`
`2`	`2`	`"name": "pasarguard-dashboard",`
`3`	`3`	`"private": true,`
`4`		`- "version": "1.10.0",`
	`4`	`+ "version": "1.10.2",`
`5`	`5`	`"scripts": {`
`6`	`6`	`"dev": "vite dev",`
`7`	`7`	`"build": "vite build",`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@ dependencies = [`
`40`	`40`	`"uvloop>=0.22.1; sys_platform != 'win32'",`
`41`	`41`	`"aiorwlock>=1.5.0",`
`42`	`42`	`"typer>=0.20.0",`
`43`		`- "pasarguard-node-bridge>=0.4.0",`
	`43`	`+ "pasarguard-node-bridge>=0.4.1",`
`44`	`44`	`"pip-system-certs>=5.3",`
`45`	`45`	`]`
`46`	`46`