feat(tunnel): 重连指数退避、多worker兼容与手动节点请求计数

- Proxy tunnel 重连策略从固定1s改为指数退避+jitter，首次重试立即执行，
  稳定连接30s后重置退避计数，上限3s保证快速恢复
- 多连接启动时增加错峰延迟，避免同时发起连接风暴
- 服务端 tunnel ping间隔和空闲超时支持环境变量配置
- 修复多worker启动时tunnel状态重置逻辑，仅leader执行重置避免覆盖其他worker连接
- Resolver增加本地tunnel缺失的限频告警和更短缓存TTL，加速多worker场景恢复
- 用量记录中统计手动代理节点的请求数和失败数

aether-proxy/src/config.rs

@@ -202,21 +202,19 @@ pub struct Config {
     #[arg(long, env = "AETHER_PROXY_LOG_JSON", default_value_t = false)]
     pub log_json: bool,
 
-    /// Deprecated: reconnect now uses a fixed 1s delay. Kept for config compatibility.
+    /// Tunnel reconnect base delay in milliseconds (used by exponential backoff)
     #[arg(
         long,
         env = "AETHER_PROXY_TUNNEL_RECONNECT_BASE_MS",
-        default_value_t = 500,
+        default_value_t = 500
-        hide = true
     )]
     pub tunnel_reconnect_base_ms: u64,
 
-    /// Deprecated: reconnect now uses a fixed 1s delay. Kept for config compatibility.
+    /// Tunnel reconnect max delay in milliseconds (cap for exponential backoff)
     #[arg(
         long,
         env = "AETHER_PROXY_TUNNEL_RECONNECT_MAX_MS",
-        default_value_t = 30000,
+        default_value_t = 30000
-        hide = true
     )]
     pub tunnel_reconnect_max_ms: u64,
 

aether-proxy/src/tunnel/mod.rs

@@ -6,16 +6,26 @@ pub mod stream_handler;
 pub mod writer;
 
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use tokio::sync::watch;
 use tracing::{error, info};
 
 use crate::state::{AppState, ServerContext};
 
-/// Fixed reconnect delay -- short enough for fast recovery, long enough to
+/// If a tunnel stays connected at least this long, treat the next disconnect
-/// avoid CPU spin when the network is completely down.
+/// as a non-failure and reset reconnect backoff.
-const RECONNECT_DELAY: Duration = Duration::from_secs(1);
+const STABLE_SESSION_RESET_AFTER: Duration = Duration::from_secs(30);
+/// Startup staggering step per secondary connection, used to avoid
+/// simultaneous bursts when a pool of tunnels starts together.
+const STARTUP_STAGGER_STEP_MS: u64 = 150;
+/// Upper bound for startup staggering.
+const MAX_STARTUP_STAGGER_MS: u64 = 1_500;
+/// Keep a tiny floor for repeated reconnects; first retry is still immediate.
+const MIN_RECONNECT_DELAY_MS: u64 = 50;
+/// Even under sustained failures, keep probing frequently so recovery is fast
+/// once cross-border network quality improves.
+const RECONNECT_PROBE_MAX_DELAY_MS: u64 = 3_000;
 
 /// Run the tunnel mode main loop (connect, dispatch, reconnect).
 ///
