feat(proxy-tunnel): 增强隧道连接稳定性与恢复速度

- writer 增加 WebSocket Ping keepalive，防止中间代理空闲超时断开
- 服务端增加应用层 PING 循环（30s 间隔），空闲超时延长至 180s
- 重连基础延迟从 1000ms 降低到 500ms
- 节点缓存 TTL 缩短至 15s，不可用节点 TTL 缩短至 5s 加速恢复感知
- 心跳检测间隔从 30s 缩短到 15s

aether-proxy/src/config.rs

@@ -206,7 +206,7 @@ pub struct Config {
     #[arg(
         long,
         env = "AETHER_PROXY_TUNNEL_RECONNECT_BASE_MS",
-        default_value_t = 1000
+        default_value_t = 500
     )]
     pub tunnel_reconnect_base_ms: u64,
 

aether-proxy/src/tunnel/client.rs

@@ -54,8 +54,9 @@ pub async fn connect_and_run(
     // Split into read/write halves
     let (ws_sink, ws_read) = futures_util::StreamExt::split(ws_stream);
 
-    // Spawn writer task
+    // Spawn writer task (with WebSocket ping keepalive)
-    let (frame_tx, mut writer_handle) = writer::spawn_writer(ws_sink);
+    let ping_interval = Duration::from_secs(state.config.tunnel_ping_interval_secs);
+    let (frame_tx, mut writer_handle) = writer::spawn_writer(ws_sink, ping_interval);
 
     // Spawn heartbeat task
     let hb_handle = heartbeat::spawn(

aether-proxy/src/tunnel/writer.rs

@@ -1,13 +1,17 @@
 //! Dedicated WebSocket writer task.
 //!
 //! All frame writes go through an mpsc channel to a single writer task,
-//! avoiding contention on the WebSocket sink.
+//! avoiding contention on the WebSocket sink.  The writer also sends
+//! periodic WebSocket Ping frames to keep the connection alive through
+//! intermediary proxies (Nginx, Cloudflare, etc.).
+
+use std::time::Duration;
 
 use futures_util::SinkExt;
 use tokio::sync::mpsc;
 use tokio::task::JoinHandle;
 use tokio_tungstenite::tungstenite::Message;
-use tracing::{debug, error};
+use tracing::{debug, error, trace};
 
 use super::protocol::Frame;
 
@@ -15,18 +19,40 @@ use super::protocol::Frame;
 pub type FrameSender = mpsc::Sender<Frame>;
 
 /// Spawn the writer task. Returns the sender and a JoinHandle for cleanup.
-pub fn spawn_writer<S>(mut sink: S) -> (FrameSender, JoinHandle<()>)
+///
+/// `ping_interval` controls WebSocket-level Ping frequency (typically 15s).
+/// This keeps the connection alive through intermediary proxies/load-balancers.
+pub fn spawn_writer<S>(mut sink: S, ping_interval: Duration) -> (FrameSender, JoinHandle<()>)
 where
     S: SinkExt<Message, Error = tokio_tungstenite::tungstenite::Error> + Unpin + Send + 'static,
 {
     let (tx, mut rx) = mpsc::channel::<Frame>(256);
 
     let handle = tokio::spawn(async move {
-        while let Some(frame) = rx.recv().await {
+        let mut ping_ticker = tokio::time::interval(ping_interval);
-            let data = frame.encode();
+        ping_ticker.tick().await; // skip first immediate tick
-            if let Err(e) = sink.send(Message::Binary(data.into())).await {
+
-                error!(error = %e, "failed to write frame to WebSocket");
+        loop {
-                break;
+            tokio::select! {
+                frame = rx.recv() => {
+                    match frame {
+                        Some(frame) => {
+                            let data = frame.encode();
+                            if let Err(e) = sink.send(Message::Binary(data.into())).await {
+                                error!(error = %e, "failed to write frame to WebSocket");
+                                break;
+                            }
+                        }
+                        None => break, // all senders dropped
+                    }
+                }
+                _ = ping_ticker.tick() => {
+                    if let Err(e) = sink.send(Message::Ping(vec![])).await {
+                        error!(error = %e, "failed to send WebSocket ping");
+                        break;
+                    }
+                    trace!("sent WebSocket ping");
+                }
             }
         }
         debug!("writer task exiting");

src/api/admin/proxy_tunnel.py

@@ -16,15 +16,18 @@ from src.services.proxy_node.tunnel_manager import (
     TunnelConnection,
     get_tunnel_manager,
 )
-from src.services.proxy_node.tunnel_protocol import Frame
+from src.services.proxy_node.tunnel_protocol import Frame, MsgType
 
 router = APIRouter()
 
 # 单帧最大 64 MB -- AI API 请求体可能包含多张 base64 图片，需要足够余量
 _MAX_FRAME_SIZE = 64 * 1024 * 1024
 
-# WebSocket 空闲超时（秒）-- proxy 端 ping 间隔默认 15s，3 倍余量
+# WebSocket 空闲超时（秒）-- proxy 端 WebSocket ping 间隔 15s + 心跳 30s，180s 提供充足余量
-_IDLE_TIMEOUT = 90.0
+_IDLE_TIMEOUT = 180.0
+
+# 服务端应用层 ping 间隔（秒）-- 确保即使 proxy 端心跳延迟，连接也不会因中间代理空闲超时而断开
+_SERVER_PING_INTERVAL = 30.0
 
 
 async def _authenticate(ws: WebSocket) -> tuple[str, str] | None:
@@ -99,6 +102,9 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
     # 更新 DB: tunnel_connected = True
     await _update_tunnel_status(node_id, connected=True)
 
+    # 启动服务端 ping 任务，防止中间代理因空闲超时关闭连接
+    ping_task = asyncio.create_task(_ping_loop(conn))
+
     try:
         oversized_count = 0
         while True:
@@ -130,10 +136,27 @@ async def proxy_tunnel_ws(ws: WebSocket) -> None:
     except Exception as e:
         logger.error("tunnel WebSocket error for node_id={}: {}", node_id, e)
     finally:
+        ping_task.cancel()
         manager.unregister(node_id)
         await _update_tunnel_status(node_id, connected=False)
 
 
+async def _ping_loop(conn: TunnelConnection) -> None:
+    """定期发送应用层 PING 帧，保持连接活跃"""
+    try:
+        while True:
+            await asyncio.sleep(_SERVER_PING_INTERVAL)
+            if not conn.is_alive:
+                break
+            try:
+                await conn.send_frame(Frame(0, MsgType.PING, 0, b""))
+            except Exception as e:
+                logger.debug("ping loop send failed for node_id={}: {}", conn.node_id, e)
+                break
+    except asyncio.CancelledError:
+        pass
+
+
 async def _update_tunnel_status(node_id: str, *, connected: bool) -> None:
     """更新 ProxyNode 的 tunnel 连接状态（在线程池中执行，避免阻塞 event loop）"""
 

src/services/proxy_node/health_scheduler.py

@@ -35,7 +35,7 @@ class ProxyNodeHealthScheduler:
         scheduler = get_scheduler()
         scheduler.add_interval_job(
             self._scheduled_check,
-            seconds=30,
+            seconds=15,
             job_id="proxy_node_health_check",
             name="代理节点心跳检测",
         )

src/services/proxy_node/resolver.py

@@ -21,7 +21,8 @@ from src.core.logger import logger
 # ProxyNode 信息缓存（降低高频 DB 查询开销）
 # ---------------------------------------------------------------------------
 _proxy_node_cache: dict[str, tuple[dict[str, Any] | None, float]] = {}
-_PROXY_NODE_CACHE_TTL_SECONDS = 60.0
+_PROXY_NODE_CACHE_TTL_SECONDS = 15.0
+_PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS = 5.0  # 不可用节点使用更短的 TTL，加速恢复感知
 _PROXY_NODE_CACHE_MAX_SIZE = 256
 
 
@@ -60,12 +61,12 @@ def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
     try:
         node = db.query(ProxyNode).filter(ProxyNode.id == node_id).first()
         if not node or node.status != ProxyNodeStatus.ONLINE:
-            _proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_TTL_SECONDS)
+            _proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS)
             return None
 
         # tunnel 模式节点必须 tunnel 已连接才可用
         if node.tunnel_mode and not node.tunnel_connected:
-            _proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_TTL_SECONDS)
+            _proxy_node_cache[node_id] = (None, now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS)
             return None
 
         if node.is_manual: