fix(node): force reconnect for invalid nodes

Author: M03ED

app/jobs/node_checker.py

@@ -85,11 +85,6 @@ async def process_node_health_check(db_node: Node, node: PasarGuardNode):
     if node is None:
         return
 
-    if node.requires_hard_reset():
-        async with GetDB() as db:
-            await node_operator.connect_single_node(db, db_node.id)
-        return
-
     try:
         health = await asyncio.wait_for(verify_node_backend_health(node, db_node.name), timeout=15)
     except asyncio.TimeoutError:
@@ -101,6 +96,11 @@ async def process_node_health_check(db_node: Node, node: PasarGuardNode):
         async with GetDB() as db:
             await NodeOperation._update_single_node_status(db, db_node.id, NodeStatus.error, message=e.detail)
 
+    if node.requires_hard_reset() or health is None or health in (Health.NOT_CONNECTED, Health.INVALID):
+        async with GetDB() as db:
+            await node_operator.connect_single_node(db, db_node.id)
+        return
+
     # Skip nodes that are already healthy and connected
     if health == Health.HEALTHY and db_node.status == NodeStatus.connected:
         return

app/operation/node.py

@@ -157,7 +157,6 @@ async def connect_node(db_node: Node, users: list) -> dict | None:
                 users=users,
                 keep_alive=db_node.keep_alive,
                 exclude_inbounds=core.exclude_inbound_tags,
-                timeout=10,
             )
             logger.info(f'Connected to "{db_node.name}" node v{info.node_version}, xray run on v{info.core_version}')
 