@@ -28,8 +38,29 @@ pub async fn run(
     mut shutdown: watch::Receiver<bool>,
 ) {
     info!(server = %server.server_label, conn = conn_idx, "starting tunnel");
+    let reconnect_salt = compute_connection_salt(server, conn_idx);
+
+    let startup_delay = compute_startup_stagger(conn_idx, reconnect_salt);
+    if !startup_delay.is_zero() {
+        info!(
+            server = %server.server_label,
+            conn = conn_idx,
+            delay_ms = startup_delay.as_millis(),
+            "startup stagger before first connect"
+        );
+        tokio::select! {
+            _ = tokio::time::sleep(startup_delay) => {}
+            _ = shutdown.changed() => {
+                info!(server = %server.server_label, conn = conn_idx, "shutdown requested during startup stagger");
+                return;
+            }
+        }
+    }
+
+    let mut consecutive_failures: u32 = 0;
 
     loop {
+        let started_at = Instant::now();
         match client::connect_and_run(state, server, conn_idx, &mut shutdown).await {
             Ok(client::TunnelOutcome::Shutdown) => {
                 info!(server = %server.server_label, conn = conn_idx, "tunnel shut down gracefully");
@@ -48,8 +79,31 @@ pub async fn run(
             return;
         }
 
+        // Reset backoff after a stable session to keep recovery snappy when
+        // failures are only occasional.
+        let connected_for = started_at.elapsed();
+        if connected_for >= STABLE_SESSION_RESET_AFTER {
+            consecutive_failures = 0;
+        } else {
+            consecutive_failures = consecutive_failures.saturating_add(1);
+        }
+
+        let reconnect_delay = compute_reconnect_delay(
+            state.config.tunnel_reconnect_base_ms,
+            state.config.tunnel_reconnect_max_ms,
+            consecutive_failures,
+            reconnect_salt,
+        );
+        info!(
+            server = %server.server_label,
+            conn = conn_idx,
+            failures = consecutive_failures,
+            delay_ms = reconnect_delay.as_millis(),
+            "waiting before reconnect"
+        );
+
         tokio::select! {
-            _ = tokio::time::sleep(RECONNECT_DELAY) => {}
+            _ = tokio::time::sleep(reconnect_delay) => {}
             _ = shutdown.changed() => {
                 info!(server = %server.server_label, conn = conn_idx, "shutdown requested during reconnect wait");
                 return;
@@ -57,3 +111,126 @@ pub async fn run(
         }
     }
 }
+
+fn compute_connection_salt(server: &ServerContext, conn_idx: usize) -> u64 {
+    // FNV-1a style hash over server label + connection index.
+    let mut h: u64 = 0xcbf29ce484222325;
+    for &b in server.server_label.as_bytes() {
+        h ^= b as u64;
+        h = h.wrapping_mul(0x100000001b3);
+    }
+    h ^= conn_idx as u64;
+    mix_u64(h)
+}
+
+fn compute_startup_stagger(conn_idx: usize, salt: u64) -> Duration {
+    if conn_idx == 0 {
+        return Duration::ZERO;
+    }
+    let base = (conn_idx as u64).saturating_mul(STARTUP_STAGGER_STEP_MS);
+    let jitter = mix_u64(salt) % 301; // 0..=300ms
+    Duration::from_millis((base + jitter).min(MAX_STARTUP_STAGGER_MS))
+}
+
+fn compute_reconnect_delay(
+    base_ms: u64,
+    max_ms: u64,
+    consecutive_failures: u32,
+    salt: u64,
+) -> Duration {
+    // First retry should be immediate to maximize recovery speed on transient
+    // blips (the user's primary expectation in poor networks).
+    if consecutive_failures <= 1 {
+        return Duration::ZERO;
+    }
+
+    // Keep a sane minimum for repeated failures.
+    let base_ms = base_ms.max(MIN_RECONNECT_DELAY_MS);
+    let max_ms = max_ms.max(base_ms);
+    let cap_ms = compute_reconnect_cap_ms(base_ms, max_ms, consecutive_failures)
+        .min(RECONNECT_PROBE_MAX_DELAY_MS.max(base_ms));
+
+    // Equal-jitter: randomize in [cap/2, cap], preventing synchronized reconnect
+    // storms while keeping reconnect latency bounded.
+    if cap_ms <= 1 {
+        return Duration::from_millis(cap_ms);
+    }
+
+    let half = cap_ms / 2;
+    let span = cap_ms - half;
+    let now_nanos = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.subsec_nanos() as u64)
+        .unwrap_or(0);
+    let mixed = mix_u64(now_nanos ^ salt);
+    let jitter = if span == 0 { 0 } else { mixed % (span + 1) };
+    Duration::from_millis(half + jitter)
+}
+
+fn compute_reconnect_cap_ms(base_ms: u64, max_ms: u64, consecutive_failures: u32) -> u64 {
+    if consecutive_failures <= 1 {
+        return base_ms.min(max_ms);
+    }
+
+    let shift = (consecutive_failures - 1).min(31);
+    let factor = 1u64 << shift;
+    base_ms.saturating_mul(factor).min(max_ms)
+}
+
+fn mix_u64(mut x: u64) -> u64 {
+    // SplitMix64 finalizer - cheap bit mixing for pseudo-random jitter.
+    x ^= x >> 30;
+    x = x.wrapping_mul(0xbf58476d1ce4e5b9);
+    x ^= x >> 27;
+    x = x.wrapping_mul(0x94d049bb133111eb);
+    x ^ (x >> 31)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use super::{
+        compute_reconnect_cap_ms, compute_reconnect_delay, compute_startup_stagger,
+        MAX_STARTUP_STAGGER_MS, RECONNECT_PROBE_MAX_DELAY_MS, STARTUP_STAGGER_STEP_MS,
+    };
+
+    #[test]
+    fn reconnect_cap_grows_exponentially_and_caps() {
+        let base = 500;
+        let max = 30_000;
+        assert_eq!(compute_reconnect_cap_ms(base, max, 0), 500);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 1), 500);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 2), 1_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 3), 2_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 4), 4_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 5), 8_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 6), 16_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 7), 30_000);
+        assert_eq!(compute_reconnect_cap_ms(base, max, 20), 30_000);
+    }
+
+    #[test]
+    fn startup_stagger_is_zero_for_primary_and_bounded_for_secondary() {
+        assert_eq!(compute_startup_stagger(0, 42), Duration::ZERO);
+
+        let d1 = compute_startup_stagger(1, 42);
+        let d2 = compute_startup_stagger(2, 42);
+
+        assert!(d1 >= Duration::from_millis(STARTUP_STAGGER_STEP_MS));
+        assert!(d1 <= Duration::from_millis(MAX_STARTUP_STAGGER_MS));
+        assert!(d2 >= Duration::from_millis(STARTUP_STAGGER_STEP_MS * 2));
+        assert!(d2 <= Duration::from_millis(MAX_STARTUP_STAGGER_MS));
+    }
+
+    #[test]
+    fn reconnect_delay_is_immediate_on_first_failure() {
+        assert_eq!(compute_reconnect_delay(700, 45_000, 1, 123), Duration::ZERO);
+    }
+
+    #[test]
+    fn reconnect_delay_stays_within_probe_ceiling_after_many_failures() {
+        let d = compute_reconnect_delay(500, 45_000, 100, 12345);
+        assert!(d <= Duration::from_millis(RECONNECT_PROBE_MAX_DELAY_MS));
+    }
+}

