feat(node): default/internal timeout

Author: M03ED

app/db/migrations/versions/a0a1125e46b1_node_default_timeout.py

@@ -0,0 +1,30 @@
+"""node default timeout
+
+Revision ID: a0a1125e46b1
+Revises: 99076844dee6
+Create Date: 2025-11-19 17:53:48.835236
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'a0a1125e46b1'
+down_revision = '99076844dee6'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('nodes', sa.Column('default_timeout', sa.Integer(), server_default=sa.text('(10)'), nullable=False))
+    op.add_column('nodes', sa.Column('internal_timeout', sa.Integer(), server_default=sa.text('(15)'), nullable=False))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('nodes', 'internal_timeout')
+    op.drop_column('nodes', 'default_timeout')
+    # ### end Alembic commands ###

app/db/models.py

@@ -522,6 +522,8 @@ class Node(Base):
         server_default=NodeConnectionType.grpc.name,
     )
     keep_alive: Mapped[int] = mapped_column(unique=False, default=0)
+    default_timeout: Mapped[int] = mapped_column(default=10, server_default=text("10"))
+    internal_timeout: Mapped[int] = mapped_column(default=15, server_default=text("15"))
 
     @hybrid_property
     def reseted_uplink(self) -> int:

app/jobs/node_checker.py

@@ -24,7 +24,7 @@ async def verify_node_backend_health(node: PasarGuardNode, node_name: str) -> He
     Verify node health by checking backend stats.
     Returns updated health status.
     """
-    current_health = await node.get_health()
+    current_health = await asyncio.wait_for(node.get_health(), timeout=10)
 
     # Skip nodes that are not connected or invalid
     if current_health in (Health.NOT_CONNECTED, Health.INVALID):
@@ -55,8 +55,8 @@ async def update_node_connection_status(node_id: int, node: PasarGuardNode):
     Update node connection status by getting backend stats and version info.
     """
     try:
-        await asyncio.wait_for(node.get_backend_stats(), timeout=20)
-        node_version, core_version = await asyncio.wait_for(node.get_versions(), timeout=20)
+        await node.get_backend_stats()
+        node_version, core_version = await asyncio.wait_for(node.get_versions(), timeout=10)
         async with GetDB() as db:
             await NodeOperation._update_single_node_status(
                 db,
@@ -66,7 +66,7 @@ async def update_node_connection_status(node_id: int, node: PasarGuardNode):
                 node_version=node_version,
             )
     except asyncio.TimeoutError:
-        logger.warning(f"Node {node_id} connection status check timed out, will retry on next check")
+        logger.warning(f"Node {node_id} get versions timed out, will retry on next check")
         return
     except NodeAPIError as e:
         if e.code > -3:
@@ -89,7 +89,7 @@ async def process_node_health_check(db_node: Node, node: PasarGuardNode):
         return
 
     try:
-        health = await asyncio.wait_for(verify_node_backend_health(node, db_node.name), timeout=20)
+        health = await verify_node_backend_health(node, db_node.name)
     except asyncio.TimeoutError:
         if db_node.status == NodeStatus.connected:
             logger.warning(

app/models/node.py

@@ -40,6 +40,8 @@ class Node(BaseModel):
     data_limit: int = Field(default=0)
     data_limit_reset_strategy: DataLimitResetStrategy = Field(default=DataLimitResetStrategy.no_reset)
     reset_time: int = Field(default=-1)
+    default_timeout: int = Field(default=10)
+    internal_timeout: int = Field(default=15)
 
 
 class NodeCreate(Node):
@@ -161,6 +163,8 @@ class NodeModify(NodeCreate):
     data_limit: int | None = None
     data_limit_reset_strategy: DataLimitResetStrategy | None = None
     reset_time: int | None = None
+    default_timeout: int | None = None
+    internal_timeout: int | None = None
 
     model_config = ConfigDict(
         json_schema_extra={

app/node/__init__.py

@@ -42,6 +42,8 @@ async def update_node(self, node: Node) -> PasarGuardNode:
                 api_key=node.api_key,
                 name=node.name,
                 logger=self.logger,
+                default_timeout=node.default_timeout,
+                internal_timeout=node.internal_timeout,
                 extra={"id": node.id, "usage_coefficient": node.usage_coefficient},
             )
 

tests/api/test_node.py

@@ -63,6 +63,8 @@ def sample_node_response(**overrides) -> NodeResponse:
         "data_limit": 0,
         "data_limit_reset_strategy": DataLimitResetStrategy.no_reset,
         "reset_time": -1,
+        "default_timeout": 10,
+        "internal_timeout": 15,
         "xray_version": "1.7.5",
         "node_version": "1.7.5",
         "status": NodeStatus.connected,
@@ -90,6 +92,8 @@ def node_create_payload(**overrides) -> dict:
         "data_limit": 0,
         "data_limit_reset_strategy": DataLimitResetStrategy.no_reset.value,
         "reset_time": -1,
+        "default_timeout": 10,
+        "internal_timeout": 15,
     }
     payload.update(overrides)
     return payload