fix: revert 0_xray_core job

app/jobs/0_xray_core.py

@@ -4,54 +4,32 @@
 from app import app, logger, scheduler, xray
 from app.db import GetDB, crud
 from app.models.node import NodeStatus
-from app.utils.concurrency import threaded_function
 from xray_api import exc as xray_exc
 
-global _node_op_in_progress
-_node_op_in_progress = {}
-
-
-# ops = {node_id: (func, kwargs)}
-@threaded_function
-def _op(ops: dict):
-    for node_id in list(ops.keys()):
-        if _node_op_in_progress.get(node_id):
-            del ops[node_id]
-        else:
-            _node_op_in_progress[node_id] = True
-
-    if not ops:
-        return
-
-    config = xray.config.include_db_users()
-    for node_id, (func, kwargs) in ops.items():
-        func(config=config, **kwargs)
-        try:
-            del _node_op_in_progress[node_id]
-        except KeyError:
-            pass
-
 
 def core_health_check():
-    ops = {}
+    config = None
 
     # main core
     if not xray.core.started:
-        ops[0] = (xray.core.restart, {})
+        if not config:
+            config = xray.config.include_db_users()
+        xray.core.restart(config)
 
     # nodes' core
     for node_id, node in list(xray.nodes.items()):
         if node.connected:
             try:
                 node.api.get_sys_stats()
             except (ConnectionError, xray_exc.ConnectionError, xray_exc.UnknownError):
-                ops[node_id] = (xray.operations.restart_node, {"node_id": node_id})
+                if not config:
+                    config = xray.config.include_db_users()
+                xray.operations.restart_node(node_id, config)
 
         if not node.connected:
-            ops[node_id] = (xray.operations.connect_node, {"node_id": node_id})
-
-    if ops:
-        _op(ops)
+            if not config:
+                config = xray.config.include_db_users()
+            xray.operations.connect_node(node_id, config)
 
 
 @app.on_event("startup")