src/api/admin/proxy_tunnel.py

@@ -8,6 +8,7 @@ aether-proxy 通过此端点建立 tunnel 连接。
 from __future__ import annotations
 
 import asyncio
+import os
 from datetime import datetime, timezone
 
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect
@@ -37,11 +38,58 @@ def _get_node_lock(node_id: str) -> asyncio.Lock:
 # 单帧最大 64 MB -- AI API 请求体可能包含多张 base64 图片，需要足够余量
 _MAX_FRAME_SIZE = 64 * 1024 * 1024
 
-# WebSocket 空闲超时（秒）-- 需覆盖客户端 stale_timeout(45s) + 重连延迟(最长30s) 的窗口期
+# 默认 WebSocket 空闲超时（秒）-- 需覆盖客户端 stale_timeout(45s) + 重连延迟窗口
-_IDLE_TIMEOUT = 90.0
+_DEFAULT_IDLE_TIMEOUT = 90.0
 
-# 服务端应用层 ping 间隔（秒）-- 与客户端 ping 间隔一致，确保高频心跳
+# 默认服务端应用层 ping 间隔（秒）-- 与客户端 ping 间隔一致，确保高频心跳
-_SERVER_PING_INTERVAL = 15.0
+_DEFAULT_SERVER_PING_INTERVAL = 15.0
+
+
+def _env_float(name: str, default: float, *, min_value: float, max_value: float) -> float:
+    """读取并校验浮点环境变量，非法时回退默认值。"""
+    raw = os.getenv(name, "").strip()
+    if not raw:
+        return default
+    try:
+        value = float(raw)
+    except ValueError:
+        logger.warning("invalid {}={}, fallback to {}", name, raw, default)
+        return default
+    if value < min_value or value > max_value:
+        logger.warning(
+            "{}={} out of range [{}, {}], fallback to {}",
+            name,
+            value,
+            min_value,
+            max_value,
+            default,
+        )
+        return default
+    return value
+
+
+# 允许通过环境变量在弱网环境下调大容忍窗口（无需改代码）
+_SERVER_PING_INTERVAL = _env_float(
+    "AETHER_PROXY_TUNNEL_SERVER_PING_INTERVAL",
+    _DEFAULT_SERVER_PING_INTERVAL,
+    min_value=5.0,
+    max_value=120.0,
+)
+_IDLE_TIMEOUT = _env_float(
+    "AETHER_PROXY_TUNNEL_SERVER_IDLE_TIMEOUT",
+    _DEFAULT_IDLE_TIMEOUT,
+    min_value=30.0,
+    max_value=600.0,
+)
+
+# 避免 idle timeout 过小导致 ping 尚未生效就被服务端断开
+if _IDLE_TIMEOUT <= _SERVER_PING_INTERVAL * 2:
+    adjusted_idle = max(_SERVER_PING_INTERVAL * 3, 30.0)
+    logger.warning(
+        "AETHER_PROXY_TUNNEL_SERVER_IDLE_TIMEOUT too low for ping interval, auto-adjust to {}",
+        adjusted_idle,
+    )
+    _IDLE_TIMEOUT = adjusted_idle
 
 
 async def _authenticate(ws: WebSocket) -> tuple[str, str] | None:

src/modules/proxy_nodes/__init__.py

@@ -76,15 +76,18 @@ async def _on_startup() -> None:
     """启动心跳检测调度器"""
     import logging
 
+    from src.config import config
     from src.services.proxy_node.health_scheduler import get_proxy_node_health_scheduler
     from src.utils.task_coordinator import StartupTaskCoordinator
 
     logger = logging.getLogger("aether.modules.proxy_nodes")
 
-    # 服务端启动时，TunnelManager 内存为空，所有 tunnel 连接都需要重新建立。
+    if config.worker_processes > 1:
-    # 重置 DB 中残留的 tunnel_connected=True 状态，避免 health_scheduler
+        logger.warning(
-    # 误将未连接的节点标记为 ONLINE。
+            "检测到 WEB_CONCURRENCY={}。Proxy tunnel 连接是进程内资源，"
-    _reset_tunnel_connected_on_startup()
+            "多 worker 场景可能出现节点显示 ONLINE 但当前 worker 无可用 tunnel 的情况。",
+            config.worker_processes,
+        )
 
     from src.clients import get_redis_client
 
@@ -94,6 +97,9 @@ async def _on_startup() -> None:
     proxy_node_health_scheduler = get_proxy_node_health_scheduler()
     active = await task_coordinator.acquire("proxy_node_health")
     if active:
+        # 仅 leader worker 执行启动重置，避免多 worker 并发启动/重启时
+        # 把其他 worker 已建立的 tunnel 状态错误重置为 OFFLINE。
+        _reset_tunnel_connected_on_startup()
         logger.info("启动 ProxyNode 心跳检测调度器...")
         await proxy_node_health_scheduler.start()
     else:

src/services/proxy_node/resolver.py

@@ -10,6 +10,7 @@ from __future__ import annotations
 import gzip
 import hashlib
 import json
+import os
 import time
 from typing import Any
 from urllib.parse import quote, urlparse
@@ -26,7 +27,10 @@ from src.core.logger import logger
 _proxy_node_cache: dict[str, tuple[dict[str, Any] | None, float]] = {}
 _PROXY_NODE_CACHE_TTL_SECONDS = 15.0
 _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS = 5.0  # 不可用节点使用更短的 TTL，加速恢复感知