@@ -296,6 +295,9 @@ async def connect_nodes_bulk(
         users = await core_users(db=db)
 
         async def connect_single(node: Node) -> dict | None:
+            if node is None or node.status in (NodeStatus.disabled, NodeStatus.limited):
+                return
+
             try:
                 await node_manager.update_node(node)
             except NodeAPIError as e:
@@ -362,7 +364,7 @@ async def connect_single_node(self, db: AsyncSession, node_id: int) -> None:
             node_id (int): ID of the node to connect.
         """
         db_node = await get_node_by_id(db, node_id)
-        if db_node is None:
+        if db_node is None or db_node.status in (NodeStatus.disabled, NodeStatus.limited):
             return
 
         # Get core users once

tests/api/helpers.py

@@ -59,7 +59,7 @@ def create_core(
 
 def delete_core(access_token: str, core_id: int) -> None:
     response = client.delete(f"/api/core/{core_id}", headers=auth_headers(access_token))
-   
+
     assert response.status_code in (status.HTTP_204_NO_CONTENT, status.HTTP_403_FORBIDDEN)
 
 

tests/api/test_bulk.py

@@ -31,10 +31,7 @@ def test_add_groups_to_users(access_token):
     """Test bulk adding groups to users."""
 
     core, groups = setup_groups(access_token, 2)
-    users = [
-        create_user(access_token, payload={"username": unique_name("bulk_user")})
-        for _ in range(2)
-    ]
+    users = [create_user(access_token, payload={"username": unique_name("bulk_user")}) for _ in range(2)]
     group_ids = [group["id"] for group in groups]
     try:
         response = client.post(
@@ -54,10 +51,7 @@ def test_add_groups_to_users(access_token):
 def test_remove_groups_from_users(access_token):
     """Test bulk removing groups from users."""
     core, groups = setup_groups(access_token, 2)
-    users = [
-        create_user(access_token, payload={"username": unique_name("bulk_user_remove")})
-        for _ in range(2)
-    ]
+    users = [create_user(access_token, payload={"username": unique_name("bulk_user_remove")}) for _ in range(2)]
     group_ids = [group["id"] for group in groups]
     try:
         client.post(
@@ -83,8 +77,12 @@ def test_update_users_datalimit(access_token):
     """Test bulk updating user data limits."""
     core, groups = setup_groups(access_token, 1)
     users = [
-        create_user(access_token, group_ids=[groups[0]["id"]], payload={"username": unique_name("user7"), "data_limit": 100}),
-        create_user(access_token, group_ids=[groups[0]["id"]], payload={"username": unique_name("user8"), "data_limit": 200}),
+        create_user(
+            access_token, group_ids=[groups[0]["id"]], payload={"username": unique_name("user7"), "data_limit": 100}
+        ),
+        create_user(
+            access_token, group_ids=[groups[0]["id"]], payload={"username": unique_name("user8"), "data_limit": 200}
+        ),
     ]
     user_ids = [user["id"] for user in users]
     try:
@@ -133,9 +131,19 @@ def test_update_users_expire(access_token):
 
         assert response.status_code == status.HTTP_200_OK
         response = client.get("/api/users", headers={"Authorization": f"Bearer {access_token}"})
-        listed = {u["username"]: u for u in response.json()["users"] if u["username"] in {users[0]["username"], users[1]["username"]}}
-        assert dt.fromisoformat(listed[users[0]["username"]]["expire"]).replace(tzinfo=None).strftime("%Y-%m-%dT%H:%M:%S") == "2025-01-01T01:00:00"
-        assert dt.fromisoformat(listed[users[1]["username"]]["expire"]).replace(tzinfo=None).strftime("%Y-%m-%dT%H:%M:%S") == "2026-01-01T01:00:00"
+        listed = {
+            u["username"]: u
+            for u in response.json()["users"]
+            if u["username"] in {users[0]["username"], users[1]["username"]}
+        }
+        assert (
+            dt.fromisoformat(listed[users[0]["username"]]["expire"]).replace(tzinfo=None).strftime("%Y-%m-%dT%H:%M:%S")
+            == "2025-01-01T01:00:00"
+        )
+        assert (
+            dt.fromisoformat(listed[users[1]["username"]]["expire"]).replace(tzinfo=None).strftime("%Y-%m-%dT%H:%M:%S")
+            == "2026-01-01T01:00:00"
+        )
     finally:
         cleanup(access_token, core, groups, users)
 
@@ -156,7 +164,11 @@ def test_update_users_proxy_settings(access_token):
 
         assert response.status_code == status.HTTP_200_OK
         response = client.get("/api/users", headers={"Authorization": f"Bearer {access_token}"})
-        listed = {u["username"]: u for u in response.json()["users"] if u["username"] in {users[0]["username"], users[1]["username"]}}
+        listed = {
+            u["username"]: u
+            for u in response.json()["users"]
+            if u["username"] in {users[0]["username"], users[1]["username"]}
+        }
         assert listed[users[0]["username"]]["proxy_settings"]["vless"]["flow"] == "xtls-rprx-vision"
         assert listed[users[1]["username"]]["proxy_settings"]["vless"]["flow"] == "xtls-rprx-vision"
     finally:

tests/api/test_core.py

@@ -5,7 +5,6 @@
 from tests.api.sample_data import XRAY_CONFIG as xray_config
 
 
-
 def test_core_create(access_token):
     """Test that the core create route is accessible."""
 

tests/api/test_group.py

@@ -17,9 +17,7 @@ def test_group_create(access_token):
         for _ in range(3):
             k = min(3, len(inbounds))
             selected_inbounds = random.sample(inbounds, k=k) if k else inbounds
-            response = create_group(
-                access_token, name=unique_name("testgroup"), inbound_tags=selected_inbounds
-            )
+            response = create_group(access_token, name=unique_name("testgroup"), inbound_tags=selected_inbounds)
             created_groups.append(response["id"])
             assert response["name"].startswith("testgroup")
             assert set(response["inbound_tags"]) == set(selected_inbounds)

tests/test_record_usages.py

@@ -269,9 +269,7 @@ async def fake_get_outbounds_stats(node: DummyNode):
         assert node_totals[node_one_id][0] > node_totals[node_two_id][0]
         assert node_totals[node_two_id][1] > 0
 
-        node_usage_rows = await session.execute(
-            select(NodeUsage.node_id, NodeUsage.uplink, NodeUsage.downlink)
-        )
+        node_usage_rows = await session.execute(select(NodeUsage.node_id, NodeUsage.uplink, NodeUsage.downlink))
         node_usage_totals = {row.node_id: (row.uplink, row.downlink) for row in node_usage_rows.all()}
         assert set(node_usage_totals.keys()) == {node_one_id, node_two_id}