mirror of
https://github.com/fawney19/Aether.git
synced 2026-09-03 01:40:21 +08:00
fix(proxy-node): 修复服务重启后 tunnel 连接状态不一致的问题
- 启动时重置 DB 中残留的 tunnel_connected=True 状态 - health_scheduler 以 TunnelManager 内存实际连接为准判断节点状态 - tunnel 模式注册时初始状态设为 UNHEALTHY,等 tunnel 连接后再上线
This commit is contained in:
@@ -20,6 +20,51 @@ if TYPE_CHECKING:
|
|||||||
from sqlalchemy.orm import Session
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
|
||||||
|
def _reset_tunnel_connected_on_startup() -> None:
|
||||||
|
"""服务端启动时将所有 tunnel_connected=True 的节点重置为 False/UNHEALTHY。
|
||||||
|
|
||||||
|
服务端重启后 TunnelManager 内存状态丢失,但 DB 中可能残留
|
||||||
|
tunnel_connected=True 的记录。如果不重置,health_scheduler 会错误地
|
||||||
|
将这些节点标记为 ONLINE,而实际上 tunnel 并未连接。
|
||||||
|
"""
|
||||||
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
from src.core.logger import logger
|
||||||
|
from src.database import create_session
|
||||||
|
from src.models.database import ProxyNode, ProxyNodeStatus
|
||||||
|
|
||||||
|
db = create_session()
|
||||||
|
try:
|
||||||
|
now = datetime.now(timezone.utc)
|
||||||
|
stale_nodes = (
|
||||||
|
db.query(ProxyNode)
|
||||||
|
.filter(
|
||||||
|
ProxyNode.tunnel_connected == True, # noqa: E712
|
||||||
|
ProxyNode.is_manual == False, # noqa: E712
|
||||||
|
)
|
||||||
|
.all()
|
||||||
|
)
|
||||||
|
if stale_nodes:
|
||||||
|
for node in stale_nodes:
|
||||||
|
node.tunnel_connected = False
|
||||||
|
node.tunnel_connected_at = now
|
||||||
|
node.status = ProxyNodeStatus.UNHEALTHY
|
||||||
|
node.updated_at = now
|
||||||
|
db.commit()
|
||||||
|
logger.info(
|
||||||
|
"重置 {} 个节点的残留 tunnel 连接状态 (tunnel_connected -> False)",
|
||||||
|
len(stale_nodes),
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
try:
|
||||||
|
db.rollback()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
logger.warning("重置 tunnel 连接状态失败: {}", e)
|
||||||
|
finally:
|
||||||
|
db.close()
|
||||||
|
|
||||||
|
|
||||||
def _get_router() -> Any:
|
def _get_router() -> Any:
|
||||||
"""延迟导入路由"""
|
"""延迟导入路由"""
|
||||||
from src.api.admin.proxy_nodes import router
|
from src.api.admin.proxy_nodes import router
|
||||||
@@ -36,6 +81,11 @@ async def _on_startup() -> None:
|
|||||||
|
|
||||||
logger = logging.getLogger("aether.modules.proxy_nodes")
|
logger = logging.getLogger("aether.modules.proxy_nodes")
|
||||||
|
|
||||||
|
# 服务端启动时,TunnelManager 内存为空,所有 tunnel 连接都需要重新建立。
|
||||||
|
# 重置 DB 中残留的 tunnel_connected=True 状态,避免 health_scheduler
|
||||||
|
# 误将未连接的节点标记为 ONLINE。
|
||||||
|
_reset_tunnel_connected_on_startup()
|
||||||
|
|
||||||
from src.clients import get_redis_client
|
from src.clients import get_redis_client
|
||||||
|
|
||||||
redis_client = await get_redis_client()
|
redis_client = await get_redis_client()
|
||||||
|
|||||||
@@ -53,6 +53,9 @@ class ProxyNodeHealthScheduler:
|
|||||||
await self._check_heartbeats()
|
await self._check_heartbeats()
|
||||||
|
|
||||||
async def _check_heartbeats(self) -> None:
|
async def _check_heartbeats(self) -> None:
|
||||||
|
from src.services.proxy_node.tunnel_manager import get_tunnel_manager
|
||||||
|
|
||||||
|
manager = get_tunnel_manager()
|
||||||
db = create_session()
|
db = create_session()
|
||||||
try:
|
try:
|
||||||
now = datetime.now(timezone.utc)
|
now = datetime.now(timezone.utc)
|
||||||
@@ -71,7 +74,20 @@ class ProxyNodeHealthScheduler:
|
|||||||
|
|
||||||
changed = 0
|
changed = 0
|
||||||
for node in nodes:
|
for node in nodes:
|
||||||
if node.tunnel_connected:
|
# 以 TunnelManager 内存中的实际连接状态为准,
|
||||||
|
# 而非仅依赖 DB 的 tunnel_connected 字段。
|
||||||
|
# 服务端重启后 DB 可能残留 tunnel_connected=True,
|
||||||
|
# 但 TunnelManager 内存中已无连接。
|
||||||
|
actually_connected = manager.has_tunnel(node.id)
|
||||||
|
|
||||||
|
# 同步修正 DB 中不一致的 tunnel_connected 字段
|
||||||
|
if node.tunnel_connected != actually_connected:
|
||||||
|
node.tunnel_connected = actually_connected
|
||||||
|
if not actually_connected:
|
||||||
|
node.tunnel_connected_at = now
|
||||||
|
changed += 1
|
||||||
|
|
||||||
|
if actually_connected:
|
||||||
new_status = ProxyNodeStatus.ONLINE
|
new_status = ProxyNodeStatus.ONLINE
|
||||||
elif node.tunnel_connected_at:
|
elif node.tunnel_connected_at:
|
||||||
# tunnel 刚断开:给 60s 缓冲期标记为 UNHEALTHY
|
# tunnel 刚断开:给 60s 缓冲期标记为 UNHEALTHY
|
||||||
|
|||||||
@@ -210,10 +210,14 @@ class ProxyNodeService:
|
|||||||
)
|
)
|
||||||
else:
|
else:
|
||||||
node = db.query(ProxyNode).filter(ProxyNode.ip == ip, ProxyNode.port == port).first()
|
node = db.query(ProxyNode).filter(ProxyNode.ip == ip, ProxyNode.port == port).first()
|
||||||
|
# tunnel 模式注册时设为 UNHEALTHY,等 WebSocket tunnel 真正连接后
|
||||||
|
# 由 _update_tunnel_status 设为 ONLINE;非 tunnel 模式保持原逻辑
|
||||||
|
initial_status = ProxyNodeStatus.UNHEALTHY if tunnel_mode else ProxyNodeStatus.ONLINE
|
||||||
|
|
||||||
if node:
|
if node:
|
||||||
node.name = name
|
node.name = name
|
||||||
node.region = region
|
node.region = region
|
||||||
node.status = ProxyNodeStatus.ONLINE
|
node.status = initial_status
|
||||||
node.last_heartbeat_at = now
|
node.last_heartbeat_at = now
|
||||||
node.heartbeat_interval = heartbeat_interval
|
node.heartbeat_interval = heartbeat_interval
|
||||||
node.tunnel_mode = tunnel_mode
|
node.tunnel_mode = tunnel_mode
|
||||||
@@ -234,7 +238,7 @@ class ProxyNodeService:
|
|||||||
ip=ip,
|
ip=ip,
|
||||||
port=port,
|
port=port,
|
||||||
region=region,
|
region=region,
|
||||||
status=ProxyNodeStatus.ONLINE,
|
status=initial_status,
|
||||||
registered_by=registered_by,
|
registered_by=registered_by,
|
||||||
last_heartbeat_at=now,
|
last_heartbeat_at=now,
|
||||||
heartbeat_interval=heartbeat_interval,
|
heartbeat_interval=heartbeat_interval,
|
||||||
|
|||||||
Reference in New Issue
Block a user