+_PROXY_NODE_CACHE_TUNNEL_LOCAL_MISS_TTL_SECONDS = 0.5  # 本地 worker 无 tunnel 时，快速重试
 _PROXY_NODE_CACHE_MAX_SIZE = 256
+_TUNNEL_LOCAL_MISS_LOG_COOLDOWN_SECONDS = 30.0
+_tunnel_local_miss_log_next_at: dict[str, float] = {}
 
 
 def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
@@ -74,9 +78,23 @@ def _get_proxy_node_info(node_id: str) -> dict[str, Any] | None:
 
             manager = get_tunnel_manager()
             if not manager.has_tunnel(node_id):
+                # 多 worker 部署时，DB 可能显示 ONLINE（其他 worker 有 tunnel），
+                # 但当前 worker 无本地连接，请求仍不可用。记录限频告警便于定位。
+                if now >= _tunnel_local_miss_log_next_at.get(node_id, 0.0):
+                    _tunnel_local_miss_log_next_at[node_id] = (
+                        now + _TUNNEL_LOCAL_MISS_LOG_COOLDOWN_SECONDS
+                    )
+                    logger.warning(
+                        "tunnel node {} has no local connection on pid={} "
+                        "(db_status={}, db_tunnel_connected={}), request may fail on this worker",
+                        node_id,
+                        os.getpid(),
+                        str(getattr(node, "status", "unknown")),
+                        bool(getattr(node, "tunnel_connected", False)),
+                    )
                 _proxy_node_cache[node_id] = (
                     None,
-                    now + _PROXY_NODE_CACHE_NEGATIVE_TTL_SECONDS,
+                    now + _PROXY_NODE_CACHE_TUNNEL_LOCAL_MISS_TTL_SECONDS,
                 )
                 return None
             value: dict[str, Any] = {
@@ -127,6 +145,7 @@ _SYSTEM_PROXY_CACHE_TTL = 60.0
 def invalidate_proxy_node_cache(node_id: str) -> None:
     """主动清除指定节点的信息缓存（tunnel 断开时调用，避免使用过期的连接状态）"""
     _proxy_node_cache.pop(node_id, None)
+    _tunnel_local_miss_log_next_at.pop(node_id, None)
 
 
 def invalidate_system_proxy_cache() -> None:
@@ -509,7 +528,10 @@ def resolve_proxy_info(proxy_config: dict[str, Any] | None) -> dict[str, Any] |
         node_id = node_id.strip()
         node_info = _get_proxy_node_info(node_id)
         node_name = node_info.get("name", "unknown") if node_info else "offline"
-        return {"node_id": node_id, "node_name": node_name, "source": source}
+        info: dict[str, Any] = {"node_id": node_id, "node_name": node_name, "source": source}
+        if node_info and node_info.get("is_manual"):
+            info["is_manual"] = True
+        return info
 
     # 旧格式 URL 模式
     proxy_url = effective_config.get("url")

src/services/usage/recording.py

@@ -7,7 +7,7 @@ from typing import Any
 from sqlalchemy.orm import Session
 
 from src.core.logger import logger
-from src.models.database import ApiKey, Provider, Usage, User, UserModelUsageCount
+from src.models.database import ApiKey, Provider, ProxyNode, Usage, User, UserModelUsageCount
 from src.services.provider_keys.codex_quota_sync_dispatcher import (
     dispatch_codex_quota_sync_from_response_headers,
 )
@@ -22,6 +22,50 @@ from src.services.usage._recording_helpers import (
 from src.services.usage._types import UsageCostInfo, UsageRecordParams
 
 
+def _extract_manual_proxy_node_id(metadata: dict[str, Any] | None) -> str | None:
+    """从 request_metadata 中提取手动代理节点 ID（仅 is_manual 节点）。
+
+    Tunnel 节点的统计由 aether-proxy 心跳上报，此处只处理手动节点以避免重复计数。
+    """
+    if not metadata:
+        return None
+    proxy = metadata.get("proxy")
+    if not isinstance(proxy, dict):
+        return None
+    if not proxy.get("is_manual"):
+        return None
+    node_id = proxy.get("node_id")
+    return node_id if isinstance(node_id, str) and node_id.strip() else None
+
+
+def _increment_proxy_node_requests(
+    db: Session,
+    node_counts: dict[str, int],
+    failed_counts: dict[str, int] | None = None,
+) -> None:
+    """批量递增手动代理节点的 total_requests 和 failed_requests（原子 SQL UPDATE）。"""
+    if not node_counts and not failed_counts:
+        return
+    from sqlalchemy import update
+
+    # 合并所有涉及的 node_id
+    all_ids = set(node_counts) | set(failed_counts or {})
+    for node_id in all_ids:
+        total = node_counts.get(node_id, 0)
+        failed = (failed_counts or {}).get(node_id, 0)
+        values: dict[str, Any] = {}
+        if total > 0:
+            values["total_requests"] = ProxyNode.total_requests + total
+        if failed > 0:
+            values["failed_requests"] = ProxyNode.failed_requests + failed
+        if values:
+            db.execute(
+                update(ProxyNode)
+                .where(ProxyNode.id == node_id, ProxyNode.is_manual == True)  # noqa: E712
+                .values(**values)
+            )
+
+
 class UsageRecordingMixin(UsageBillingIntegrationMixin):
     """记录用量相关方法"""
 
@@ -404,6 +448,12 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                 .values(monthly_used_usd=Provider.monthly_used_usd + actual_total_cost)
             )
 
+        # 更新手动代理节点请求计数（tunnel 节点由心跳上报，不在此处统计）
+        manual_node_id = _extract_manual_proxy_node_id(metadata)
+        if manual_node_id:
+            failed = {manual_node_id: 1} if status == "failed" else None
+            _increment_proxy_node_requests(db, {manual_node_id: 1}, failed)
+
         # 结算标记：终态请求写入 settled + finalized_at
         if status not in ("pending", "streaming"):
             usage.billing_status = "settled"
@@ -785,6 +835,8 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
             int
         )  # (user_id, model) -> count
         provider_costs: dict[str, float] = defaultdict(float)  # provider_id -> cost
