Files
pangolin/client/lib/state/connection_provider.dart
T
2026-06-30 17:47:16 +08:00

396 lines
18 KiB
Dart

// connection_provider.dart — 连接状态机(严格三态,状态由内核事件驱动)
//
// 设计约定:
// - UI 调用 toggle(),控制器内部读取有效节点 + 认证令牌,调用 ConnectApi
// 并启动 VpnBridge 子进程。
// - 严禁乐观翻转:VpnPhase.on 必须由 bridge.statusStream 确认后才置。
// - 状态来源:bridge.statusStream(来自内核回调),非 Timer 模拟。
import 'dart:async';
import 'package:flutter_riverpod/flutter_riverpod.dart';
import '../bridge/log.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';
import 'app_providers.dart';
import 'auth_provider.dart';
import 'nodes_provider.dart';
import 'settings_provider.dart';
// 设备 ID 由 deviceIdentityProvider 提供(secure storage 持久化的稳定 UUID)。
// API base URL 统一用 api_config.dart 的 kApiBaseUrl(单一来源,勿再重复声明)。
// ── 连接阶段枚举 ──────────────────────────────────────────────────
/// 连接阶段。严格三态,与设计稿一致。
enum VpnPhase { off, connecting, on }
// ── 连接状态快照 ──────────────────────────────────────────────────
class ConnectionState {
const ConnectionState({required this.phase, this.elapsed = Duration.zero, this.error});
final VpnPhase phase;
final Duration elapsed;
/// 连接失败/中断原因(已本地化);null = 无错误。供 UI 提示,不再静默吞掉。
final String? error;
ConnectionState copyWith({VpnPhase? phase, Duration? elapsed}) =>
ConnectionState(phase: phase ?? this.phase, elapsed: elapsed ?? this.elapsed);
@override
bool operator ==(Object other) =>
other is ConnectionState &&
other.phase == phase &&
other.elapsed == elapsed &&
other.error == error;
@override
int get hashCode => Object.hash(phase, elapsed, error);
}
// ── 连通看门狗 ─────────────────────────────────────────────────────
//
// 已连接(本地 TUN 已起)≠ 远端节点数据面可用。三条路径发现节点异常,全平台一致:
// A. urltest 陈旧(跨端统一主路径):内核 urltest 经 proxy 出站(REALITY→节点)探测,
// 节点死则探测失败、urltest 停更。看门狗发现「上次 urltest 成功超过 _kUrltestStale」
// 即判上游死 → _onNodeUnhealthy。urltest 四端都有(stats),不依赖各原生端的状态上报。
// A'. 意外内核掉线:原生端真把「上游断」上报成 off/error 时(桌面子进程确认有),走
// _onKernelStatus 的非用户主动 off → 节点异常提示。属 A 的快路径,有则更快、无则靠 A 兜底。
// B. 服务端权威 down:看门狗连接态周期刷 /v1/nodes,所连节点被判 down(dp_healthy=0 /
// agent 掉线 / 运维下线)即处理。覆盖「本地 TUN 还没断、但节点已被判不可用」。
// 注:不做客户端 TCP 探数据口——全局 TUN 下 App 的 socket 被本地 TUN 当场接住(假 1ms),
// 测不到真节点。判活针对「实际所连节点」(_connectedNode),非会随 ping 漂移的 effectiveNode。
const _kWatchdogInterval = Duration(seconds: 15);
// urltest 超过此时长未成功(连接态)即判上游(节点数据面)死。须 > 原生 urltest 探测间隔。
const _kUrltestStale = Duration(seconds: 45);
// ── 状态机 ───────────────────────────────────────────────────────
class ConnectionController extends StateNotifier<ConnectionState> {
ConnectionController(this._ref, this._bridge)
: super(const ConnectionState(phase: VpnPhase.off)) {
// 订阅桥状态流:状态由内核事件驱动,严禁 UI 乐观翻转。
_statusSub = _bridge.statusStream.listen(_onKernelStatus);
// 订阅内核 stats:连接态把 urltest 实测回写连接节点的 ping,让连接页与节点列表同源一致。
_statsSub = _bridge.statsStream.listen(_onStats);
logLine('Stats', 'ConnectionController constructed, statsStream subscribed');
// 登出(主动退出 / 被强制下线 / 会话过期)→ 一律断开隧道:别让已登出/被踢的设备继续用
// 数据面。一处监听 auth 覆盖所有登出路径。token 续期(仍登录)不触发。
_authSub = _ref.listen<AuthState>(authProvider, (prev, next) {
if ((prev?.isLoggedIn ?? false) && !next.isLoggedIn) {
_userDisconnect = true; // 视为「非节点异常」的主动断开,不弹「节点异常」
unawaited(_disconnect());
}
});
}
final Ref _ref;
final VpnBridge _bridge;
StreamSubscription<VpnStatus>? _statusSub;
StreamSubscription<VpnStatsEvent>? _statsSub;
ProviderSubscription<AuthState>? _authSub;
Timer? _elapsed;
Timer? _watchdog;
bool _probing = false;
// 最近一次 urltest 成功(经 proxy 真实可达)的时刻;路径 A 用其陈旧度判上游死。
DateTime? _lastUrltestOk;
// 本次 off 的提示语(节点异常/自动切换);kernel off 与显式断开都读它,避免被互相覆盖。
// 仅 _connect 开始时清空。用户主动断开则保持 null(无提示)。
String? _offNotice;
// 标记「用户主动断开」:其引发的 kernel off 不当作节点异常。
bool _userDisconnect = false;
ConnectApi? _api;
// 实际所连节点(连接时锁定):看门狗探测/判活针对它,而非会随 ping 漂移的 effectiveNode。
Node? _connectedNode;
// ── 公有 API ───────────────────────────────────────────────────
/// 用户点击连接键:按当前状态决定动作,握手中忽略(禁乐观)。
void toggle() {
switch (state.phase) {
case VpnPhase.off:
_connect();
case VpnPhase.on:
_userDisconnect = true; // 用户主动断开:其 kernel off 不当作节点异常
_disconnect();
case VpnPhase.connecting:
break; // 握手进行中,不响应
}
}
/// 节点切换时触发:已连接则重连。
void onNodeChanged() {
if (state.phase == VpnPhase.on || state.phase == VpnPhase.connecting) {
_disconnect().then((_) => _connect());
}
}
// ── 内部 ─────────────────────────────────────────────────────
Future<void> _connect() async {
_offNotice = null; // 新连接:清掉上次的中断提示
_userDisconnect = false;
_lastUrltestOk = null; // 重置 urltest 判活基准(连上后由 _onStats 首次成功置位)
state = const ConnectionState(phase: VpnPhase.connecting);
final node = _ref.read(effectiveNodeProvider);
final zh = _ref.read(localeProvider) == AppLang.zh;
logLine('Connect', '_connect node=${node.code} uuid=${node.uuid.isEmpty ? "EMPTY" : "ok"} '
'selected=${_ref.read(selectedNodeCodeProvider)} nodes=${(_ref.read(nodesProvider).valueOrNull ?? const []).length}');
// 节点未就绪(列表加载中/为空):不连接,提示用户。
if (node.uuid.isEmpty) {
if (mounted) {
state = ConnectionState(
phase: VpnPhase.off,
error: zh ? '节点尚未就绪,请稍候重试' : 'Nodes not ready, please retry',
);
}
return;
}
_connectedNode = node; // 锁定本次实际所连节点,供看门狗探测/判活
try {
final configJson = await _fetchConfigWithRefresh(node.uuid);
// bridge.start() 不阻塞至连接建立;on 状态由 statusStream 回调驱动。
await _bridge.start(configJson);
} on ConnectApiException catch (e) {
// 把后端/网络错误冒泡到 UI(原静默回 off,用户不知所以)。
if (mounted) state = ConnectionState(phase: VpnPhase.off, error: zh ? e.messageZh : e.messageEn);
} catch (e) {
if (mounted) {
state = ConnectionState(
phase: VpnPhase.off,
error: zh ? '连接失败,请重试' : 'Connection failed, please retry',
);
}
}
}
/// 取配置;access token 过期(401)时用 refresh token 续期后**重试一次**。
/// 续期失败(refresh 也过期 / 被拒)由 authProvider.refresh() 触发登出 → UI 回登录页。
Future<String> _fetchConfigWithRefresh(String nodeUuid) async {
final deviceId = await _ref.read(deviceIdentityProvider).deviceId();
Future<String> doFetch() {
final token = _ref.read(authProvider).accessToken ?? '';
_api?.dispose();
_api = _ref.read(connectApiFactoryProvider)(token);
return _api!.fetchConfig(
nodeId: nodeUuid,
deviceId: deviceId,
// smartRoute 偏好 → 国内分流(#5):国内 IP/域名直连,不走隧道。
splitCN: _ref.read(settingsProvider).smartRoute,
);
}
try {
return await doFetch();
} on ConnectApiException catch (e) {
if (e.statusCode == 401) {
// token 过期 → 续期后重试一次;续期成功则用新 token 再取。
final ok = await _ref.read(authProvider.notifier).refresh();
if (ok) return await doFetch();
}
rethrow;
}
}
Future<void> _disconnect() async {
_stopElapsed();
_stopWatchdog();
try {
await _bridge.stop();
} catch (_) {}
// 携带 _offNotice(节点异常/null);与 kernel off 读同一字段,不互相覆盖。
if (mounted) state = ConnectionState(phase: VpnPhase.off, error: _offNotice);
}
void _onKernelStatus(VpnStatus s) {
if (!mounted) return;
switch (s) {
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:
// 路径 A:仅「真正连上过(on)再非用户主动掉线」才判节点异常(数据面死/REALITY 断)。
// 必须排除 connecting 握手期的瞬态 off——macOS NE 连接序列是 off→connecting→off→
// connecting→on,connecting 期的 off 是握手抖动、不是异常(否则一连接就误报「节点异常」)。
final wasConnected = state.phase == VpnPhase.on;
if (!_userDisconnect && wasConnected && _offNotice == null) {
_offNotice = _ref.read(appTextProvider).nodeUnhealthyError;
logLine('Watchdog', 'unexpected kernel ${s.name} after connected → node interrupted');
unawaited(_ref.read(nodesProvider.notifier).refresh());
}
state = ConnectionState(phase: VpnPhase.off, error: _offNotice);
_userDisconnect = false;
_stopElapsed();
_stopWatchdog();
}
}
// 连接态:取内核 urltest 最小正延迟,回写连接节点 ping。全局 TUN 下 App 直接探测节点不可行,
// 内核(经 REALITY 真实直连数据面)是唯一能实测的;回写到 node.ping → 连接页/节点列表同源。
void _onStats(VpnStatsEvent e) {
// 无条件打点:确认 _onStats 是否被调用、当前 phase、urltest 是否到达 Dart 侧。
logLine('Stats',
'onStats phase=${state.phase.name} '
'urltest=[${e.urltestResults.map((r) => "${r.tag}:${r.delayMs}").join(",")}]');
if (state.phase != VpnPhase.on) return;
// 自动连接/重连到已在跑的隧道时不会走 _connect → _connectedNode 为 null;回退到
// effectiveNode(当前/智能所选),否则 urltest 实测无处回写、连接页延迟一直显示 —。
final Node node = _connectedNode ?? _ref.read(effectiveNodeProvider);
final ds = e.urltestResults.where((r) => r.delayMs > 0).map((r) => r.delayMs).toList();
logLine('Stats',
'conn=${_connectedNode == null ? "null" : _connectedNode!.code} '
'eff=${node.code}/${node.uuid.isEmpty ? "EMPTY" : "ok"} ds=$ds');
if (node.uuid.isEmpty) return;
if (ds.isEmpty) return;
// urltest 成功:经 proxy 出站(REALITY 到节点)真实可达 → 记录时刻(供路径 A 判活)+ 回写延迟。
_lastUrltestOk = DateTime.now();
final best = ds.reduce((a, b) => a < b ? a : b);
logLine('Stats', 'setLivePing node=${node.code} = ${best}ms');
_ref.read(nodesProvider.notifier).setLivePing(node.uuid, best);
}
// ── 连通看门狗(路径 B:服务端权威 down 轮询)───────────────────────
void _startWatchdog() {
_watchdog?.cancel();
_watchdog = Timer.periodic(_kWatchdogInterval, (_) => _checkHealth());
unawaited(_checkHealth()); // 立即查一次,早发现「TUN 未断但节点已被判 down」
}
void _stopWatchdog() {
_watchdog?.cancel();
_watchdog = null;
_probing = false;
}
Future<void> _checkHealth() async {
if (_probing || !mounted || state.phase != VpnPhase.on) return;
// 实际所连节点(测试态直接 emit on 时未走 _connect → 回退 effectiveNode)。
final Node node = _connectedNode ?? _ref.read(effectiveNodeProvider);
if (node.uuid.isEmpty) return;
_probing = true;
var serverDown = false;
try {
await _ref.read(nodesProvider.notifier).refresh();
final cur = (_ref.read(nodesProvider).valueOrNull ?? const <Node>[])
.where((n) => n.uuid == node.uuid);
if (cur.isNotEmpty) serverDown = cur.first.isDown;
} catch (_) {
// 拉取失败保留旧列表,不误判;等下一轮。
}
_probing = false;
if (!mounted || state.phase != VpnPhase.on) return;
if (serverDown) {
logLine('Watchdog', 'server marked node ${node.code} down → unhealthy');
_stopWatchdog();
await _onNodeUnhealthy();
return;
}
// 路径 A(跨端统一):urltest 曾成功过、但已超过 _kUrltestStale 未再成功 → 上游(节点
// 数据面)死。urltest 经 proxy 出站探测,四端都有;不依赖原生端把「上游断」报成 off/error。
final lastOk = _lastUrltestOk;
if (lastOk != null && DateTime.now().difference(lastOk) > _kUrltestStale) {
logLine('Watchdog', 'urltest stale ${DateTime.now().difference(lastOk).inSeconds}s → upstream dead → unhealthy');
_stopWatchdog();
await _onNodeUnhealthy();
}
}
/// 所连节点被服务端判 down:智能 → 切到其他最优可用节点重连;手动 → 断开并提示。
Future<void> _onNodeUnhealthy() async {
final t = _ref.read(appTextProvider);
final Node node = _connectedNode ?? _ref.read(effectiveNodeProvider);
final smart = _ref.read(selectedNodeCodeProvider) == kSmartNodeCode;
final altCode = smart ? _pickAlternativeCode(node.code) : null;
if (smart && altCode != null) {
_ref.read(selectedNodeCodeProvider.notifier).select(altCode);
await _disconnect();
await _connect();
// 重连进行中给出「已自动切换」提示(connecting 态显示,连上后随 copyWith 清除)。
if (mounted && state.phase != VpnPhase.off) {
state = ConnectionState(phase: state.phase, error: t.nodeUnhealthySwitched);
}
return;
}
// 手动选定节点(或智能模式无其他可用节点):断开并提示,尊重用户选择、不自动换。
// 提示走 _offNotice,由 _disconnect/_onKernelStatus 应用,避免被 kernel off 覆盖。
_offNotice = t.nodeUnhealthyError;
await _disconnect();
}
/// 选延迟最优、可用(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), (_) {
if (mounted && state.phase == VpnPhase.on) {
state = state.copyWith(elapsed: state.elapsed + const Duration(seconds: 1));
}
});
}
void _stopElapsed() {
_elapsed?.cancel();
_elapsed = null;
}
@override
void dispose() {
_statusSub?.cancel();
_statsSub?.cancel();
_authSub?.close();
_stopElapsed();
_stopWatchdog();
_api?.dispose();
super.dispose();
}
}
// ── Provider ──────────────────────────────────────────────────────
final connectionProvider =
StateNotifierProvider<ConnectionController, ConnectionState>(
(ref) => ConnectionController(ref, ref.watch(vpnBridgeProvider)),
);
/// 内核实时统计流(上/下行瞬时速率、字节数)。连接页速度行的真实数据源。
final vpnStatsProvider = StreamProvider<VpnStatsEvent>(
(ref) => ref.watch(vpnBridgeProvider).statsStream,
);
/// 按 authToken 造 ConnectApi 的工厂(支柱 1:接缝即接口)。默认走真控制面;
/// 测试 override 注入 MockClient,即可驱动「连接成功」路径而不打真网络。
typedef ConnectApiFactory = ConnectApi Function(String authToken);
final connectApiFactoryProvider = Provider<ConnectApiFactory>(
(_) => (authToken) => ConnectApi(baseUrl: kApiBaseUrl, authToken: authToken),
);