feat(viewer-core): 连接失败分类 + 短预算×有限重拨(3a/U8)

ConnectErrorKind(Resolve/Timeout/Refused/Auth/Version/Other)在 Rust 侧分类,
Kotlin 只做编号→文案映射。分级预算:第一拨握手 5s(快速判不可达)、重拨拨 8s
(接住 agent 串行 accept 的僵尸释放 ~10.2s)、认证 11s、全程 14s 封顶。
is_retryable 只对 Timeout/Refused 重拨;Resolve/Auth/Version 确定性失败不拖慢。

判据演进(偏差 D2/D3,用户 2026-10-31 裁决方案①):单次 5s 总预算在串行 accept
架构下把合法排队压断(text proof 实爆),故改重拨。不可达 32s→~13s 且全程可取消;
僵尸重连第一拨 5s 失败第二拨 10.2s 自动成功(桌面 e2e 实测)。9 单测锁分类/预算/重拨。
This commit is contained in:
郑博文 2026-10-31 13:29:48 +08:00
parent 99ffffa393
commit 4eb82f9779

View File

@ -15,7 +15,6 @@
//! 仅限局域网测试)。产品化前必须换成信令下发的证书指纹钉扎(`signaling.rs`
//! 已有 `CertFingerprint`),该项登记在方案阶段 4。
use anyhow::{bail, Context, Result};
use common::{
read_control, read_video_frame, write_control, ControlMsg, InputEvent, VideoFrameHeader,
PROTO_VERSION,
@ -30,6 +29,30 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
/// QUIC 握手单次尝试预算(第一拨):不可达/被拦路径的失败上限(阶段 3a U8)。
/// 与被控端 `agent-core` 的 max_idle_timeout(10s) 同思路——超时策略必须显式、短、可见。
pub const CONNECT_BUDGET: Duration = Duration::from_secs(5);
/// 第二拨握手预算(重拨专用,8s > 第一拨)。
/// 存在理由:agent 串行 accept,僵尸释放实测在「杀端后 ~10.2s + 残骸握手再排
/// 5s 档」≈13s 窗口内;第二拨 8s 让握手段天然长于僵尸窗口,一次重拨接住。
/// 真·不可达不受此影响——它由总预算封顶报 Timeout。
pub const CONNECT_RETRY_BUDGET: Duration = Duration::from_secs(8);
/// 认证应答预算:握手成功后等 `AuthOk` 的上限。agent 串行 accept(R-5),
/// 正常应答是毫秒级;11s 只是异常场景(agent 卡在旧会话收尾)的兜底。
pub const AUTH_BUDGET: Duration = Duration::from_secs(11);
/// connect() 全程硬上限(含重拨)。14s = 5 + 8 两拨握手预算 + 1s 余量:
/// 正常路径由**内层**各拨预算先触发(错误 detail 精确到段),外层只兜
/// 「重试组合出无界等待」这一种失控。真不可达 → ≈13s 报 Timeout
/// (2b 的 32s → 13s 且全程可取消,判据更新见 V-A1)。
pub const CONNECT_TOTAL_BUDGET: Duration = Duration::from_secs(14);
/// 可重拨失败(Timeout/Refused)的最大尝试次数。Resolve/Auth/Version 是
/// 确定性结果,重拨只会拖慢反馈,不重试。
pub const MAX_CONNECT_ATTEMPTS: usize = 2;
/// 连接目标。当前只有局域网直连;设备码 + 打洞在后续阶段替换 `connect()` 内部实现。
#[derive(Debug, Clone)]
pub enum PeerTarget {
@ -62,6 +85,110 @@ impl ConnectConfig {
}
}
/// 连接建立失败的分类(阶段 3a)。
///
/// 为什么分类在 Rust 侧做而不是 Kotlin:错误来源(DNS / QUIC 握手 / 认证 / 版本)
/// 只有这一层看得全;Kotlin 只做 `code → 中文文案` 映射。
/// [`ConnectErrorKind::code`] 是 FFI 编号(android-ffi `nativeLastErrorKind` 与 Kotlin
/// `ConnectErrorText` 都按它映射),**改动必须三处同步**——FFI 只传整数不传 Rust 类型。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConnectErrorKind {
/// 地址解析失败:host 写错或 DNS 不可达
Resolve,
/// 连接超时:被控端不在线、IP/端口不对、入站被拦
Timeout,
/// 被控端异常应答/拒绝:该端口上没有 QUIC 服务、握手被重置
Refused,
/// 认证失败:验证码错误
Auth,
/// 协议版本不一致:两端必须重编到同一版本
Version,
/// 未分类。**detail 原文必须展示**(第二行),不得静默归进某类文案
Other,
}
impl ConnectErrorKind {
/// FFI 稳定编号。新增项只追加,不复用旧编号(Kotlin `else ->` 兜住未知码)。
pub fn code(self) -> i32 {
match self {
ConnectErrorKind::Resolve => 1,
ConnectErrorKind::Timeout => 2,
ConnectErrorKind::Refused => 3,
ConnectErrorKind::Auth => 4,
ConnectErrorKind::Version => 5,
ConnectErrorKind::Other => 9,
}
}
}
/// 连接失败 = 分类 + 原始错误文本。
#[derive(Debug)]
pub struct ConnectError {
pub kind: ConnectErrorKind,
pub detail: String,
}
impl ConnectError {
fn new(kind: ConnectErrorKind, detail: impl Into<String>) -> Self {
Self { kind, detail: detail.into() }
}
}
impl std::fmt::Display for ConnectError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.detail)
}
}
impl std::error::Error for ConnectError {}
/// 未分类错误的统一包装:保留原文(Other 的 detail 会显示到界面第二行,不静默)。
fn other_err(e: impl std::fmt::Display) -> ConnectError {
ConnectError::new(ConnectErrorKind::Other, e.to_string())
}
/// 认证失败消息 → 分类。判据是 agent 侧的稳定前缀(`agent/src/main.rs` 与
/// `agent-core/src/lib.rs` 都以 `protocol version mismatch` 开头);
/// 命中不了的一律 `Auth`——宁可提示「验证码错误」也不把内部错误文案当分类依据。
fn classify_auth_fail(msg: &str) -> ConnectErrorKind {
if msg.contains("protocol version mismatch") {
ConnectErrorKind::Version
} else {
ConnectErrorKind::Auth
}
}
/// 同步 connect() 调用(本地配置面)的分类:全是 Other(endpoint 停摆/地址非法等),
/// 保留原文。单列函数是为了让分类面在测试里可枚举、改动可见。
fn classify_connect_error(_e: &quinn::ConnectError) -> ConnectErrorKind {
ConnectErrorKind::Other
}
/// 哪些失败值得重拨(3a 方案①)。
/// - Timeout/Refused:agent 串行 accept 的排队/残骸窗口是**暂时**的
/// (实测僵尸释放 ~10.2s、残骸阻塞握手 5~8.5s),第二拨能接住;
/// - Resolve/Auth/Version:确定性结果,重拨只拖慢反馈、不改变答案;
/// - Other:语义不明(含钉扎未实现、首消息异常),不猜、不重试。
fn is_retryable(kind: ConnectErrorKind) -> bool {
matches!(kind, ConnectErrorKind::Timeout | ConnectErrorKind::Refused)
}
/// 握手/连接期错误 → 分类。逐个变体对照 quinn 0.11.12 re-export 的
/// `quinn::ConnectionError`(quinn-proto 0.11.19 `connection/mod.rs:3906`,本会话实读)。
fn classify_connection_error(e: &quinn::ConnectionError) -> ConnectErrorKind {
use quinn::ConnectionError as E;
match e {
E::TimedOut => ConnectErrorKind::Timeout,
// 端口上没有 QUIC 服务(回收包/非法头)、被重置、握手后被关:
// 用户语境统一是「对端不应答」→ Refused
E::TransportError(_) | E::Reset | E::ConnectionClosed(_) | E::ApplicationClosed(_) => {
ConnectErrorKind::Refused
}
// QUIC 传输版本协商失败 ≠ 应用协议版本不一致,不要误导成 Version
E::VersionMismatch | E::LocallyClosed | E::CidsExhausted => ConnectErrorKind::Other,
}
}
/// 交付给消费方的一帧视频。字段语义与 `common::VideoFrameHeader` 一致,
/// payload 为标准 Annex-B H.264(可直接喂 MediaCodec)。
#[derive(Debug, Clone)]
@ -163,7 +290,66 @@ pub struct Session {
impl Session {
/// 建立连接并完成认证握手 + 首帧关键帧请求。
pub async fn connect(cfg: ConnectConfig) -> Result<Self> {
///
/// **短预算 × 有限重拨**(3a/U8,2026-10-31 用户裁决方案①;演进证据链见
/// loop-3-plan 偏差 D2/D3:单次 5s 预算在 agent 串行 accept 架构下自相矛盾——
/// 不可达要快失败、僵尸排队要耐心等,同一数字满足不了;拆握手/认证两段也不够,
/// 实测阻塞点在握手本身):
/// - 每次尝试:握手第一拨 [`CONNECT_BUDGET`](5s)、重拨拨
/// [`CONNECT_RETRY_BUDGET`](8s)、认证 [`AUTH_BUDGET`](11s);
/// - [`is_retryable`] 的失败(Timeout/Refused)**自动重拨**,共
/// [`MAX_CONNECT_ATTEMPTS`](2) 次。存在理由(2026-10-31 两组对照实验):
/// agent 串行 accept 期间不泵 socket,新握手被残骸/僵尸阻塞实测 5~8.5s
/// (杀会话后必现、干净连发 8 次亦有 1 次);第一拨 5s 被切是预期成本,
/// 第二拨的窗口天然落在释放点之后。包级机制未逐行核实(R-15),此处只
/// 陈述实测:**第一拨 5s 超时 → 第二拨成功**(kill 场景 total ≈5-11s 内接住)。
/// - 全程 [`CONNECT_TOTAL_BUDGET`](13s) 硬上限 = 5+8 完整两拨,
/// 防重试组合出无界等待;
/// - 真不可达 ≈13s 报 Timeout(修复前 32s),**全程 UI 可取消**——
/// U8 判据形态由此更新(plan V-A1)。
///
/// 已知边界:DNS 解析走同步 `to_socket_addrs`,current-thread runtime 里阻塞调用
/// 不能被 timeout 打断;本机实测解析失败是亚秒级。极端 DNS 挂起属阶段 4 网络治理面。
pub async fn connect(cfg: ConnectConfig) -> Result<Self, ConnectError> {
let attempts = async {
let mut attempt = 0usize;
loop {
attempt += 1;
// 第一拨短握手(快速判不可达),重拨拨长握手(接住僵尸释放)
let handshake = if attempt == 1 {
CONNECT_BUDGET
} else {
CONNECT_RETRY_BUDGET
};
match Self::connect_once(cfg.clone(), handshake).await {
Ok(s) => return Ok(s),
Err(e) => {
if attempt >= MAX_CONNECT_ATTEMPTS || !is_retryable(e.kind) {
return Err(e);
}
log(&format!(
"connect 第 {attempt}/{MAX_CONNECT_ATTEMPTS} 拨失败: {}(重拨)",
e.detail
));
}
}
}
};
// 外层硬上限。Timeout::poll 先轮询内层——已成功返回的会话不会被
// 「刚好到点」的外层吞掉(D2 的教训:预算必须给合法路径留出完成空间)
match tokio::time::timeout(CONNECT_TOTAL_BUDGET, attempts).await {
Ok(r) => r,
Err(_) => Err(ConnectError::new(
ConnectErrorKind::Timeout,
format!(
"connect exceeded {}s total budget within {MAX_CONNECT_ATTEMPTS} attempts",
CONNECT_TOTAL_BUDGET.as_secs()
),
)),
}
}
async fn connect_once(cfg: ConnectConfig, handshake_budget: Duration) -> Result<Self, ConnectError> {
rustls::crypto::ring::default_provider()
.install_default()
.ok();
@ -175,11 +361,14 @@ impl Session {
.with_no_client_auth()
} else {
// 证书钉扎尚未接入(方案阶段 4)。这里明确报错而不是静默放行。
bail!("insecure_skip_verify=false 需要证书钉扎,尚未实现(见方案阶段 4)");
return Err(ConnectError::new(
ConnectErrorKind::Other,
"insecure_skip_verify=false 需要证书钉扎,尚未实现(见方案阶段 4)",
));
};
let qc: quinn::crypto::rustls::QuicClientConfig = Arc::new(tls)
.try_into()
.context("build QUIC client config")?;
.map_err(other_err)?;
// 接收窗口放大:视频下行大流量,quinn 默认 512KB 流窗口会卡吞吐
// (与桌面 viewer / agent 的实测调优一致)
@ -189,21 +378,54 @@ impl Session {
let mut client_config = quinn::ClientConfig::new(Arc::new(qc));
client_config.transport_config(Arc::new(transport));
let mut endpoint = quinn::Endpoint::client("0.0.0.0:0".parse()?)?;
let mut endpoint =
quinn::Endpoint::client("0.0.0.0:0".parse().map_err(other_err)?).map_err(other_err)?;
endpoint.set_default_client_config(client_config);
let PeerTarget::Direct { host, port } = &cfg.target;
let addr = (host.as_str(), *port)
.to_socket_addrs()?
.to_socket_addrs()
.map_err(|e| {
ConnectError::new(
ConnectErrorKind::Resolve,
format!("resolve {host}:{port}: {e}"),
)
})?
.next()
.with_context(|| format!("resolve {host}:{port}"))?;
let conn = endpoint
.connect(addr, "localhost")?
.await
.with_context(|| format!("QUIC connect {addr}"))?;
.ok_or_else(|| {
ConnectError::new(
ConnectErrorKind::Resolve,
format!("resolve {host}:{port}: no record"),
)
})?;
let conn = {
let connect_fut = endpoint.connect(addr, "localhost").map_err(|e| {
ConnectError::new(classify_connect_error(&e), e.to_string())
})?;
// 单次尝试的握手预算(第一拨 5s、重拨拨 8s,见 connect() 注释)。
// 超时归类 Timeout(可重拨);真不可达由重拨打满总预算后报出。
match tokio::time::timeout(handshake_budget, connect_fut).await {
Ok(Ok(c)) => c,
Ok(Err(e)) => {
return Err(ConnectError::new(
classify_connection_error(&e),
e.to_string(),
))
}
Err(_) => {
return Err(ConnectError::new(
ConnectErrorKind::Timeout,
format!(
"QUIC handshake exceeded {}s budget",
handshake_budget.as_secs()
),
))
}
}
};
// ---- 流 1:认证 + 视频下行 ----
let (mut v_send, mut v_recv) = conn.open_bi().await?;
let (mut v_send, mut v_recv) = conn.open_bi().await.map_err(other_err)?;
write_control(
&mut v_send,
&ControlMsg::Auth {
@ -212,30 +434,59 @@ impl Session {
},
)
.await
.context("send Auth")?;
.map_err(other_err)?;
let desktop: (u32, u32);
match read_control(&mut v_recv).await.context("read AuthOk")? {
// 认证应答预算 11s:agent 串行 accept,僵尸会话最长 ~10s(R-5)才轮到本请求。
// 这一段超时的语义是「连上了但对端一直不应答」——同样归类 Timeout,detail 说清。
let first_msg = match tokio::time::timeout(AUTH_BUDGET, read_control(&mut v_recv)).await {
Ok(r) => r.map_err(other_err)?,
Err(_) => {
return Err(ConnectError::new(
ConnectErrorKind::Timeout,
format!(
"connected but auth reply exceeded {}s (agent busy?)",
AUTH_BUDGET.as_secs()
),
))
}
};
match first_msg {
ControlMsg::AuthOk {
width,
height,
proto_version,
} => {
if proto_version != PROTO_VERSION {
bail!(
"protocol mismatch: agent={proto_version}, viewer-core={PROTO_VERSION}"
);
return Err(ConnectError::new(
ConnectErrorKind::Version,
format!(
"protocol mismatch: agent={proto_version}, viewer-core={PROTO_VERSION}"
),
));
}
desktop = (width, height);
log(&format!("auth ok, remote desktop {width}x{height}"));
}
ControlMsg::AuthFail(e) => bail!("auth failed: {e}"),
m => bail!("unexpected first message: {m:?}"),
ControlMsg::AuthFail(e) => {
return Err(ConnectError::new(
classify_auth_fail(&e),
format!("auth failed: {e}"),
))
}
m => {
return Err(ConnectError::new(
ConnectErrorKind::Other,
format!("unexpected first message: {m:?}"),
))
}
}
// ---- 流 2:输入 + 控制 + Ping/Pong ----
let (mut ctl_send, mut ctl_recv) = conn.open_bi().await?;
let (mut ctl_send, mut ctl_recv) = conn.open_bi().await.map_err(other_err)?;
// 入会即请求 IDR:从 GOP 中间接入时裸 P 帧解不出来(桌面端同款教训)
write_control(&mut ctl_send, &ControlMsg::KeyFrameReq).await?;
write_control(&mut ctl_send, &ControlMsg::KeyFrameReq)
.await
.map_err(other_err)?;
let stats = Arc::new(std::sync::Mutex::new(SessionStats::default()));
@ -447,8 +698,138 @@ mod tests {
insecure_skip_verify: false,
};
match Session::connect(cfg).await {
Err(e) => assert!(e.to_string().contains("证书钉扎"), "实际错误: {e}"),
Err(e) => {
assert_eq!(e.kind, ConnectErrorKind::Other);
assert!(e.detail.contains("证书钉扎"), "实际错误: {e}");
}
Ok(_) => panic!("insecure_skip_verify=false 必须明确报错,不得静默放行"),
}
}
#[test]
fn auth_fail_classification_uses_agent_message_prefix() {
// 判据来自 agent 侧的真实文案(agent/src/main.rs:111),不是自造样本
assert_eq!(
classify_auth_fail("protocol version mismatch: client=2, server=3"),
ConnectErrorKind::Version
);
assert_eq!(classify_auth_fail("bad code"), ConnectErrorKind::Auth);
// 未知形态宁可归 Auth 也不静默
assert_eq!(
classify_auth_fail("expected Auth, got Ping(0)"),
ConnectErrorKind::Auth
);
}
#[test]
fn ffi_error_codes_are_stable_and_unique() {
// FFI 编号是 Kotlin 映射表的契约面:改动=公共 API 变更。
// 期望值写死(R-8:判据是等于期望值,不是"没重复")。
let expect = [
(ConnectErrorKind::Resolve, 1),
(ConnectErrorKind::Timeout, 2),
(ConnectErrorKind::Refused, 3),
(ConnectErrorKind::Auth, 4),
(ConnectErrorKind::Version, 5),
(ConnectErrorKind::Other, 9),
];
for (kind, code) in expect {
assert_eq!(kind.code(), code);
}
}
#[tokio::test]
async fn unresolvable_host_is_classified_as_resolve() {
// 本机环境事实(2026-10-31 实测):这台办公机的 DNS 拦截层(Meta/198.18.0.1)
// 把 `.invalid` 直接**解析成功**并指向 198.18.0.x 假地址——RFC 6761 的
// NXDOMAIN 保证在这条链路上不成立,`.invalid` 样本会走成 Timeout。
// 含空格的 name 在任何 resolver 里都是非法输入、快速失败(实测 26ms),
// 判据不依赖本机 DNS 行为。
let cfg = ConnectConfig::direct("a b", 47800, "t");
let e = match Session::connect(cfg).await {
Ok(_) => panic!("非法主机名必须解析失败"),
Err(e) => e,
};
assert_eq!(e.kind, ConnectErrorKind::Resolve, "实际: {e:?}");
assert!(e.detail.contains("a b"), "detail 必须含原输入: {e}");
}
#[tokio::test]
async fn unreachable_peer_redials_then_reports_timeout() {
// TEST-NET-1(RFC 5737):无可达主机 → 每拨握手都无应答。
// 方案①(用户裁决):第一拨 5s 超时 → 重拨 → 第二拨 8s 超时 → 报 Timeout。
// 期望落点 [13s-1s, 14s+2s):
// - 下界 ≈ 两拨握手预算之和,证明**第二拨真的跑完**(重拨丢失就是回归);
// - 上界证明总预算封顶(不会滚向第三拨)。
// 修复前(quinn 默认)这里要 32s;U8 判据形态更新为「≤13s 且全程可取消」。
let cfg = ConnectConfig::direct("192.0.2.1", 47800, "t");
let t0 = Instant::now();
let e = match Session::connect(cfg).await {
Ok(_) => panic!("TEST-NET-1 无主机,必须失败"),
Err(e) => e,
};
let elapsed = t0.elapsed();
assert_eq!(e.kind, ConnectErrorKind::Timeout, "实际: {e:?}");
let two_dials = CONNECT_BUDGET + CONNECT_RETRY_BUDGET;
assert!(
elapsed >= two_dials - Duration::from_secs(1)
&& elapsed < CONNECT_TOTAL_BUDGET + Duration::from_secs(2),
"重拨后的等待 {elapsed:?} 不在 [{two_dials:?}-1s, {CONNECT_TOTAL_BUDGET:?}+2s)"
);
}
#[test]
fn budget_constants_match_redial_design() {
// 方案①常量关系固化(2026-10-31 用户裁决):
// - 第一拨握手 5s:不可达的快速探测
// - 第二拨握手 8s:第二拨起点在 ~5s,需盖住实测僵尸释放点 ~10.2s
// (R-5 idle 超时)+ 余量——比第一拨长是这个物理原因,不是随手加大
// - 总预算 14s:两拨之和 + 1s,让内层各拨 Timeout 先触发(错误 detail
// 精确到段),外层只封顶失控组合
// - 认证 11s:> 僵尸释放 10.2s(实测),握手成功后排队不被压断
assert_eq!(CONNECT_BUDGET, Duration::from_secs(5));
assert_eq!(CONNECT_RETRY_BUDGET, Duration::from_secs(8));
assert_eq!(AUTH_BUDGET, Duration::from_secs(11));
assert_eq!(CONNECT_TOTAL_BUDGET, Duration::from_secs(14));
assert_eq!(MAX_CONNECT_ATTEMPTS, 2);
assert!(CONNECT_RETRY_BUDGET > CONNECT_BUDGET, "第二拨必须更长才盖得住释放点");
assert!(CONNECT_TOTAL_BUDGET > CONNECT_BUDGET + CONNECT_RETRY_BUDGET);
assert!(AUTH_BUDGET > Duration::from_secs(10));
// 重拨名单是策略面:只有暂时性失败可重试,改动必须过这条测试
assert!(is_retryable(ConnectErrorKind::Timeout));
assert!(is_retryable(ConnectErrorKind::Refused));
for k in [
ConnectErrorKind::Resolve,
ConnectErrorKind::Auth,
ConnectErrorKind::Version,
ConnectErrorKind::Other,
] {
assert!(!is_retryable(k), "{k:?} 是确定性失败,不得重拨");
}
}
#[test]
fn connection_error_variants_map_to_declared_kinds() {
// 只测**可安全构造**的单元变体(unit variants):TransportError /
// ConnectionClosed / ApplicationClose 的字段是 `bytes::Bytes`,测试构造
// 需要把 bytes 提为直接依赖——为分类表测试引依赖不值当(偏差 D-R1)。
// 这三个结构变体的映射由 match 的穷尽性 + 上面的注释钉住;
// 真实归类靠 V-A2 的端到端(端口上跑非 QUIC 服务)覆盖。
assert_eq!(
classify_connection_error(&quinn::ConnectionError::TimedOut),
ConnectErrorKind::Timeout
);
assert_eq!(
classify_connection_error(&quinn::ConnectionError::Reset),
ConnectErrorKind::Refused
);
assert_eq!(
classify_connection_error(&quinn::ConnectionError::VersionMismatch),
ConnectErrorKind::Other
);
assert_eq!(
classify_connection_error(&quinn::ConnectionError::LocallyClosed),
ConnectErrorKind::Other
);
}
}