diff --git a/client/lib/l10n/app_text.dart b/client/lib/l10n/app_text.dart index 4a78df1..14d8e62 100644 --- a/client/lib/l10n/app_text.dart +++ b/client/lib/l10n/app_text.dart @@ -56,6 +56,9 @@ abstract class AppText { String nodeSwitchTitle(String name); String get nodeSwitchBody; String get nodeSwitchConfirm; + // 连通看门狗:当前节点数据面不可用时的提示 + String get nodeUnhealthySwitched; // 智能选择已自动切换 + String get nodeUnhealthyError; // 手动节点:断开并提示 String get smartSub; String get recommended; diff --git a/client/lib/l10n/strings_en.dart b/client/lib/l10n/strings_en.dart index 91a0f90..ba3de27 100644 --- a/client/lib/l10n/strings_en.dart +++ b/client/lib/l10n/strings_en.dart @@ -70,6 +70,10 @@ class StringsEn extends AppText { @override String get nodeSwitchConfirm => 'Switch'; @override + String get nodeUnhealthySwitched => 'Current node unreachable — switched automatically'; + @override + String get nodeUnhealthyError => 'Current node is unreachable — reconnect or switch node'; + @override String get smartSub => 'Picks the best node for your network'; @override String get recommended => 'Recommended'; diff --git a/client/lib/l10n/strings_zh.dart b/client/lib/l10n/strings_zh.dart index b8008ed..d0852bf 100644 --- a/client/lib/l10n/strings_zh.dart +++ b/client/lib/l10n/strings_zh.dart @@ -69,6 +69,10 @@ class StringsZh extends AppText { @override String get nodeSwitchConfirm => '切换'; @override + String get nodeUnhealthySwitched => '当前节点异常,已自动切换'; + @override + String get nodeUnhealthyError => '当前节点异常,请重连或更换节点'; + @override String get smartSub => '根据当前网络环境,自动选择最优节点'; @override String get recommended => '推荐'; diff --git a/client/lib/state/connection_provider.dart b/client/lib/state/connection_provider.dart index 522b285..3cb2a65 100644 --- a/client/lib/state/connection_provider.dart +++ b/client/lib/state/connection_provider.dart @@ -6,12 +6,14 @@ // - 严禁乐观翻转:VpnPhase.on 必须由 bridge.statusStream 确认后才置。 // - 状态来源:bridge.statusStream(来自内核回调),非 Timer 模拟。 import 'dart:async'; +import 'dart:io'; import 'package:flutter_riverpod/flutter_riverpod.dart'; import '../bridge/vpn_bridge.dart'; import '../bridge/vpn_bridge_provider.dart'; import '../l10n/app_text.dart'; +import '../models/node.dart'; import '../services/api_config.dart'; import '../services/connect_api.dart'; import '../services/device_identity.dart'; @@ -53,19 +55,59 @@ class ConnectionState { int get hashCode => Object.hash(phase, elapsed, error); } +// ── 连通看门狗 ───────────────────────────────────────────────────── +// +// 已连接(本地 TUN 已起)≠ 远端节点数据面可用:节点 sing-box 崩了/数据口不通而 +// agent 仍在线时,客户端会一直显示「已连接」但流量全失败。看门狗周期性经隧道做 +// 连通性探测,连续失败即判当前节点不可用 → 智能选择自动切节点 / 手动选定的只断开提示。 + +/// 连通探测接缝:true=经隧道能到外网。默认走海外 204 端点;测试可注入假实现。 +typedef HealthProber = Future Function(); + +const _kWatchdogInterval = Duration(seconds: 15); +const _kHealthFailLimit = 3; // 连续 N 次失败(≈45s)才判定不可用,防抖 + +/// 默认探测:GET 海外 generate_204(全局 TUN 下走隧道 → 真实检验代理出网)。 +/// 任一端点 2xx/204 即视为通;多端点容错,避免单点误判。 +Future defaultHealthProbe() async { + const urls = [ + 'https://www.gstatic.com/generate_204', + 'https://cp.cloudflare.com/generate_204', + ]; + for (final u in urls) { + final client = HttpClient()..connectionTimeout = const Duration(seconds: 4); + try { + final req = await client.getUrl(Uri.parse(u)).timeout(const Duration(seconds: 5)); + final resp = await req.close().timeout(const Duration(seconds: 5)); + await resp.drain(); + if (resp.statusCode >= 200 && resp.statusCode < 300) return true; + } catch (_) { + // 该端点不通,试下一个 + } finally { + client.close(force: true); + } + } + return false; +} + // ── 状态机 ─────────────────────────────────────────────────────── class ConnectionController extends StateNotifier { - ConnectionController(this._ref, this._bridge) - : super(const ConnectionState(phase: VpnPhase.off)) { + ConnectionController(this._ref, this._bridge, {HealthProber? prober}) + : _prober = prober ?? defaultHealthProbe, + super(const ConnectionState(phase: VpnPhase.off)) { // 订阅桥状态流:状态由内核事件驱动,严禁 UI 乐观翻转。 _statusSub = _bridge.statusStream.listen(_onKernelStatus); } final Ref _ref; final VpnBridge _bridge; + final HealthProber _prober; StreamSubscription? _statusSub; Timer? _elapsed; + Timer? _watchdog; + int _healthFails = 0; + bool _probing = false; ConnectApi? _api; // ── 公有 API ─────────────────────────────────────────────────── @@ -167,16 +209,84 @@ class ConnectionController extends StateNotifier { case VpnStatus.on: state = state.copyWith(phase: VpnPhase.on); _startElapsed(); + _startWatchdog(); case VpnStatus.connecting: state = state.copyWith(phase: VpnPhase.connecting); _stopElapsed(); + _stopWatchdog(); case VpnStatus.off: case VpnStatus.error: state = const ConnectionState(phase: VpnPhase.off); _stopElapsed(); + _stopWatchdog(); } } + // ── 连通看门狗 ─────────────────────────────────────────────── + void _startWatchdog() { + _watchdog?.cancel(); + _healthFails = 0; + _watchdog = Timer.periodic(_kWatchdogInterval, (_) => _checkHealth()); + } + + void _stopWatchdog() { + _watchdog?.cancel(); + _watchdog = null; + _healthFails = 0; + _probing = false; + } + + Future _checkHealth() async { + if (_probing || !mounted || state.phase != VpnPhase.on) return; + _probing = true; + final ok = await _prober(); + _probing = false; + if (!mounted || state.phase != VpnPhase.on) return; + if (ok) { + _healthFails = 0; + return; + } + _healthFails++; + if (_healthFails >= _kHealthFailLimit) { + _stopWatchdog(); + await _onNodeUnhealthy(); + } + } + + /// 当前节点连续探测失败:智能选择 → 切到其他最优可用节点重连;手动选定 → 断开并提示。 + Future _onNodeUnhealthy() async { + final t = _ref.read(appTextProvider); + final smart = _ref.read(selectedNodeCodeProvider) == kSmartNodeCode; + final altCode = smart ? _pickAlternativeCode(_ref.read(effectiveNodeProvider).code) : null; + if (smart && altCode != null) { + _ref.read(selectedNodeCodeProvider.notifier).state = altCode; + await _disconnect(); + await _connect(); + // 重连进行中给出「已自动切换」提示(connecting 态显示,连上后随 copyWith 清除)。 + if (mounted && state.phase != VpnPhase.off) { + state = ConnectionState(phase: state.phase, error: t.nodeUnhealthySwitched); + } + return; + } + // 手动选定节点(或智能模式无其他可用节点):断开并提示,尊重用户选择、不自动换。 + await _disconnect(); + if (mounted) state = ConnectionState(phase: VpnPhase.off, error: t.nodeUnhealthyError); + } + + /// 选延迟最优、可用(status up)、非当前节点的 code;无则 null。 + String? _pickAlternativeCode(String excludeCode) { + final nodes = (_ref.read(nodesProvider).valueOrNull ?? const []) + .where((n) => !n.isDown && n.uuid.isNotEmpty && n.code != excludeCode) + .toList(); + if (nodes.isEmpty) return null; + nodes.sort((a, b) { + final pa = a.ping > 0 ? a.ping : 1 << 30; + final pb = b.ping > 0 ? b.ping : 1 << 30; + return pa.compareTo(pb); + }); + return nodes.first.code; + } + void _startElapsed() { _elapsed?.cancel(); _elapsed = Timer.periodic(const Duration(seconds: 1), (_) { @@ -195,6 +305,7 @@ class ConnectionController extends StateNotifier { void dispose() { _statusSub?.cancel(); _stopElapsed(); + _stopWatchdog(); _api?.dispose(); super.dispose(); } diff --git a/client/test/unit/connection_watchdog_test.dart b/client/test/unit/connection_watchdog_test.dart new file mode 100644 index 0000000..79f9fb6 --- /dev/null +++ b/client/test/unit/connection_watchdog_test.dart @@ -0,0 +1,132 @@ +// connection_watchdog_test.dart — 连通看门狗:已连接后探测连续失败时, +// 智能选择自动切到其他节点;手动选定的节点只断开并提示。 +import 'dart:async'; + +import 'package:flutter/widgets.dart'; +import 'package:flutter_riverpod/flutter_riverpod.dart'; +import 'package:flutter_test/flutter_test.dart'; +import 'package:pangolin_vpn/bridge/vpn_bridge.dart'; +import 'package:pangolin_vpn/bridge/vpn_bridge_provider.dart'; +import 'package:pangolin_vpn/l10n/strings_zh.dart'; +import 'package:pangolin_vpn/models/node.dart'; +import 'package:pangolin_vpn/services/connect_api.dart'; +import 'package:pangolin_vpn/state/connection_provider.dart'; +import 'package:pangolin_vpn/state/nodes_provider.dart'; + +// 可控假桥:能手动推 VpnStatus,无内部计时器(避免 pending timer)。 +class _FakeBridge implements VpnBridge { + final _status = StreamController.broadcast(); + final _stats = StreamController.broadcast(); + void emit(VpnStatus s) => _status.add(s); + @override + Stream get statusStream => _status.stream; + @override + Stream get statsStream => _stats.stream; + @override + Future start(String configJson) async {} + @override + Future stop() async {} + @override + Future getStatus() async => VpnStatus.off; + @override + Future selectOutbound(String tag) async {} + @override + Future getActiveOutbound() async => 'auto'; + @override + Future setKillSwitch({required bool on}) async {} + @override + void dispose() { + _status.close(); + _stats.close(); + } +} + +// 重连时 fetchConfig 立即抛错,绕开真实网络(不留超时计时器)。 +class _FakeConnectApi extends ConnectApi { + _FakeConnectApi() : super(baseUrl: 'http://test.local', authToken: ''); + @override + Future fetchConfig({required String nodeId, required String deviceId, bool splitCN = false}) async { + throw const ConnectApiException(statusCode: 0, messageZh: 'x', messageEn: 'x'); + } + @override + void dispose() {} +} + +class _StubNodes extends NodesNotifier { + @override + Future> build() async => const [ + Node(code: 'HK', nameZh: '香港', nameEn: 'Hong Kong', ping: 20, uuid: 'hk-uuid', host: 'hk', port: 443), + Node(code: 'JP', nameZh: '东京', nameEn: 'Tokyo', ping: 35, uuid: 'jp-uuid', host: 'jp', port: 443), + ]; + @override + Future refresh() async {} +} + +void main() { + const t = StringsZh(); + + ProviderContainer makeContainer(_FakeBridge bridge, {required bool healthy}) => + ProviderContainer(overrides: [ + vpnBridgeProvider.overrideWithValue(bridge), + nodesProvider.overrideWith(_StubNodes.new), + connectApiFactoryProvider.overrideWithValue((_) => _FakeConnectApi()), + connectionProvider.overrideWith((ref) => ConnectionController( + ref, ref.watch(vpnBridgeProvider), + prober: () async => healthy, + )), + ]); + + // 驱动到 on,再走 3 个看门狗周期(每个 15s)。 + Future driveOnAndProbe(WidgetTester tester, ProviderContainer c, _FakeBridge bridge) async { + c.read(nodesProvider); // 触发(异步)节点加载 + await tester.pump(); // 等 valueOrNull 就绪(智能切换要读节点列表) + c.read(connectionProvider); // 建 controller(订阅状态流) + bridge.emit(VpnStatus.on); + await tester.pump(); + expect(c.read(connectionProvider).phase, VpnPhase.on); + for (var i = 0; i < 3; i++) { + await tester.pump(const Duration(seconds: 15)); + await tester.pump(); + } + await tester.pump(); + } + + testWidgets('智能模式:探测连续失败 → 自动切到其他最优节点', (tester) async { + final bridge = _FakeBridge(); + final c = makeContainer(bridge, healthy: false); + addTearDown(bridge.dispose); + addTearDown(c.dispose); + await tester.pumpWidget(UncontrolledProviderScope(container: c, child: const SizedBox())); + // 默认即智能选择(selectedNodeCode = AUTO);有效节点取最小延迟 HK,故应切到 JP。 + await driveOnAndProbe(tester, c, bridge); + expect(c.read(selectedNodeCodeProvider), 'JP', reason: '智能模式应自动切到其他最优节点'); + }); + + testWidgets('手动选定节点:探测连续失败 → 断开并提示,不自动换', (tester) async { + final bridge = _FakeBridge(); + final c = makeContainer(bridge, healthy: false); + addTearDown(bridge.dispose); + addTearDown(c.dispose); + await tester.pumpWidget(UncontrolledProviderScope(container: c, child: const SizedBox())); + c.read(selectedNodeCodeProvider.notifier).state = 'HK'; // 手动选定 HK + await driveOnAndProbe(tester, c, bridge); + expect(c.read(selectedNodeCodeProvider), 'HK', reason: '手动模式不应自动换节点'); + final st = c.read(connectionProvider); + expect(st.phase, VpnPhase.off, reason: '应断开'); + expect(st.error, t.nodeUnhealthyError, reason: '应给出节点异常提示'); + }); + + testWidgets('探测正常:不触发任何动作', (tester) async { + final bridge = _FakeBridge(); + final c = makeContainer(bridge, healthy: true); + addTearDown(bridge.dispose); + addTearDown(c.dispose); + await tester.pumpWidget(UncontrolledProviderScope(container: c, child: const SizedBox())); + c.read(selectedNodeCodeProvider.notifier).state = 'HK'; + await driveOnAndProbe(tester, c, bridge); + expect(c.read(connectionProvider).phase, VpnPhase.on, reason: '健康时应保持连接'); + expect(c.read(selectedNodeCodeProvider), 'HK'); + bridge.emit(VpnStatus.off); // 收尾:停看门狗/计时器,避免 pending timer + await tester.pump(); + }); +} diff --git a/server/cmd/server/main.go b/server/cmd/server/main.go index 056b754..daabb10 100644 --- a/server/cmd/server/main.go +++ b/server/cmd/server/main.go @@ -322,7 +322,7 @@ func mountV1(r chi.Router, sqlDB *sql.DB, rdb *redis.Client, nodeSvc *nodes.Serv slog.Warn("PANGOLIN_PUBLIC_URL 未设置:国内分流(split_cn)将被静默跳过,客户端全量走隧道。" + "如需国内直连,设为控制面对外公网基址(如 http://<公网IP>:8080)") } - nodeAPI = httpapi.NewNodeAPI(nodeStore, nodeSvc.Hub(), os.Getenv("NODE_DERIVE_KEY"), publicURL) + nodeAPI = httpapi.NewNodeAPI(nodeStore, nodeSvc.Hub(), nodeSvc.Load(), os.Getenv("NODE_DERIVE_KEY"), publicURL) } // 国内分流(#5)的 rule-set 静态服务:GET /v1/rules/{name}.srs(自托管, diff --git a/server/internal/agentd/agent.go b/server/internal/agentd/agent.go index aedbcec..7794b95 100644 --- a/server/internal/agentd/agent.go +++ b/server/internal/agentd/agent.go @@ -23,6 +23,7 @@ type Agent struct { enrollDial EnrollDialer load LoadSource usage UsageSource + health HealthSource clock func() time.Time lastCommandID atomic.Int64 @@ -39,6 +40,7 @@ func WithDialer(d Dialer) Option { return func(a *Agent) { a.dial = func WithEnrollDialer(d EnrollDialer) Option { return func(a *Agent) { a.enrollDial = d } } func WithLoadSource(l LoadSource) Option { return func(a *Agent) { a.load = l } } func WithUsageSource(u UsageSource) Option { return func(a *Agent) { a.usage = u } } +func WithHealthSource(h HealthSource) Option { return func(a *Agent) { a.health = h } } func WithClock(f func() time.Time) Option { return func(a *Agent) { a.clock = f } } // New builds an Agent. Production callers pass nothing extra and get mTLS dialers; @@ -71,6 +73,9 @@ func New(cfg Config, opts ...Option) *Agent { if a.usage == nil { a.usage = nopUsageSource{} } + if a.health == nil { + a.health = newClashHealthSource() + } return a } diff --git a/server/internal/agentd/heartbeat.go b/server/internal/agentd/heartbeat.go index 2b76e03..5103302 100644 --- a/server/internal/agentd/heartbeat.go +++ b/server/internal/agentd/heartbeat.go @@ -2,6 +2,7 @@ package agentd import ( "context" + "net/http" "time" agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1" @@ -23,6 +24,49 @@ func (d defaultLoadSource) Load() (int32, int64, int64, float64) { return int32(d.sb.OnlinePeers()), 0, 0, 0 } +// HealthSource reports whether the local data plane (sing-box) is healthy, sent +// in each heartbeat so the control plane can mark a node down when its sing-box +// is dead even though the agent's gRPC stream is still up. +type HealthSource interface{ Healthy() bool } + +// clashHealthSource probes sing-box's clash_api /version. A reachable, 200-OK API +// means the sing-box process is alive and its config loaded. Applies a 2-strike +// debounce: a single failed probe is tolerated (covers the sing-box restart +// window); two consecutive failures report unhealthy. +type clashHealthSource struct { + probe func() bool // single probe, true = healthy; injectable for tests + fails int +} + +func newClashHealthSource() *clashHealthSource { + client := &http.Client{Timeout: 2 * time.Second} + return &clashHealthSource{probe: func() bool { + req, err := http.NewRequestWithContext( + context.Background(), http.MethodGet, "http://"+clashAPIAddr+"/version", nil) + if err != nil { + return false + } + if clashAPISecret != "" { + req.Header.Set("Authorization", "Bearer "+clashAPISecret) + } + resp, err := client.Do(req) + if err != nil { + return false + } + defer resp.Body.Close() + return resp.StatusCode == http.StatusOK + }} +} + +func (h *clashHealthSource) Healthy() bool { + if h.probe() { + h.fails = 0 + return true + } + h.fails++ + return h.fails < 2 // tolerate 1 failure; unhealthy from the 2nd consecutive +} + // runHeartbeat sends a heartbeat every cfg.HeartbeatInterval until ctx is // cancelled or an RPC error occurs. A need_full_resync response returns // errResync so the session re-Registers and overwrites local state. @@ -51,6 +95,7 @@ func (a *Agent) sendHeartbeat(ctx context.Context, client agentv1.AgentServiceCl BandwidthDownBps: down, CPUPercent: cpu, TimestampUnix: a.clock().Unix(), + DataPlaneHealthy: a.health.Healthy(), }) if err != nil { return err diff --git a/server/internal/agentd/heartbeat_health_test.go b/server/internal/agentd/heartbeat_health_test.go new file mode 100644 index 0000000..a0c01f6 --- /dev/null +++ b/server/internal/agentd/heartbeat_health_test.go @@ -0,0 +1,23 @@ +package agentd + +import "testing" + +// TestClashHealthSource_Debounce verifies the 2-strike debounce: a single failed +// probe is tolerated (sing-box restart window), two consecutive failures report +// unhealthy, and a success resets the strike counter. +func TestClashHealthSource_Debounce(t *testing.T) { + probeResults := []bool{true, false, false, false, true, false} + want := []bool{true, true, false, false, true, true} + // ↑ok ↑1st ↑2nd ↑3rd ↑ok ↑1st-after-reset + i := 0 + h := &clashHealthSource{probe: func() bool { + r := probeResults[i] + i++ + return r + }} + for n, w := range want { + if got := h.Healthy(); got != w { + t.Errorf("Healthy() call %d = %v, want %v (probe=%v)", n, got, w, probeResults[n]) + } + } +} diff --git a/server/internal/httpapi/nodes.go b/server/internal/httpapi/nodes.go index 1906dec..e96b9b4 100644 --- a/server/internal/httpapi/nodes.go +++ b/server/internal/httpapi/nodes.go @@ -1,6 +1,7 @@ package httpapi import ( + "context" "encoding/json" "log/slog" "net/http" @@ -23,19 +24,40 @@ const ( freeMinuteTTL = time.Minute ) +// nodeLoadReader reads a node's last-reported runtime load (for the data-plane +// health flag). *nodes.LoadCache satisfies it; nil = treat all nodes healthy. +type nodeLoadReader interface { + Get(ctx context.Context, nodeUUID string) (*nodes.NodeLoad, bool, error) +} + // NodeAPI serves the /v1/nodes endpoints. type NodeAPI struct { store nodes.NodeStore hub *nodes.Hub + load nodeLoadReader deriveKey string // rulesBaseURL 是控制面对外公网基址(PANGOLIN_PUBLIC_URL),供国内分流的 // rule_set .srs 下载用;空则分流不生效。 rulesBaseURL string } -// NewNodeAPI creates a NodeAPI. -func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, deriveKey, rulesBaseURL string) *NodeAPI { - return &NodeAPI{store: store, hub: hub, deriveKey: deriveKey, rulesBaseURL: rulesBaseURL} +// NewNodeAPI creates a NodeAPI. load may be nil (then all nodes are treated as +// data-plane healthy — agent gRPC liveness still gates status). +func NewNodeAPI(store nodes.NodeStore, hub *nodes.Hub, load nodeLoadReader, deriveKey, rulesBaseURL string) *NodeAPI { + return &NodeAPI{store: store, hub: hub, load: load, deriveKey: deriveKey, rulesBaseURL: rulesBaseURL} +} + +// dataPlaneHealthy reports the node's last sing-box health (default true when +// unknown: just registered / load expired / read error — agent liveness gates). +func (a *NodeAPI) dataPlaneHealthy(ctx context.Context, nodeUUID string) bool { + if a.load == nil { + return true + } + l, ok, err := a.load.Get(ctx, nodeUUID) + if err != nil || !ok || l == nil { + return true + } + return l.DataPlaneHealthy } // ─── GET /v1/nodes ─────────────────────────────────────────────────────────── @@ -53,12 +75,13 @@ type nodeResponse struct { } // effectiveNodeStatus downgrades a DB-'up' node to "down" when its agent is -// offline. The DB status column reflects provisioning/scheduler state, NOT agent -// liveness — without this, a node whose agent has been offline for days still -// shows healthy (the gap that hid a 6-day agent outage and let clients "connect" -// to a node that couldn't serve them). -func effectiveNodeStatus(dbStatus string, agentOnline bool) string { - if dbStatus == "up" && !agentOnline { +// offline OR its data plane (sing-box) is unhealthy. The DB status column reflects +// provisioning/scheduler state, NOT live health — without this, a node whose agent +// has been offline for days, or whose sing-box has crashed while the agent stays +// connected, still shows healthy and lets clients "connect" to a node that can't +// serve them. +func effectiveNodeStatus(dbStatus string, agentOnline, dataPlaneHealthy bool) string { + if dbStatus == "up" && (!agentOnline || !dataPlaneHealthy) { return "down" } return dbStatus @@ -85,7 +108,9 @@ func (a *NodeAPI) ListNodes(w http.ResponseWriter, r *http.Request) { NameZH: n.NameZH, NameEN: n.NameEN, Tier: n.Tier, - Status: effectiveNodeStatus(n.Status, a.hub == nil || a.hub.IsOnline(n.UUID)), + Status: effectiveNodeStatus(n.Status, + a.hub == nil || a.hub.IsOnline(n.UUID), + a.dataPlaneHealthy(r.Context(), n.UUID)), Host: host, Port: port, }) @@ -188,7 +213,9 @@ func (a *NodeAPI) ConnectNode(w http.ResponseWriter, r *http.Request) { apierr.WriteJSON(w, http.StatusInternalServerError, apierr.ErrInternal) return } - if node == nil || node.Status != "up" { + if node == nil || effectiveNodeStatus(node.Status, + a.hub == nil || a.hub.IsOnline(nodeUUID), + a.dataPlaneHealthy(r.Context(), nodeUUID)) != "up" { apierr.WriteJSON(w, http.StatusNotFound, apierr.ErrNotFound) return } diff --git a/server/internal/httpapi/nodes_status_test.go b/server/internal/httpapi/nodes_status_test.go index 9fc05c3..29326dc 100644 --- a/server/internal/httpapi/nodes_status_test.go +++ b/server/internal/httpapi/nodes_status_test.go @@ -1,25 +1,69 @@ package httpapi -import "testing" +import ( + "context" + "errors" + "testing" -// TestEffectiveNodeStatus guards the "don't show a node healthy when its agent is -// offline" fix: a DB-'up' node with an offline agent must report "down". -func TestEffectiveNodeStatus(t *testing.T) { + "github.com/wangjia/pangolin/server/internal/nodes" +) + +type fakeLoadReader struct { + load *nodes.NodeLoad + ok bool + err error +} + +func (f fakeLoadReader) Get(context.Context, string) (*nodes.NodeLoad, bool, error) { + return f.load, f.ok, f.err +} + +// TestDataPlaneHealthy: unknown (nil reader / missing / error) defaults to healthy +// (agent liveness gates); a present load returns its reported flag. +func TestDataPlaneHealthy(t *testing.T) { + ctx := context.Background() cases := []struct { - name string - db string - online bool - want string + name string + api *NodeAPI + want bool }{ - {"up + agent online", "up", true, "up"}, - {"up + agent offline → down", "up", false, "down"}, // the core fix - {"draining untouched when offline", "draining", false, "draining"}, - {"down stays down", "down", true, "down"}, + {"nil reader → healthy", &NodeAPI{}, true}, + {"missing load → healthy", &NodeAPI{load: fakeLoadReader{ok: false}}, true}, + {"read error → healthy", &NodeAPI{load: fakeLoadReader{err: errors.New("boom")}}, true}, + {"present healthy", &NodeAPI{load: fakeLoadReader{load: &nodes.NodeLoad{DataPlaneHealthy: true}, ok: true}}, true}, + {"present unhealthy → false", &NodeAPI{load: fakeLoadReader{load: &nodes.NodeLoad{DataPlaneHealthy: false}, ok: true}}, false}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - if got := effectiveNodeStatus(c.db, c.online); got != c.want { - t.Errorf("effectiveNodeStatus(%q, %v) = %q, want %q", c.db, c.online, got, c.want) + if got := c.api.dataPlaneHealthy(ctx, "n1"); got != c.want { + t.Errorf("dataPlaneHealthy = %v, want %v", got, c.want) + } + }) + } +} + +// TestEffectiveNodeStatus guards the "don't show a node healthy when it can't +// serve" fix: a DB-'up' node reports "down" when its agent is offline OR its data +// plane (sing-box) is unhealthy. +func TestEffectiveNodeStatus(t *testing.T) { + cases := []struct { + name string + db string + online bool + healthy bool + want string + }{ + {"up + agent online + healthy", "up", true, true, "up"}, + {"up + agent offline → down", "up", false, true, "down"}, // agent-liveness fix + {"up + agent online + dp unhealthy → down", "up", true, false, "down"}, // data-plane fix + {"up + offline + unhealthy → down", "up", false, false, "down"}, + {"draining untouched", "draining", false, false, "draining"}, + {"down stays down", "down", true, true, "down"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := effectiveNodeStatus(c.db, c.online, c.healthy); got != c.want { + t.Errorf("effectiveNodeStatus(%q, %v, %v) = %q, want %q", c.db, c.online, c.healthy, got, c.want) } }) } diff --git a/server/internal/nodes/handler_grpc.go b/server/internal/nodes/handler_grpc.go index c2e3414..0eb80fe 100644 --- a/server/internal/nodes/handler_grpc.go +++ b/server/internal/nodes/handler_grpc.go @@ -183,6 +183,7 @@ func (h *Handler) Heartbeat(ctx context.Context, req *agentv1.HeartbeatRequest) BandwidthUpBps: req.BandwidthUpBps, BandwidthDownBps: req.BandwidthDownBps, CPUPercent: req.CPUPercent, + DataPlaneHealthy: req.DataPlaneHealthy, }); err != nil { slog.Warn("nodes.Handler.Heartbeat: load write failed", "node_uuid", nodeUUID, "err", err) diff --git a/server/internal/nodes/load.go b/server/internal/nodes/load.go index c2af1c7..9699b57 100644 --- a/server/internal/nodes/load.go +++ b/server/internal/nodes/load.go @@ -24,6 +24,10 @@ type NodeLoad struct { BandwidthUpBps int64 BandwidthDownBps int64 CPUPercent float64 + // DataPlaneHealthy is the agent's last sing-box health probe result. The + // control plane downgrades a node to "down" when this is false (even if the + // agent's gRPC stream is still up). Absent in older hashes → read as true. + DataPlaneHealthy bool } // LoadCache reads and writes node:load Redis hashes. @@ -41,12 +45,17 @@ func NewLoadCache(rdb redis.Cmdable) *LoadCache { // The hash fields match the names in doc/03 §4: online_count, bw_up, bw_down, cpu. func (c *LoadCache) Set(ctx context.Context, nodeUUID string, load NodeLoad) error { key := nodeLoadKeyPrefix + nodeUUID + dpHealthy := 0 + if load.DataPlaneHealthy { + dpHealthy = 1 + } pipe := c.rdb.Pipeline() pipe.HSet(ctx, key, "online_count", load.OnlineCount, "bw_up", load.BandwidthUpBps, "bw_down", load.BandwidthDownBps, "cpu", load.CPUPercent, + "dp_healthy", dpHealthy, ) pipe.Expire(ctx, key, nodeLoadTTL) if _, err := pipe.Exec(ctx); err != nil { @@ -66,7 +75,12 @@ func (c *LoadCache) Get(ctx context.Context, nodeUUID string) (*NodeLoad, bool, if len(vals) == 0 { return nil, false, nil } - var l NodeLoad + // Default healthy: hashes written before dp_healthy existed (or by an older + // control plane during rollout) must not be read as "unhealthy" → false-down. + l := NodeLoad{DataPlaneHealthy: true} + if v, ok := vals["dp_healthy"]; ok { + l.DataPlaneHealthy = v == "1" + } if v, ok := vals["online_count"]; ok { _, _ = fmt.Sscan(v, &l.OnlineCount) } diff --git a/server/internal/pb/agentv1/types.go b/server/internal/pb/agentv1/types.go index 90dfcbd..5581e8e 100644 --- a/server/internal/pb/agentv1/types.go +++ b/server/internal/pb/agentv1/types.go @@ -108,6 +108,9 @@ type HeartbeatRequest struct { BandwidthDownBps int64 `json:"bandwidth_down_bps,omitempty"` CPUPercent float64 `json:"cpu_percent,omitempty"` TimestampUnix int64 `json:"timestamp_unix,omitempty"` + // DataPlaneHealthy: agent 本地探测 sing-box(clash_api) 是否健康。false → 控制面把 + // 该节点判为 down(列表置灰 + connect 拦截),即便 agent 自身在线。 + DataPlaneHealthy bool `json:"data_plane_healthy,omitempty"` } type HeartbeatResponse struct { diff --git a/server/proto/agent/v1/agent.proto b/server/proto/agent/v1/agent.proto index d8b6bd8..bfe177c 100644 --- a/server/proto/agent/v1/agent.proto +++ b/server/proto/agent/v1/agent.proto @@ -127,6 +127,9 @@ message HeartbeatRequest { int64 bandwidth_down_bps = 5; double cpu_percent = 6; int64 timestamp_unix = 7; + // data_plane_healthy: agent 本地探测 sing-box(clash_api)是否健康。false → 控制面把 + // 该节点判为 down(列表置灰 + connect 拦截),即便 agent 自身在线。 + bool data_plane_healthy = 8; } message HeartbeatResponse {