fix: jobs import error

ImMohammad20000 · ImMohammad20000 · commit f039f4c148e4 · 2026-01-05T19:09:25.000+03:30
diff --git a/app/jobs/__init__.py b/app/jobs/__init__.py
@@ -1,5 +1,6 @@
 import glob
 import importlib.util
+import sys
 from os.path import basename, dirname, join
 
 modules = glob.glob(join(dirname(__file__), "*.py"))
@@ -9,5 +10,9 @@
     if name.startswith("_"):
         continue
 
-    spec = importlib.util.spec_from_file_location(name, file)
-    spec.loader.exec_module(importlib.util.module_from_spec(spec))
+    # Load job modules under the proper package path so multiprocessing pickling works
+    module_name = f"{__name__}.{name}"
+    spec = importlib.util.spec_from_file_location(module_name, file)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
diff --git a/app/jobs/record_usages.py b/app/jobs/record_usages.py
@@ -437,6 +437,16 @@ async def calculate_users_usage(api_params: dict, usage_coefficient: dict) -> li
     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))
@@ -450,45 +460,42 @@ async def calculate_users_usage(api_params: dict, usage_coefficient: dict) -> li
     # For small datasets, process synchronously to avoid overhead
     total_params = sum(len(params) for _, params, _ in chunks)
     if total_params < 1000:
-        # Small dataset - process synchronously
-        users_usage = defaultdict(int)
-        for node_id, params, coeff in chunks:
-            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()]
+        return _process_usage_sync(chunks)
 
     # Large dataset - use multiprocessing
     loop = asyncio.get_running_loop()
-    process_pool = await _get_process_pool()
-
-    # Process chunks in parallel
-    tasks = [
-        loop.run_in_executor(process_pool, _process_node_chunk, chunk)
-        for chunk in chunks
-    ]
-    
-    chunk_results = await asyncio.gather(*tasks)
-
-    # Merge results - this is also CPU-bound, so parallelize if many chunks
-    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(process_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)
+    try:
+        process_pool = await _get_process_pool()
+    except Exception:
+        logger.exception("Falling back to synchronous user usage calculation: failed to init process pool")
+        return _process_usage_sync(chunks)
 
-    return [{"uid": uid, "value": value} for uid, value in final_result.items()]
+    try:
+        # Process chunks in parallel
+        tasks = [loop.run_in_executor(process_pool, _process_node_chunk, chunk) for chunk in chunks]
+        chunk_results = await asyncio.gather(*tasks)
+
+        # Merge results - this is also CPU-bound, so parallelize if many chunks
+        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(process_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():
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.3.10",
+    "pasarguard-node-bridge>=0.3.11",
     "pip-system-certs>=5.3",
 ]
 
diff --git a/uv.lock b/uv.lock

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.3.10",`
	`43`	`+ "pasarguard-node-bridge>=0.3.11",`
`44`	`44`	`"pip-system-certs>=5.3",`
`45`	`45`	`]`
`46`	`46`