+        proxy_node_counts: dict[str, int] = defaultdict(int)  # node_id -> request count
+        proxy_node_failed: dict[str, int] = defaultdict(int)  # node_id -> failed count
         quota_update_candidates: dict[str, dict[str, Any]] = {}
 
         # 合并所有需要处理的记录（用于预取 user/api_key）
@@ -944,6 +996,12 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                     apikey_stats[key_id]["cost"] += total_cost
                     apikey_stats[key_id]["is_standalone"] = api_key.is_standalone
 
+                manual_nid = _extract_manual_proxy_node_id(record.get("metadata"))
+                if manual_nid:
+                    proxy_node_counts[manual_nid] += 1
+                    if record.get("status") == "failed":
+                        proxy_node_failed[manual_nid] += 1
+
                 provider_api_key_id = record.get("provider_api_key_id")
                 response_headers = record.get("response_headers")
                 if (
@@ -1005,6 +1063,12 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                     apikey_stats[key_id]["cost"] += total_cost
                     apikey_stats[key_id]["is_standalone"] = api_key.is_standalone
 
+                manual_nid = _extract_manual_proxy_node_id(record.get("metadata"))
+                if manual_nid:
+                    proxy_node_counts[manual_nid] += 1
+                    if record.get("status") == "failed":
+                        proxy_node_failed[manual_nid] += 1
+
                 provider_api_key_id = record.get("provider_api_key_id")
                 response_headers = record.get("response_headers")
                 if (
@@ -1130,6 +1194,9 @@ class UsageRecordingMixin(UsageBillingIntegrationMixin):
                     )
                 )
 
+        # 批量更新手动代理节点请求计数
+        _increment_proxy_node_requests(db, proxy_node_counts, proxy_node_failed)
+
         # 配额头实时同步：同一 key 仅取本批次最后一组响应头并执行一次对比更新。
         for provider_api_key_id, response_headers in quota_update_candidates.items():
             dispatch_codex_quota_sync_from_response_headers(