From cc0def3415b72ef1171c891a63a52d81231f44a6 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Thu, 12 Nov 2026 22:25:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(download):=20=E7=BB=99=E6=AF=8F=E6=9D=A1?= =?UTF-8?q?=E4=B8=8B=E8=BD=BD=20await=20=E4=B8=8A=E7=95=8C=E5=B9=B6?= =?UTF-8?q?=E5=9C=A8=E8=B6=85=E6=97=B6=E6=97=B6=E6=8A=A5=E5=87=BA=E3=80=8C?= =?UTF-8?q?=E5=8D=A1=E5=9C=A8=E5=93=AA=E4=B8=80=E6=AE=B5=E3=80=8D=EF=BC=88?= =?UTF-8?q?M289=20/=20P19=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 全仓下载路径此前只有两处无界 await(分块与单连接的 `send()`),reqwest 自己的 `timeout` 又是 `None`:服务器接了连接不回一个字节 ⇒ future 永不结束, `ull java install` 那条 `silent()` 路径零事件零日志,用户只看到「卡住不动」。 上界选「单次请求 + 读空闲」,**不是**「整作业总预算」: * `region.rs` 的 3s 是探测极小响应(整作业 = 一次 GET,两者同一件事); * `font.rs` 的 ATTEMPT/TOTAL 是几百 KB 的字体(整作业几秒就完); * 188MB 的 JDK 两者都不能照抄,所以分三段互不顶替的界,理由与取数**只写在** `crates/core/src/download/stall.rs` 模块头一处(含「别改成 reqwest 的 timeout, 它覆盖整个响应体读取」)。 改动: * 新增 `download::stall`:`StallStage`(连接 / 等响应头 / 等字节 / 等分块回收 / 校验)+ `StallReason`(期限用完 / 没有新字节)+ `StallReport`(阶段 + 已收 / 预期字节 + 无进展时长)+ `job_budget`(总期限按 32 KiB/s 平均吞吐反推)。 * `http::send_bounded` / `http::read_step`:两处 `send()` 与两处读循环的**唯一**构造点; 超时错误带阶段与三个数,`message` 里也抄一份(整体失败时 `suppressed` 只搬 message)。 * `transfer_chunk::await_chunks`:`JoinSet::join_next` 加总期限与无进度窗口,两者靠 `StallReason` 分开报;窗口内有新字节就续期(慢但有效不误杀)。 * `verify_part` 挪到 blocking 池并套上界:校验阶段卡住也报得出来(阶段=校验)。 * `engine.rs`:整体失败时把**最后一次**尝试的失速现场带上去,并写进用户可见的那行字。 * 配置:`response_timeout` 60s / `chunk_stall_timeout` 120s / `chunk_job_floor` 10min / `verify_timeout` 120s(`EngineConfig`,未动设置 schema)。 测试(全离线、`127.0.0.1`、<2s):夹具新增 `silent()` / `stalling_after` / `short_body` / `oversize_body` / `dripping`;用例覆盖静默服务端、中途卡住、短包、超长包、慢但持续 (4 倍于上界也要成功)、超时后引擎仍能下载(RAII 许可自还,不写手动归还)、以及分块 作业的「无进度 vs 总期限」两条。 另:`config/tests.rs` 的一处表达式在去缩进后行宽变化,随本 commit 由 rustfmt 合成一行 (纯搬移 commit 提交时它仍是逐字节原文)。 --- crates/core/src/download/chunked.rs | 25 +- crates/core/src/download/config.rs | 39 +++ crates/core/src/download/config/tests.rs | 4 +- crates/core/src/download/engine.rs | 15 +- crates/core/src/download/engine/inspect.rs | 1 - crates/core/src/download/error.rs | 39 +++ crates/core/src/download/http.rs | 65 ++++ crates/core/src/download/mod.rs | 2 + crates/core/src/download/stall.rs | 312 +++++++++++++++++ crates/core/src/download/transfer.rs | 32 +- crates/core/src/download/transfer_chunk.rs | 148 ++++++-- crates/core/tests/download/chunked/timeout.rs | 170 ++++++++++ crates/core/tests/download/chunked_test.rs | 4 + .../tests/download/engine_mock/timeout.rs | 319 ++++++++++++++++++ .../core/tests/download/engine_mock_test.rs | 4 + crates/core/tests/download/support/mod.rs | 132 +++++++- 16 files changed, 1243 insertions(+), 68 deletions(-) create mode 100644 crates/core/src/download/stall.rs create mode 100644 crates/core/tests/download/chunked/timeout.rs create mode 100644 crates/core/tests/download/engine_mock/timeout.rs diff --git a/crates/core/src/download/chunked.rs b/crates/core/src/download/chunked.rs index 61853e3..9a0a948 100644 --- a/crates/core/src/download/chunked.rs +++ b/crates/core/src/download/chunked.rs @@ -7,10 +7,10 @@ use std::path::Path; use std::sync::atomic::{AtomicU64, Ordering}; -use std::time::Duration; use super::error::{DownloadError, DownloadErrorKind}; use super::http; +use super::stall::Deadlines; use super::types::ResourceKind; /// 默认分块连接数(设计文档 Q19 定案 N=4)。 @@ -209,12 +209,15 @@ pub fn content_range_matches(requested: ByteRange, value: Option<&str>) -> bool /// /// 服务器若对带 `Range` 的请求返回 200(无视 Range),立刻返回 `RangeUnsupported` /// —— 引擎据此退化单连接,而不是把整份文件重复写 N 遍。 +/// +/// 两处 `await` 都有界(M289):等响应头用 [`http::send_bounded`],等正文用 +/// [`http::read_step`],上界来自 [`Deadlines`]。失速时错误里带阶段与本段口径的字节数。 pub async fn fetch_chunk( client: &reqwest::Client, url: &str, range: ByteRange, part: &Path, - read_timeout: Duration, + deadlines: Deadlines, progress: &AtomicU64, ) -> Result { use tokio::io::{AsyncSeekExt, AsyncWriteExt}; @@ -222,10 +225,7 @@ pub async fn fetch_chunk( let request = client .get(url) .header(reqwest::header::RANGE, range.header_value()); - let response = request - .send() - .await - .map_err(|err| http::classify_reqwest(&err, url))?; + let response = http::send_bounded(request, url, deadlines, Some(range.len())).await?; let status = response.status().as_u16(); if status == 200 { @@ -264,11 +264,14 @@ pub async fn fetch_chunk( let mut stream = response.bytes_stream(); let mut written = 0_u64; loop { - let next = tokio::time::timeout(read_timeout, futures_util::StreamExt::next(&mut stream)) - .await - .map_err(|_| { - DownloadError::new(DownloadErrorKind::Timeout, format!("分块读取超时({url})")) - })?; + let next = http::read_step( + futures_util::StreamExt::next(&mut stream), + deadlines.read_idle, + written, + Some(range.len()), + url, + ) + .await?; let Some(chunk) = next else { break }; let chunk = chunk.map_err(|err| http::classify_reqwest(&err, url))?; file.write_all(&chunk) diff --git a/crates/core/src/download/config.rs b/crates/core/src/download/config.rs index 3de7b08..94232d8 100644 --- a/crates/core/src/download/config.rs +++ b/crates/core/src/download/config.rs @@ -14,6 +14,7 @@ use std::time::Duration; use super::backoff::BackoffPolicy; use super::chunked::ChunkedConfig; use super::region::RegionProfile; +use super::stall::{self, Deadlines}; use super::types::ResumePolicy; /// 全局并发许可数的出厂默认(`download.concurrency` 的 schema 默认值同源)。 @@ -51,6 +52,21 @@ pub struct EngineConfig { pub connect_timeout: Duration, /// 单次读超时(读空闲判定,不是整体超时),默认 30s。 pub read_timeout: Duration, + /// 等响应头的上界(**单次请求**:`send()` 到响应头为止,不含正文),默认 60s。 + /// + /// 取数与「为什么不是整作业总预算」见 [`super::stall`] 模块头那一处,别在调用点另写一套理由。 + pub response_timeout: Duration, + /// 分块作业「已这么久没有任何新字节」的判定窗口,默认 120s。 + /// + /// 抓的是「读超时不响、但整个作业不动」的那类卡住(分块任务卡在写盘上)。 + pub chunk_stall_timeout: Duration, + /// 分块作业总期限的下限,默认 10min(实际期限取它与「按平均吞吐下限反推的时长」的较大者, + /// 见 [`super::stall::job_budget`])。 + pub chunk_job_floor: Duration, + /// 整文件重算摘要(校验阶段)的上界,默认 120s。 + /// + /// 正常盘上 188MB 的 SHA-1 不到 1s;这条只抓「校验被卡住」,不参与慢链路判定。 + pub verify_timeout: Duration, /// User-Agent(唯一定义在 `http::USER_AGENT`;两条 HTTP 路径发的都是这一份,别在这里另拼)。 pub user_agent: String, /// 进度事件节流间隔,默认 250ms。 @@ -70,6 +86,10 @@ impl Default for EngineConfig { max_attempts_per_source: DEFAULT_MAX_ATTEMPTS_PER_SOURCE, connect_timeout: Duration::from_secs(10), read_timeout: Duration::from_secs(30), + response_timeout: Duration::from_secs(60), + chunk_stall_timeout: Duration::from_secs(120), + chunk_job_floor: Duration::from_secs(600), + verify_timeout: Duration::from_secs(120), user_agent: super::http::USER_AGENT.to_string(), progress_throttle: Duration::from_millis(250), backoff: BackoffPolicy::default(), @@ -79,6 +99,25 @@ impl Default for EngineConfig { } } +impl EngineConfig { + /// 两段共存的界(等响应头 / 等下一块字节):所有 HTTP 读路径都从这里取,别各自读字段。 + #[must_use] + pub const fn deadlines(&self) -> Deadlines { + Deadlines { + response: self.response_timeout, + read_idle: self.read_timeout, + } + } + + /// 分块作业的总期限(总长未知时只用下限)。 + #[must_use] + pub fn chunk_job_budget(&self, expected: Option) -> Duration { + expected.map_or(self.chunk_job_floor, |total| { + stall::job_budget(total, self.chunk_job_floor) + }) + } +} + /// `download.gh_proxy` 的三种取值(三态语义见 [`gh_proxy_override`])。 #[derive(Debug, Clone, PartialEq, Eq, Default)] pub enum GhProxyOverride { diff --git a/crates/core/src/download/config/tests.rs b/crates/core/src/download/config/tests.rs index 49d2add..1538681 100644 --- a/crates/core/src/download/config/tests.rs +++ b/crates/core/src/download/config/tests.rs @@ -140,8 +140,8 @@ fn gh_proxy_override_reaches_the_profile_and_the_chain() { "出厂套前缀" ); - let off = DownloadTuning::from_settings(None, None, Some("off")) - .apply_gh_proxy(mirror_first.clone()); + let off = + DownloadTuning::from_settings(None, None, Some("off")).apply_gh_proxy(mirror_first.clone()); assert!( !sources(&off).contains(&SOURCE_GH_PROXY.to_string()), "关了就没候选" diff --git a/crates/core/src/download/engine.rs b/crates/core/src/download/engine.rs index 8ad6173..e3fe3b5 100644 --- a/crates/core/src/download/engine.rs +++ b/crates/core/src/download/engine.rs @@ -30,6 +30,7 @@ use super::http; use super::mirrors::MirrorTable; use super::progress::{ProgressEmitter, ProgressEvent}; use super::region::RegionProfile; +use super::stall::StallReport; use super::types::{ Candidate, DownloadId, DownloadOutcome, DownloadRequest, ResourceKind, ResumePolicy, SOURCE_OFFICIAL, @@ -175,6 +176,8 @@ impl DownloadEngine { let mut part = PartFile::for_dest(&req.dest); let mut suppressed: Vec = Vec::new(); let mut previous: Option = None; + // 失速现场只跟**最后一次**尝试:更早的源怎么挂的不能冒充根因。 + let mut last_stall: Option = None; for candidate in &plan { if let Some(from) = previous.take() { @@ -203,6 +206,7 @@ impl DownloadEngine { return Ok(outcome); } Err(err) => { + last_stall = err.stall.as_deref().copied(); // 4xx(NotFound)不算「源坏」:院校镜像只留最新补丁,请求旧补丁必然 404, // 记成失败会把唯一可用镜像冷却掉(M33/M34 实测:TUNA 每目录只保 1 个归档)。 if err.kind == DownloadErrorKind::NotFound { @@ -245,11 +249,18 @@ impl DownloadEngine { // 全部候选失败:保留中间文件与元数据,下次(甚至下次启动)可续传。 part.preserve(); self.flush_health(); + // 用户唯一看得见的那一行(`silent()` 引擎零事件、零日志):把最后一次失速的 + // 阶段与三个数带上,否则「卡住不动」只是换成「下载失败」,仍然什么也说不出来。 + let detail = match &last_stall { + Some(stall) => format!("(最后一次失速:{stall})"), + None => String::new(), + }; Err(DownloadError::new( DownloadErrorKind::AllSourcesFailed, - format!("{} 个候选源全部失败:{}", plan.len(), req.url), + format!("{} 个候选源全部失败:{}{detail}", plan.len(), req.url), ) - .with_suppressed(suppressed)) + .with_suppressed(suppressed) + .with_last_stall(last_stall)) } fn finish( diff --git a/crates/core/src/download/engine/inspect.rs b/crates/core/src/download/engine/inspect.rs index d03b2c4..2cc9756 100644 --- a/crates/core/src/download/engine/inspect.rs +++ b/crates/core/src/download/engine/inspect.rs @@ -54,5 +54,4 @@ impl DownloadEngine { pub fn cache(&self) -> &ContentCache { &self.cache } - } diff --git a/crates/core/src/download/error.rs b/crates/core/src/download/error.rs index 3152de9..afc8cf1 100644 --- a/crates/core/src/download/error.rs +++ b/crates/core/src/download/error.rs @@ -6,6 +6,8 @@ use std::fmt; use serde::{Deserialize, Serialize}; +use super::stall::StallReport; + /// 错误分类:决定「同源重试」「立刻换源」还是「整体失败」。 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -82,6 +84,17 @@ pub struct DownloadError { /// 全 5xx ⇒ `ServerError`;混合按「网络错 > 5xx > 404」取最可诊断的那个。 /// `None` = 没有汇总过(单源直接失败,或调用方自己构造的错误)。 pub leaf: Option, + /// **失速现场**(M289):卡在哪一段、已收/预期多少字节、多久没进展。 + /// + /// 为什么挂在错误上而不是只拼进 `message`:上层(CLI、诊断页)要能不解字符串就分类; + /// `message` 里也照抄一份(见 [`DownloadError::stalled`]),因为整体失败时 + /// `suppressed` 只搬 `message`,而那一行正是用户唯一看得见的「卡在哪」。 + /// `None` = 不是失速失败(或没带现场)。 + /// + /// 为什么是 `Box`:`DownloadError` 已经贴着 clippy 的 `result_large_err` 上限 + /// (128 字节),再加 48 字节的现场会让**所有** `Result<_, DownloadError>` 变红; + /// 失速是罕见事件,多一次分配换回小错误类型是划算的。 + pub stall: Option>, } impl DownloadError { @@ -95,9 +108,35 @@ impl DownloadError { attempts: 0, suppressed: Vec::new(), leaf: None, + stall: None, } } + /// 失速错误(`kind = Timeout`):`message` 里带阶段 + 已收/预期/无进展三个数。 + /// + /// 为什么 `message` 也要带一份:候选源外循环最后只剩下 + /// `format!("{source}: {message}")` 进 `suppressed`,上层拿到的就是这行字; + /// 只把现场放在结构化字段里,用户照样只看到「下载失败」。 + #[must_use] + pub fn stalled(stall: StallReport, context: impl fmt::Display) -> Self { + Self::new(DownloadErrorKind::Timeout, format!("{context}:{stall}")).with_stall(stall) + } + + /// 标注失速现场。 + #[must_use] + pub fn with_stall(mut self, stall: StallReport) -> Self { + self.stall = Some(Box::new(stall)); + self + } + + /// 有现场就挂上(整体失败时把**最后一个**源的现场带上来;`None` 表示最后一次尝试 + /// 不是失速失败,那就别拿更早的源来冒充根因)。 + #[must_use] + pub fn with_last_stall(mut self, stall: Option) -> Self { + self.stall = stall.map(Box::new); + self + } + /// 标注出错的源。 #[must_use] pub fn with_source(mut self, source: impl Into) -> Self { diff --git a/crates/core/src/download/http.rs b/crates/core/src/download/http.rs index 9db7b1e..33263ce 100644 --- a/crates/core/src/download/http.rs +++ b/crates/core/src/download/http.rs @@ -7,6 +7,8 @@ //! * 重定向上限 20 跳(承 HMCL),跳转后仍按最终 URL 做 host 校验由调用方负责; //! * 错误分类只做「同源重试 / 立刻换源 / 整体失败」三态,不做细颗粒映射。 +use std::time::Duration; + use reqwest::header::{ HeaderMap, HeaderName, ACCEPT_ENCODING, CONTENT_LENGTH, CONTENT_TYPE, ETAG, LAST_MODIFIED, }; @@ -14,6 +16,7 @@ use reqwest::redirect::Policy; use super::config::EngineConfig; use super::error::{DownloadError, DownloadErrorKind}; +use super::stall::{Deadlines, StallReport, StallStage}; /// 重定向上限(承 HMCL 的 20 跳)。 pub const MAX_REDIRECTS: usize = 20; @@ -57,6 +60,68 @@ pub fn build_client(cfg: &EngineConfig) -> Result, +) -> Result { + let started = std::time::Instant::now(); + match tokio::time::timeout(deadlines.response, request.send()).await { + Ok(Ok(response)) => Ok(response), + Ok(Err(err)) => { + // 只有**超时**才是失速:连接被拒 / DNS 失败 / 证书错误照旧走原分类, + // 别把「连不上」包装成 Timeout(那会让上层对错误的判断全部偏一档)。 + if err.is_timeout() { + let stage = if err.is_connect() { + StallStage::Connect + } else { + StallStage::AwaitHeaders + }; + return Err(DownloadError::stalled( + StallReport::deadline(stage, 0, expected, started.elapsed()), + format!("{err}({url})"), + )); + } + Err(classify_reqwest(&err, url)) + } + Err(_) => Err(DownloadError::stalled( + StallReport::deadline(StallStage::AwaitHeaders, 0, expected, deadlines.response), + format!("等响应头超过 {}ms({url})", deadlines.response.as_millis()), + )), + } +} + +/// 等下一块正文(两处读循环共用这一处构造):超时 ⇒ 阶段「等字节」+ 已收/预期/无进展三个数。 +/// +/// 注意它是**空闲**判定:`step` 只要在两块之间按时完成就一直续期,慢但持续的流不会被它误杀。 +pub async fn read_step( + step: F, + idle: Duration, + received: u64, + expected: Option, + url: &str, +) -> Result, DownloadError> +where + F: std::future::Future>, +{ + match tokio::time::timeout(idle, step).await { + Ok(item) => Ok(item), + Err(_) => Err(DownloadError::stalled( + StallReport::no_progress(StallStage::AwaitBytes, received, expected, idle), + format!("等下一块字节超过 {}ms({url})", idle.as_millis()), + )), + } +} + /// 把 `reqwest::Error` 映射成我们的错误分类。 #[must_use] pub fn classify_reqwest(err: &reqwest::Error, url: &str) -> DownloadError { diff --git a/crates/core/src/download/mod.rs b/crates/core/src/download/mod.rs index fbccf5e..d070221 100644 --- a/crates/core/src/download/mod.rs +++ b/crates/core/src/download/mod.rs @@ -62,6 +62,7 @@ pub mod region; pub mod region_system; pub mod resume; pub mod shared; +pub mod stall; pub mod types; pub mod verify; @@ -92,6 +93,7 @@ pub use self::progress::{ProgressEmitter, ProgressEvent, ProgressThrottle}; pub use self::region::{RegionMode, RegionProfile, RegionTrigger, Signals}; pub use self::region_system::detect_signals; pub use self::resume::{RestartReason, ResumeDecision, ResumeMeta}; +pub use self::stall::{Deadlines, StallReason, StallReport, StallStage, MIN_AVG_THROUGHPUT}; pub use self::types::{ Candidate, DownloadId, DownloadOutcome, DownloadRequest, HashAlgo, Integrity, ResourceKind, ResumePolicy, SOURCE_OFFICIAL, diff --git a/crates/core/src/download/stall.rs b/crates/core/src/download/stall.rs new file mode 100644 index 0000000..0334d8c --- /dev/null +++ b/crates/core/src/download/stall.rs @@ -0,0 +1,312 @@ +//! 失速诊断与下载的三段上界(M289 / 台账 P19)。 +//! +//! # 事实:下载引擎曾是全仓唯一没有超时的那条路 +//! +//! 全仓下载路径此前只有两处**无界** `await`:分块请求与单连接请求的 `send()`。 +//! `build_client` 只设了 `connect_timeout`,reqwest 自己的 `timeout` 默认 `None` —— +//! 于是「服务器接了连接、不回一个字节」= 这个 future 永不结束,而下 188MB 的作业 +//! 就在那里静静挂着(`ull java install` 走的是 `silent()` 引擎:零事件、零日志, +//! 用户只能看到一个不动的主界面)。 +//! +//! 同仓两个先例:`region.rs` 给探测的 `send()` 包了 3s,`font.rs` 给整个 +//! `download_bytes` 包了总预算——下载引擎自己反而是唯一没有界的那条路。 +//! +//! # 取舍:三段可以共存的界,**不是**「整作业必须在 N 秒内下完」 +//! +//! 两个先例为什么取值天差地别,看它们的作业有多大: +//! +//! * `region.rs:288` 的 3s —— 探测一个**极小响应**(`Range: bytes=0-0`)。整个作业就是 +//! 一次只用拿几个字节的 GET,所以「单次请求上界」与「整作业预算」在这里是同一件事, +//! 3s 对两者都成立;它紧是因为**慢探测本身就是坏结果**。 +//! * `font.rs:232-238` 的 `ATTEMPT_DEADLINE` / `TOTAL_DEADLINE` —— 字体是几百 KB 的 +//! 单文件,整作业几秒就完,所以能一刀切死「慢」;代价是它必须对最后一个候选放宽 +//! (每个候选都从 0 字节重下,单次上限就是一条吞吐门槛,卡死最后一个等于把 +//! 「慢但能下完」判成永久失败——那里的注释已经记了这笔账)。 +//! +//! 下 188MB 的 JDK 与两者都不同:整作业预算必须 ≥ 慢链路上的真实耗时,否则 +//! 「慢但有效」被判死——这是加超时最常见的自杀方式。所以这里**不设**「作业必须 N 秒 +//! 内下完」,而设三段互不顶替的界: +//! +//! 1. [`StallStage::AwaitHeaders`](`EngineConfig::response_timeout`,**单次请求**): +//! `send()` 只到「响应头到手」为止,不含正文,所以对 188MB 的作业它是一次性开销, +//! 取多紧都不会误杀正在下的大文件; +//! 2. [`StallStage::AwaitBytes`](`EngineConfig::read_timeout`,**空闲**判定): +//! 字节一直在动就一直续期。每秒只挤几 KB 的链路也算「在动」,只有**真的不动**了才判失速; +//! 3. [`StallStage::AwaitChunkJoin`](分块作业):一个分块卡住不许 4 条连接陪着挂到天荒地老。 +//! 这里同时给「无进度窗口」与「总期限」,但总期限按**平均吞吐下限**反推 +//! ([`job_budget`]),不是拍一个固定秒数:能把 188MB 在期限内下完的慢链路照样活。 +//! +//! **不要**把它改成 `reqwest::ClientBuilder::timeout(...)`:reqwest 的 `timeout` 覆盖 +//! **整个响应体**读取,那正好就是「慢但有效被判死」的那个坑(它的默认 `None` 也是 P19 +//! 记的另一半事实)。每个阶段各自的界才分得清「连不上」「不回话」「不回字节」。 +//! +//! # 数字从哪来(全仓只写这一处) +//! +//! * `response_timeout` = 60s:连接已有 10s 的 `connect_timeout` 兜底,这 60s 是**连接 +//! 建立之后**等响应头的余额——CDN 排队、回源冷启动都在几十秒量级;响应头本身只有 +//! 几百字节,能回的话 60s 内必回。 +//! * `read_timeout` = 30s(既有值,不动):空闲判定。持续有字节就永远不触发。 +//! * `chunk_stall_timeout` = 120s:4 条连接**整体**一个新字节都没有两分钟,只可能是 +//! 分块任务卡住(写盘卡在杀软/网络盘上时,读超时不会响,只有这个会)。 +//! * [`MIN_AVG_THROUGHPUT`] = 32 KiB/s:分块作业的总期限 = 总字节 / 32 KiB/s +//! (188MB ≈ 100 分钟)。即「整作业的平均速度掉到 32 KiB/s 以下才判死」;比这更快的 +//! 慢链路仍会下完。**这是平均吞吐门槛,不是慢链路的死刑**:真实公网慢链路的取值 +//! 没有真机验证过(M289 报告里列为未验项)。 +//! * `verify_timeout` = 120s:188MB 的 SHA-1 在正常盘上不到 1s,120s 只抓「整文件重算 +//! 摘要被卡住」(网络盘掉线、杀软锁文件),不参与慢链路判定。 +//! +//! # 口径(读这两个数之前先看这里) +//! +//! `received` / `expected` 是**该阶段自己的口径**:等字节 = 这条连接收到多少 / 该连接 +//! 被要求收多少(分块就是那一段的长度);等分块回收 = 整个作业收了多少 / 作业总长。 + +use std::fmt; +use std::time::Duration; + +/// 分块作业的**最低平均吞吐**(字节/秒):整作业总期限按它反推(见 [`job_budget`])。 +pub const MIN_AVG_THROUGHPUT: u64 = 32 * 1024; + +/// 卡在哪一段(诊断字段;`as_str` 是稳定 id,`label` 是给人看的中文)。 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum StallStage { + /// 连接还没建立(DNS / TCP / TLS)。 + Connect, + /// 连接已建立,等响应头(`send()` 的上界)。 + AwaitHeaders, + /// 响应头已到,等下一块正文字节(读空闲判定的上界)。 + AwaitBytes, + /// 分块作业在等分块任务回收(整作业的无进度 / 总期限上界)。 + AwaitChunkJoin, + /// 整文件重算摘要(校验阶段)。 + Verify, +} + +impl StallStage { + /// 稳定字符串 id(日志 / 诊断页用,不要拿来显示给用户)。 + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Connect => "connect", + Self::AwaitHeaders => "await_headers", + Self::AwaitBytes => "await_bytes", + Self::AwaitChunkJoin => "await_chunk_join", + Self::Verify => "verify", + } + } + + /// 给人看的中文短名。 + #[must_use] + pub const fn label(self) -> &'static str { + match self { + Self::Connect => "建立连接", + Self::AwaitHeaders => "等响应头", + Self::AwaitBytes => "等字节", + Self::AwaitChunkJoin => "等分块回收", + Self::Verify => "校验", + } + } +} + +/// 为什么判失速:期限用完,还是「一段时间没有新字节」。 +/// +/// 两者必须分得开:一条 1 KB/s 但一直在动的链路**没有**无进展,只是慢; +/// 而 4 条连接同时不动两分钟与「整作业平均吞吐跌破下限」是两种不同的坏法, +/// 修法也不同(前者等一等/换源,后者基本只能换源)。 +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum StallReason { + /// 一段期限用完(等响应头 / 校验 / 整作业总期限)。 + Deadline, + /// 已有一段时间没有任何新字节。 + NoProgress, +} + +impl StallReason { + /// 稳定字符串 id。 + #[must_use] + pub const fn as_str(self) -> &'static str { + match self { + Self::Deadline => "deadline", + Self::NoProgress => "no_progress", + } + } + + /// 给人看的中文短名。 + #[must_use] + pub const fn label(self) -> &'static str { + match self { + Self::Deadline => "期限用完", + Self::NoProgress => "没有新字节", + } + } +} + +/// 一次失速的现场:阶段 + 已收 / 预期字节 + 无进展时长(口径见模块头)。 +/// +/// 挂在 [`super::error::DownloadError::stall`] 上(结构化),同时由 +/// [`super::error::DownloadError::stalled`] 抄一份进 `message`——因为整体失败时 +/// `suppressed` 只搬 `message`,而上层(CLI / 诊断页)正靠那一行回答「卡在哪」。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StallReport { + /// 卡在哪一段。 + pub stage: StallStage, + /// 为什么判失速。 + pub reason: StallReason, + /// 已收字节(口径见模块头)。 + pub received: u64, + /// 预期字节(未知时 `None`:总长不明的大文件)。 + pub expected: Option, + /// 已经多久没有新字节。 + pub idle: Duration, +} + +impl StallReport { + /// 「一段期限用完」的现场。 + #[must_use] + pub const fn deadline( + stage: StallStage, + received: u64, + expected: Option, + waited: Duration, + ) -> Self { + Self { + stage, + reason: StallReason::Deadline, + received, + expected, + idle: waited, + } + } + + /// 「已 `idle` 没有新字节」的现场。 + #[must_use] + pub const fn no_progress( + stage: StallStage, + received: u64, + expected: Option, + idle: Duration, + ) -> Self { + Self { + stage, + reason: StallReason::NoProgress, + received, + expected, + idle, + } + } +} + +impl fmt::Display for StallReport { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!( + f, + "阶段={}({});已收 {} 字节", + self.stage.label(), + self.stage.as_str(), + self.received + )?; + match self.expected { + Some(expected) => write!(f, " / 预期 {expected} 字节")?, + None => write!(f, " / 预期未知")?, + } + write!( + f, + ";已 {}ms 无进展;原因={}", + self.idle.as_millis(), + self.reason.label() + ) + } +} + +/// 两段可以共存的界(等响应头 / 等下一块字节),由 `EngineConfig::deadlines()` 构造。 +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct Deadlines { + /// 发出请求 → 响应头到手的上界(单次请求,不含正文)。 + pub response: Duration, + /// 两块正文之间的空闲上界(字节在动就一直续期)。 + pub read_idle: Duration, +} + +/// 分块作业的总期限:`max(floor, 总字节 / MIN_AVG_THROUGHPUT)`。 +/// +/// 为什么不是固定秒数:固定值对小作业太松、对大作业太紧;按平均吞吐反推, +/// 「能下得完的慢链路」永远不会撞上它,撞上的只有「慢到已经没意义」的作业。 +#[must_use] +pub fn job_budget(total: u64, floor: Duration) -> Duration { + floor.max(Duration::from_secs(total / MIN_AVG_THROUGHPUT)) +} + +#[cfg(test)] +mod tests { + use super::{job_budget, Deadlines, StallReason, StallReport, StallStage, MIN_AVG_THROUGHPUT}; + use std::time::Duration; + + /// 总期限按平均吞吐反推:188MB 的作业拿到 ≈100 分钟,小作业只拿到下限。 + /// + /// 变异证明:把 `job_budget` 写成常量(比如 `Duration::from_secs(30)`)⇒ 第一条红; + /// 把 `floor.max(..)` 写成 `min` ⇒ 第二条红(188MB 只剩 100 分钟里的 `floor`)。 + #[test] + fn the_job_budget_scales_with_size_and_never_drops_below_the_floor() { + let floor = Duration::from_secs(600); + let big = job_budget(188 * 1024 * 1024, floor); + let by_throughput = Duration::from_secs(188 * 1024 * 1024 / MIN_AVG_THROUGHPUT); + assert_eq!(big, by_throughput, "大作业的期限由平均吞吐决定"); + assert!(big > Duration::from_secs(90 * 60), "188MB 至少给到 90 分钟"); + assert_eq!( + job_budget(8 * 1024, floor), + floor, + "小作业只拿到下限(吞吐算式在这里没有意义)" + ); + } + + /// 诊断串必须同时有:阶段(中英文都给)、已收字节、预期字节、无进展时长、原因。 + /// + /// 变异证明:删掉 `stage` 那一段 ⇒ 前两条 red;把 `received` / `expected` 写成常量 + /// 或删掉 ⇒ 对应断言 red;把 `reason` 删掉 ⇒ 最后一条 red。 + #[test] + fn the_report_spells_out_stage_three_numbers_and_reason() { + let report = StallReport::no_progress( + StallStage::AwaitBytes, + 4096, + Some(188 * 1024 * 1024), + Duration::from_millis(30_000), + ); + let text = report.to_string(); + assert!(text.contains("等字节"), "{text}"); + assert!(text.contains("await_bytes"), "{text}"); + assert!(text.contains("已收 4096 字节"), "{text}"); + assert!(text.contains("预期 197132288 字节"), "{text}"); + assert!(text.contains("已 30000ms 无进展"), "{text}"); + assert!(text.contains("原因=没有新字节"), "{text}"); + assert!(report.reason == StallReason::NoProgress); + + // 总长未知也要能说清楚(不是「预期 0」)。 + let unknown = + StallReport::deadline(StallStage::AwaitHeaders, 0, None, Duration::from_secs(60)); + assert!(unknown.to_string().contains("预期未知"), "{unknown}"); + assert_eq!(unknown.reason, StallReason::Deadline); + } + + /// 阶段 id 是日志里的 grep 锚点:改名字等于改诊断口径,得先改这里。 + #[test] + fn stage_and_reason_ids_are_stable() { + let stages = [ + (StallStage::Connect, "connect", "建立连接"), + (StallStage::AwaitHeaders, "await_headers", "等响应头"), + (StallStage::AwaitBytes, "await_bytes", "等字节"), + (StallStage::AwaitChunkJoin, "await_chunk_join", "等分块回收"), + (StallStage::Verify, "verify", "校验"), + ]; + for (stage, id, label) in stages { + assert_eq!(stage.as_str(), id); + assert_eq!(stage.label(), label); + } + assert_eq!(StallReason::Deadline.as_str(), "deadline"); + assert_eq!(StallReason::NoProgress.as_str(), "no_progress"); + + let deadlines = Deadlines { + response: Duration::from_secs(60), + read_idle: Duration::from_secs(30), + }; + assert_eq!(deadlines.response, Duration::from_secs(60)); + } +} diff --git a/crates/core/src/download/transfer.rs b/crates/core/src/download/transfer.rs index 42f27ee..c1b7aeb 100644 --- a/crates/core/src/download/transfer.rs +++ b/crates/core/src/download/transfer.rs @@ -73,7 +73,8 @@ impl DownloadEngine { let mut cut = ResumeCut::none(); match decision { ResumeDecision::Complete { size } => { - self.verify_part(part, integrity, id)?; + self.verify_part(part, integrity, req.expected_size, id) + .await?; return Ok(AttemptOk { bytes: size, resumed_from: size, @@ -133,15 +134,21 @@ impl DownloadEngine { } } - let response = request.send().await.map_err(|err| { - http::classify_reqwest(&err, &candidate.url).with_source(candidate.source.as_str()) - })?; + let response = http::send_bounded( + request, + &candidate.url, + self.cfg.deadlines(), + req.expected_size, + ) + .await + .map_err(|err| err.with_source(candidate.source.as_str()))?; let status = response.status(); if status.as_u16() == 416 { return match resume::on_range_unsatisfiable(part.len(), req.expected_size) { ResumeDecision::Complete { size } => { - self.verify_part(part, integrity, id)?; + self.verify_part(part, integrity, req.expected_size, id) + .await?; Ok(AttemptOk { bytes: size, resumed_from: size, @@ -226,18 +233,15 @@ impl DownloadEngine { let mut written = offset; let mut stream = response.bytes_stream(); loop { - let next = tokio::time::timeout( - self.cfg.read_timeout, + let next = http::read_step( futures_util::StreamExt::next(&mut stream), + self.cfg.read_timeout, + written, + total, + &candidate.url, ) .await - .map_err(|_| { - DownloadError::new( - DownloadErrorKind::Timeout, - format!("读取超时({})", candidate.url), - ) - .with_source(candidate.source.as_str()) - })?; + .map_err(|err| err.with_source(candidate.source.as_str()))?; let Some(chunk) = next else { break }; let chunk = chunk.map_err(|err| http::classify_reqwest(&err, &candidate.url))?; if let Some(hasher) = hasher.as_mut() { diff --git a/crates/core/src/download/transfer_chunk.rs b/crates/core/src/download/transfer_chunk.rs index c9bbe8a..bd13a29 100644 --- a/crates/core/src/download/transfer_chunk.rs +++ b/crates/core/src/download/transfer_chunk.rs @@ -12,6 +12,7 @@ use super::chunked; use super::engine::{AttemptOk, DownloadEngine}; use super::error::{DownloadError, DownloadErrorKind}; use super::progress::ProgressEvent; +use super::stall::{StallReason, StallReport, StallStage}; use super::types::{Candidate, DownloadId, DownloadRequest, Integrity}; use super::verify; @@ -58,37 +59,22 @@ impl DownloadEngine { let counter = Arc::new(AtomicU64::new(0)); let reporter = self.spawn_reporter(id, Arc::clone(&counter), total); + let deadlines = self.cfg.deadlines(); let mut set = tokio::task::JoinSet::new(); for range in ranges { let client = self.http.clone(); let url = candidate.url.clone(); let path = part.path().to_path_buf(); - let read_timeout = self.cfg.read_timeout; let progress = Arc::clone(&counter); set.spawn(async move { - chunked::fetch_chunk(&client, &url, range, &path, read_timeout, &progress).await + chunked::fetch_chunk(&client, &url, range, &path, deadlines, &progress).await }); } - let mut failure: Option = None; - while let Some(joined) = set.join_next().await { - match joined { - Ok(Ok(_)) => {} - Ok(Err(err)) => { - failure = Some(err.with_source(candidate.source.as_str())); - set.abort_all(); - break; - } - Err(err) => { - failure = Some(DownloadError::new( - DownloadErrorKind::Network, - format!("分块任务异常:{err}"), - )); - set.abort_all(); - break; - } - } - } + let failure = self + .await_chunks(&mut set, &counter, total, &candidate.url) + .await + .map(|err| err.with_source(candidate.source.as_str())); reporter.abort(); if let Some(err) = failure { @@ -97,7 +83,8 @@ impl DownloadEngine { } // 分块是乱序写,只能整文件重算一次摘要。 - self.verify_part(part, req.integrity.as_ref(), id)?; + self.verify_part(part, req.integrity.as_ref(), Some(total), id) + .await?; self.emit(ProgressEvent::Progress { id, downloaded: total, @@ -109,6 +96,85 @@ impl DownloadEngine { }) } + /// 回收分块任务,同时盯着「整个作业有没有在动」。 + /// + /// 两道判据必须分得开(另一处是单块自己的读空闲超时,阶段 = 等字节): + /// * **无进度**:`counter` 连续 `chunk_stall_timeout` 没有新字节。分块任务卡在写盘上 + /// (杀软/网络盘)时单块读超时不会响,只有这道抓得住——现场留给 `join_next` 的正是 + /// 「188MB 落完 99% 之后永久挂住」那个形态; + /// * **总期限**:`chunk_job_budget`(按平均吞吐下限反推,见 `stall` 模块头)用完。 + /// + /// 两者都归到阶段 [`StallStage::AwaitChunkJoin`],靠 [`StallReason`] 区分。 + async fn await_chunks( + &self, + set: &mut tokio::task::JoinSet>, + counter: &AtomicU64, + total: u64, + url: &str, + ) -> Option { + let budget = self.cfg.chunk_job_budget(Some(total)); + let hard_deadline = tokio::time::Instant::now() + budget; + let mut last_seen = 0_u64; + let mut last_progress = tokio::time::Instant::now(); + loop { + let seen = counter.load(Ordering::Relaxed); + if seen > last_seen { + last_seen = seen; + last_progress = tokio::time::Instant::now(); + } + let wake = (last_progress + self.cfg.chunk_stall_timeout).min(hard_deadline); + match tokio::time::timeout_at(wake, set.join_next()).await { + Ok(Some(Ok(Ok(_)))) => {} + Ok(Some(Ok(Err(err)))) => { + set.abort_all(); + return Some(err); + } + Ok(Some(Err(err))) => { + set.abort_all(); + return Some(DownloadError::new( + DownloadErrorKind::Network, + format!("分块任务异常:{err}"), + )); + } + // 全部回收(成功)。 + Ok(None) => return None, + Err(_) => { + let now = tokio::time::Instant::now(); + // 窗口内确实有字节进来 ⇒ 续期(「慢但有效」不被误杀的机制就在这一行)。 + let seen = counter.load(Ordering::Relaxed); + if now < hard_deadline && seen > last_seen { + last_seen = seen; + last_progress = now; + continue; + } + let report = if now >= hard_deadline { + StallReport::deadline( + StallStage::AwaitChunkJoin, + seen, + Some(total), + now - last_progress, + ) + } else { + StallReport::no_progress( + StallStage::AwaitChunkJoin, + seen, + Some(total), + now - last_progress, + ) + }; + set.abort_all(); + let context = match report.reason { + StallReason::Deadline => { + format!("分块作业超过总期限 {}s({url})", budget.as_secs()) + } + StallReason::NoProgress => format!("分块作业长时间没有新字节({url})"), + }; + return Some(DownloadError::stalled(report, context)); + } + } + } + } + fn spawn_reporter( &self, id: DownloadId, @@ -131,10 +197,11 @@ impl DownloadEngine { }) } - pub(crate) fn verify_part( + pub(crate) async fn verify_part( &self, part: &mut PartFile, integrity: Option<&Integrity>, + expected: Option, id: DownloadId, ) -> Result<(), DownloadError> { let Some(integrity) = integrity else { @@ -145,15 +212,42 @@ impl DownloadEngine { algo: integrity.algo, expected_prefix: integrity.short_prefix().to_string(), }); - let actual = verify::digest_file(part.path(), integrity.algo) - .map_err(|err| io_err("读取中间文件失败", &err))?; - if verify::matches(&integrity.hex, &actual) { + // 整文件重算是同步阻塞 IO:挪到 blocking 池并套上界,否则「校验阶段卡住」 + // (网络盘掉线、杀软锁文件)照样是永久挂住,且它发生在字节全部收完之后, + // 从进度上看与「下载完成」一模一样。 + let on_disk = part.len(); + let path = part.path().to_path_buf(); + let algo = integrity.algo; + let digest = tokio::time::timeout( + self.cfg.verify_timeout, + tokio::task::spawn_blocking(move || verify::digest_file(&path, algo)), + ) + .await + .map_err(|_| { + DownloadError::stalled( + StallReport::deadline( + StallStage::Verify, + on_disk, + expected, + self.cfg.verify_timeout, + ), + format!( + "整文件重算摘要超过 {}ms", + self.cfg.verify_timeout.as_millis() + ), + ) + })? + .map_err(|err| { + DownloadError::new(DownloadErrorKind::Io, format!("校验任务异常退出:{err}")) + })? + .map_err(|err| io_err("读取中间文件失败", &err))?; + if verify::matches(&integrity.hex, &digest) { return Ok(()); } part.discard(); Err(DownloadError::new( DownloadErrorKind::Integrity, - format!("校验不符:期望 {} 实得 {actual}", integrity.hex), + format!("校验不符:期望 {} 实得 {digest}", integrity.hex), )) } } diff --git a/crates/core/tests/download/chunked/timeout.rs b/crates/core/tests/download/chunked/timeout.rs new file mode 100644 index 0000000..bf780e1 --- /dev/null +++ b/crates/core/tests/download/chunked/timeout.rs @@ -0,0 +1,170 @@ +//! 分块作业的两道失速判据(M289 / 台账 P19 的「188MB 落完 + 永久挂住」现场)。 +//! +//! `JoinSet::join_next` 此前**没有总期限**:一个分块卡在 #1,`while let Some(...)` 就永远 +//! 不结束——文件已经 `set_len(total)` 预分配(NTFS 上真占空间),复审看到的 `.ullpart` +//! 是满大小,看起来像「下完了」。这里钉住两件事,且两者必须**分得开**(错误里靠 +//! `StallReason` 区分,而不是一律「下载失败」): +//! +//! * 无进度:`counter` 连续 `chunk_stall_timeout` 没有新字节 ⇒ `reason = no_progress`; +//! * 总期限:`chunk_job_budget`(按平均吞吐下限反推)用完 ⇒ `reason = deadline`。 +//! +//! 变异:把 `await_chunks` 换回 `while let Some(joined) = set.join_next().await`(无期限) +//! ⇒ 两条都靠外层 `GATE` 挂断而红;把 `StallReason` 的分支写成常量 ⇒ 第二条红 +//! (它断言的是 `deadline`,而无进度那条断言 `no_progress`)。 +//! +//! 拆成子模块:`chunked_test.rs` 已 379 行。 + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use unlucky_core::download::backoff::BackoffPolicy; +use unlucky_core::download::cache::ContentCache; +use unlucky_core::download::chunked::ChunkedConfig; +use unlucky_core::download::config::EngineConfig; +use unlucky_core::download::engine::DownloadEngine; +use unlucky_core::download::error::DownloadErrorKind; +use unlucky_core::download::health::HealthRegistry; +use unlucky_core::download::mirrors::MirrorTable; +use unlucky_core::download::region::RegionProfile; +use unlucky_core::download::stall::{StallReason, StallStage}; +use unlucky_core::download::types::{DownloadRequest, ResourceKind}; + +use super::support::{temp_dir, MockResponse, MockServer}; + +/// 外层保险:分块作业若真的无期限,用例在这里挂断并说明「这就是 bug」。 +const GATE: Duration = Duration::from_secs(5); + +fn config() -> EngineConfig { + EngineConfig { + max_attempts_per_source: 1, + backoff: BackoffPolicy::none(), + connect_timeout: Duration::from_secs(2), + response_timeout: Duration::from_secs(2), + // 读空闲给到 10s:让**整体**判据先响,证明它自己有效(不是靠单块读超时兜底)。 + read_timeout: Duration::from_secs(10), + chunk_stall_timeout: Duration::from_secs(10), + chunk_job_floor: Duration::from_secs(10), + verify_timeout: Duration::from_secs(10), + chunked: ChunkedConfig { + enabled: true, + connections: 4, + min_size: 1024, + }, + ..EngineConfig::default() + } +} + +fn harness(tag: &str, cfg: EngineConfig) -> (DownloadEngine, std::path::PathBuf) { + let dir = temp_dir(tag); + let cache = Arc::new(ContentCache::open(dir.join("cache"))); + let engine = DownloadEngine::new( + cfg, + Arc::new(MirrorTable::builtin()), + Arc::new(RegionProfile::direct(0)), + Arc::new(HealthRegistry::in_memory()), + cache, + ); + (engine, dir) +} + +fn request(url: &str, dest: std::path::PathBuf, size: u64) -> DownloadRequest { + DownloadRequest::new(ResourceKind::JavaRuntime, url, &dest).with_size(size) +} + +/// 4 条分块连接全部「回了头、不给字节」:整个作业两分钟不动 ⇒ 「等分块回收 + 没有新字节」。 +#[tokio::test] +async fn a_chunked_job_with_no_progress_is_named_await_chunk_join() { + let payload = vec![1_u8; 16 * 1024]; + let size = payload.len() as u64; + let stalled = |_| MockResponse::ok(payload.clone()).stalling_after(0); + let server = MockServer::start((0..4).map(stalled).collect()).await; + let (engine, dir) = harness( + "chunked-timeout-stall", + EngineConfig { + chunk_stall_timeout: Duration::from_millis(200), + ..config() + }, + ); + let url = server.url("/jdk.zip"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("jdk.zip"), size)), + ) + .await + .expect("分块回收必须有期限:外层 5s 挂断正说明 `join_next` 仍无界") + .expect_err("4 条连接都不给字节时必须失败"); + let elapsed = started.elapsed(); + + assert!( + elapsed < Duration::from_secs(2), + "应在无进度窗口附近返回:{elapsed:?}" + ); + assert_eq!(error.kind, DownloadErrorKind::AllSourcesFailed); + let stall = error.stall.as_deref().expect("分块失速必须带现场"); + assert_eq!(stall.stage, StallStage::AwaitChunkJoin, "{stall}"); + assert_eq!(stall.reason, StallReason::NoProgress, "{stall}"); + assert_eq!( + stall.received, 0, + "一个字节都没到(口径是整个作业):{stall}" + ); + assert_eq!(stall.expected, Some(size), "{stall}"); + assert!( + stall.idle >= Duration::from_millis(200), + "无进展时长必须 ≥ 判定窗口:{stall}" + ); + assert!( + error.to_string().contains("等分块回收"), + "用户那行字要说清卡在哪:{error}" + ); + server.stop(); +} + +/// 分块作业的**总期限**:一直在动,但平均吞吐低到没意义 ⇒ 「等分块回收 + 期限用完」。 +/// +/// 与上一条的区别只有 `reason`:这条的 `received > 0`(一直在动),上一条是 0。 +#[tokio::test] +async fn a_chunked_job_that_is_too_slow_hits_the_total_budget() { + let payload = vec![2_u8; 16 * 1024]; + let size = payload.len() as u64; + // 每条连接每 20ms 滴 1 字节:一直在动(不会触发无进度),但远低于平均吞吐下限。 + let dripping = |_| MockResponse::ok(payload.clone()).dripping(1, Duration::from_millis(20)); + let server = MockServer::start((0..4).map(dripping).collect()).await; + let (engine, dir) = harness( + "chunked-timeout-budget", + EngineConfig { + chunk_job_floor: Duration::from_millis(300), + ..config() + }, + ); + let url = server.url("/jdk.zip"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("jdk.zip"), size)), + ) + .await + .expect("分块作业总期限必须有界(外层挂断 = 无界)") + .expect_err("慢到没有意义的作业必须在总期限处失败"); + let elapsed = started.elapsed(); + + assert!( + elapsed < Duration::from_secs(2), + "总期限 300ms,实际 {elapsed:?}(太长说明期限没被用上)" + ); + let stall = error.stall.as_deref().expect("总期限失败也要带现场"); + assert_eq!(stall.stage, StallStage::AwaitChunkJoin, "{stall}"); + assert_eq!( + stall.reason, + StallReason::Deadline, + "总期限要用 deadline,不能与「无进度」混成一句:{stall}" + ); + assert!( + stall.received > 0, + "这条一直在动,已收字节必须 > 0(0 就说明它其实是卡住而不是慢):{stall}" + ); + assert_eq!(stall.expected, Some(size), "{stall}"); + server.stop(); +} diff --git a/crates/core/tests/download/chunked_test.rs b/crates/core/tests/download/chunked_test.rs index 906baf3..71c059d 100644 --- a/crates/core/tests/download/chunked_test.rs +++ b/crates/core/tests/download/chunked_test.rs @@ -373,3 +373,7 @@ async fn rejects_lying_content_range_and_falls_back() { assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload); server.stop(); } + +/// 分块作业的两道失速判据(无进度 / 总期限):拆到子模块守 400 行硬线。 +#[path = "chunked/timeout.rs"] +mod timeout; diff --git a/crates/core/tests/download/engine_mock/timeout.rs b/crates/core/tests/download/engine_mock/timeout.rs new file mode 100644 index 0000000..ccb63b6 --- /dev/null +++ b/crates/core/tests/download/engine_mock/timeout.rs @@ -0,0 +1,319 @@ +//! 超时的**可诊断性**与**不误杀**(M289 / 台账 P19 的下半段)。 +//! +//! 这条 mission 的价值不在「加了超时」,而在两件别的事: +//! +//! 1. **下次挂住时能一眼看出卡在哪**:每个超时错误都带阶段(连接 / 等响应头 / 等字节 / +//! 等分块回收 / 校验)+ 已收字节 / 预期字节 / 无进展时长三个数,且这几个数真的来自现场 +//! (不是常量)。变异:把 `StallReport` 的 `stage` 字段删掉、或把 `received` 写成 `0` +//! 常量 ⇒ `the_stage_numbers_are_real` 红(下面贴了红原文)。 +//! 2. **慢但有效不许被误杀**:单次请求上界只覆盖「响应头到手」,正文靠空闲判定; +//! 所以一条 4 秒才下完 96 字节的滴流必须成功。变异:把等响应头的上界套到整个传输上 +//! ⇒ `a_slow_but_steady_stream_is_not_killed` 红。 +//! +//! 还有一条来自复审的纠正:`engine.rs` 的 `_permit` 是 RAII 守卫(future 被 drop 时自己 +//! 归还),所以这里**不写**「手动归还」补丁;用例断言的是**行为**——超时之后引擎还能 +//! 正常完成一次新下载(`max_concurrency = 1` 时这条才是可观测的)。 +//! +//! 拆成子模块:`engine_mock_test.rs` 已 364 行,再加这几条会越 400 行硬线。 +//! 全离线(`127.0.0.1` 回环)、每条 <2s(含外层 `timeout` 保险)。 + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use unlucky_core::download::cache::ContentCache; +use unlucky_core::download::chunked::ChunkedConfig; +use unlucky_core::download::config::EngineConfig; +use unlucky_core::download::engine::DownloadEngine; +use unlucky_core::download::error::DownloadErrorKind; +use unlucky_core::download::health::HealthRegistry; +use unlucky_core::download::mirrors::MirrorTable; +use unlucky_core::download::stall::{StallReason, StallStage}; +use unlucky_core::download::types::{DownloadRequest, ResourceKind}; + +use super::direct_region; +use super::support::{sha1_hex, temp_dir, MockResponse, MockServer}; + +/// 外层保险:引擎若真的无界,用例在这里挂断并给出「这就是 bug」的说明,而不是跑不完。 +const GATE: Duration = Duration::from_secs(5); + +fn engine(cfg: EngineConfig, tag: &str) -> (DownloadEngine, std::path::PathBuf) { + let dir = temp_dir(tag); + let cache = Arc::new(ContentCache::open(dir.join("cache"))); + let engine = DownloadEngine::new( + cfg, + Arc::new(MirrorTable::builtin()), + Arc::new(direct_region()), + Arc::new(HealthRegistry::in_memory()), + cache, + ); + (engine, dir) +} + +fn config(response_ms: u64, read_ms: u64) -> EngineConfig { + EngineConfig { + max_attempts_per_source: 1, + backoff: unlucky_core::download::backoff::BackoffPolicy::none(), + connect_timeout: Duration::from_secs(2), + response_timeout: Duration::from_millis(response_ms), + read_timeout: Duration::from_millis(read_ms), + chunked: ChunkedConfig::disabled(), + ..EngineConfig::default() + } +} + +fn silent_responses(count: usize) -> Vec { + (0..count).map(|_| MockResponse::silent()).collect() +} + +fn request(url: &str, dest: std::path::PathBuf, size: Option) -> DownloadRequest { + let request = DownloadRequest::new(ResourceKind::ClientJar, url, &dest); + match size { + Some(size) => request.with_size(size), + None => request, + } +} + +/// `silent()`:服务端接了连接、读完请求头、一个字都不回。 +/// +/// 这就是 P19 记的完全一致现场——`send()` 无界 ⇒ 这个 future 永不结束。 +/// 断言四件事:①引擎**自己**在 2s 内返回(外层 5s 挂断 = 无界)②阶段是「等响应头」 +/// ③三个数是真的(已收 0 / 预期 = 文件长度 / 无进展 ≈ 上界)④用户看得见的那行字里 +/// 也有阶段与数字(`silent()` 引擎零事件、零日志,只有这一行)。 +#[tokio::test] +async fn a_silent_server_times_out_and_names_the_stage_with_numbers() { + let payload = vec![7_u8; 64 * 1024]; + let size = payload.len() as u64; + let server = MockServer::start(silent_responses(2)).await; + let (engine, dir) = engine(config(300, 300), "timeout-silent"); + let url = server.url("/silent.jar"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("silent.jar"), Some(size))), + ) + .await + .expect("引擎必须自己超时返回:外层 5s 挂断正说明内部仍是**无界**的(这就是要修的 bug)") + .expect_err("silent 服务端必须失败"); + let elapsed = started.elapsed(); + + assert!( + elapsed < Duration::from_secs(2), + "超时必须在上界附近发生,实际 {elapsed:?}" + ); + assert_eq!(error.kind, DownloadErrorKind::AllSourcesFailed); + let stall = error + .stall + .as_deref() + .expect("整体失败必须带最后一次失速现场(否则用户只看到「下载失败」)"); + assert_eq!(stall.stage, StallStage::AwaitHeaders, "{stall}"); + assert_eq!(stall.reason, StallReason::Deadline, "{stall}"); + assert_eq!(stall.received, 0, "一个字节都还没到:{stall}"); + assert_eq!( + stall.expected, + Some(size), + "预期字节是这次请求的可诊断参数:{stall}" + ); + assert!( + stall.idle >= Duration::from_millis(300), + "无进展时长不得小于上界:{stall}" + ); + + let text = error.to_string(); + for needle in ["等响应头", "await_headers", "已收 0 字节"] { + assert!(text.contains(needle), "错误文案缺少「{needle}」:{text}"); + } + assert!( + text.contains("预期 65536 字节"), + "错误文案缺少预期字节:{text}" + ); + server.stop(); +} + +/// `stall_after`:响应头到手、正文写出一截、然后卡住。 +/// +/// 阶段必须是「等字节」而不是「等响应头」——两者同样是「没数据」,但修法不同 +/// (前者等/换源,后者多半是源坏了)。三个数按**这条连接**的口径:已收 = 真读到的量。 +#[tokio::test] +async fn a_stall_mid_body_is_named_await_bytes_with_the_real_byte_count() { + let payload = vec![9_u8; 32 * 1024]; + let size = payload.len() as u64; + let server = + MockServer::start(vec![MockResponse::ok(payload.clone()).stalling_after(64)]).await; + let (engine, dir) = engine(config(2_000, 300), "timeout-stall-body"); + let url = server.url("/stall.jar"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("stall.jar"), Some(size))), + ) + .await + .expect("读取必须有界(外层挂断 = 读循环仍无界)") + .expect_err("中途卡住必须失败"); + let elapsed = started.elapsed(); + + assert!( + elapsed < Duration::from_secs(2), + "应在上界附近返回:{elapsed:?}" + ); + let stall = error.stall.as_deref().expect("中途卡住必须带失速现场"); + assert_eq!(stall.stage, StallStage::AwaitBytes, "{stall}"); + assert_eq!(stall.reason, StallReason::NoProgress, "{stall}"); + assert_eq!( + stall.received, 64, + "已收字节必须是**真实**读到的量(服务端只发了 64 字节):{stall}" + ); + assert_eq!(stall.expected, Some(size), "{stall}"); + assert!(stall.idle >= Duration::from_millis(300), "{stall}"); +} + +/// 阳性对照①:**短包**——头部声称全量、正文只到一半就关连接。 +/// +/// 必须快速、确定地失败(长度不符),不是挂住、也不是「成功但文件短了」。 +#[tokio::test] +async fn a_short_body_fails_definitively_instead_of_hanging() { + let payload = vec![3_u8; 8 * 1024]; + let size = payload.len() as u64; + let server = MockServer::start(vec![MockResponse::ok(payload.clone()).short_body(4096)]).await; + let (engine, dir) = engine(config(2_000, 2_000), "timeout-short-body"); + let url = server.url("/short.jar"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("short.jar"), Some(size))), + ) + .await + .expect("短包必须快速失败,不许等读取上界") + .expect_err("短包必须失败"); + assert!( + started.elapsed() < Duration::from_secs(2), + "短包应立即判失败" + ); + assert_eq!(error.kind, DownloadErrorKind::AllSourcesFailed); + // reqwest 对「Content-Length 没读满就被关掉」报的是 `error decoding response body` + // (不是我们的「长度不符」),所以这里不钉它的文案:钉的是**有界 + 有成因 + 坏文件不落盘**。 + assert!( + !error.suppressed.is_empty(), + "短包必须留下一条逐源成因,而不是一句「下载失败」" + ); + assert!( + !dir.join("short.jar").exists(), + "短包不许落盘(宁可失败也不能交付半个文件)" + ); + assert!(error.stall.is_none(), "这不是失速(有明确终点):{error}"); +} + +/// 阳性对照②:**超长包**——不发 `Content-Length`、正文之后再多吐一堆字节。 +/// +/// 客户端按期望长度收,多出来的字节不能悄悄算进文件:必须判长度不符。 +#[tokio::test] +async fn an_oversize_body_is_rejected_by_length() { + let payload = vec![5_u8; 8 * 1024]; + let size = payload.len() as u64; + let server = + MockServer::start(vec![MockResponse::ok(payload.clone()).oversize_body(4096)]).await; + let (engine, dir) = engine(config(2_000, 2_000), "timeout-oversize-body"); + let url = server.url("/oversize.jar"); + + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("oversize.jar"), Some(size))), + ) + .await + .expect("超长包必须有界地失败") + .expect_err("超长包必须失败"); + assert!( + error + .suppressed + .iter() + .any(|item| item.contains("长度不符")), + "超长包要报长度不符(多出来的字节不许算进文件):{:?}", + error.suppressed + ); +} + +/// **慢但有效不许被误杀**(加超时最常见的自杀方式)。 +/// +/// 服务端每 40ms 滴 8 字节,96 字节要 ≈440ms 才下完;等响应头的上界只有 100ms。 +/// 只要上界只覆盖「响应头到手」,这条就成功;一旦有人把它套到整个传输上就红。 +#[tokio::test] +async fn a_slow_but_steady_stream_is_not_killed() { + let payload: Vec = (0..96).map(|index| (index % 251) as u8).collect(); + let device = MockResponse::ok(payload.clone()).dripping(8, Duration::from_millis(40)); + let server = MockServer::start(vec![device]).await; + let (engine, dir) = engine(config(100, 1_000), "timeout-slow-steady"); + let url = server.url("/slow.jar"); + let dest = dir.join("slow.jar"); + + let outcome = tokio::time::timeout( + GATE, + engine.download( + DownloadRequest::new(ResourceKind::ClientJar, url.as_str(), &dest) + .with_sha1(sha1_hex(&payload)) + .with_size(payload.len() as u64), + ), + ) + .await + .expect("慢但持续的流被误杀:时间上界被套到了整个传输上") + .expect("慢但持续的流必须成功"); + + assert_eq!(outcome.bytes, payload.len() as u64); + assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload); + server.stop(); +} + +/// 复审纠正的判据:**超时之后引擎还能正常下载**(`_permit` 是 RAII,问题在无界,不在归还)。 +/// +/// `max_concurrency = 1` 让「许可有没有回到池里」可观测:第一条被超时打断后, +/// 第二条若拿不到许可就会挂住 ⇒ 外层 `GATE` 挂断并报「许可没还回来」。 +#[tokio::test] +async fn after_a_timeout_the_engine_still_downloads_again() { + let payload = b"second download works".to_vec(); + let silent = MockServer::start(silent_responses(2)).await; + let good = MockServer::start(vec![MockResponse::ok(payload.clone())]).await; + let (engine, dir) = engine( + EngineConfig { + max_concurrency: 1, + ..config(200, 200) + }, + "timeout-permit", + ); + + let first = tokio::time::timeout( + GATE, + engine.download(request( + &silent.url("/dead.jar"), + dir.join("dead.jar"), + None, + )), + ) + .await + .expect("第一条必须自己超时(外层挂断 = 无界)") + .expect_err("silent 服务端必须失败"); + assert_eq!(first.kind, DownloadErrorKind::AllSourcesFailed); + assert_eq!( + engine.concurrency_probe(), + 1, + "超时之后许可必须回到池里(RAII 自还,不许写手动归还补丁)" + ); + + let outcome = tokio::time::timeout( + GATE, + engine.download( + DownloadRequest::new( + ResourceKind::ClientJar, + good.url("/ok.jar"), + dir.join("ok.jar"), + ) + .with_sha1(sha1_hex(&payload)), + ), + ) + .await + .expect("许可没还回来:第二次下载挂住了") + .expect("超时之后引擎必须还能完成一次新下载"); + assert_eq!(outcome.bytes, payload.len() as u64); + silent.stop(); + good.stop(); +} diff --git a/crates/core/tests/download/engine_mock_test.rs b/crates/core/tests/download/engine_mock_test.rs index 30604c4..eac51a3 100644 --- a/crates/core/tests/download/engine_mock_test.rs +++ b/crates/core/tests/download/engine_mock_test.rs @@ -362,3 +362,7 @@ async fn reports_progress_events_and_errors_when_all_sources_fail() { /// 「404 不是源坏了」的守卫(M197 P2-2 的第二档):拆到子模块守 400 行硬线。 #[path = "engine_mock/not_found.rs"] mod not_found; + +/// 超时的可诊断性 / 不误杀 / 超时后许可自还(M289):同样拆到子模块守 400 行硬线。 +#[path = "engine_mock/timeout.rs"] +mod timeout; diff --git a/crates/core/tests/download/support/mod.rs b/crates/core/tests/download/support/mod.rs index fb9330d..28a9f99 100644 --- a/crates/core/tests/download/support/mod.rs +++ b/crates/core/tests/download/support/mod.rs @@ -8,6 +8,7 @@ use std::net::SocketAddr; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; +use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; @@ -74,8 +75,23 @@ pub struct MockResponse { pub etag: Option, /// 返回 206 时故意把 `Content-Range` 写成从 0 开始(模拟段错位的坏服务器)。 pub lie_content_range: bool, + /// 读完请求头后**一个字节都不写**(服务端接了连接但永不回话)。 + /// + /// 这是「无界 `await`」唯一能表达成用例的形态:读方不设上界就永远不返回。 + pub silent: bool, + /// 写出这么多字节正文后就**停住不再写**(中途卡住),连接保持打开。 + pub stall_at: Option, + /// 正文实际只发这么多字节(头部仍声称全量 = 短包)。 + pub send_len: Option, + /// 正文之后**多**发这么多字节,且故意不发 `Content-Length`(超长包:HTTP/1.1 靠关连接分帧)。 + pub trailing_extra: Option, + /// 正文按 `piece` 字节一片、片间睡 `gap` 滴流(慢但持续,用来证明超时不会误杀它)。 + pub drip: Option<(usize, Duration)>, } +/// 夹具「挂住」的时长:远大于用例自己的上界,保证读方只能靠超时脱身。 +const HOLD: Duration = Duration::from_secs(30); + impl MockResponse { /// 200 + 完整内容(支持 Range)。 pub fn ok(body: impl Into>) -> Self { @@ -85,6 +101,11 @@ impl MockResponse { range_support: true, etag: Some("\"v1\"".to_string()), lie_content_range: false, + silent: false, + stall_at: None, + send_len: None, + trailing_extra: None, + drip: None, } } @@ -96,9 +117,46 @@ impl MockResponse { range_support: false, etag: None, lie_content_range: false, + silent: false, + stall_at: None, + send_len: None, + trailing_extra: None, + drip: None, } } + /// 200 但**永不回话**:读完请求头就挂住,一个字都不写。 + pub fn silent() -> Self { + Self { + silent: true, + ..Self::ok(Vec::new()) + } + } + + /// 写完 `bytes` 字节正文后停住(中途卡住,连接不断)。 + pub fn stalling_after(mut self, bytes: usize) -> Self { + self.stall_at = Some(bytes); + self + } + + /// 头部声称全量、正文只发 `sent` 字节(短包,客户端读不满)。 + pub fn short_body(mut self, sent: usize) -> Self { + self.send_len = Some(sent); + self + } + + /// 正文之后多发 `extra` 字节且不发 `Content-Length`(超长包)。 + pub fn oversize_body(mut self, extra: usize) -> Self { + self.trailing_extra = Some(extra); + self + } + + /// 正文按 `piece` 字节一片、片间 `gap` 滴流(慢但持续)。 + pub fn dripping(mut self, piece: usize, gap: Duration) -> Self { + self.drip = Some((piece, gap)); + self + } + /// 声明不支持 Range(模拟部分镜像)。 pub fn without_range(mut self) -> Self { self.range_support = false; @@ -198,7 +256,15 @@ async fn serve(mut stream: TcpStream, response: MockResponse) -> std::io::Result .map(str::trim) .and_then(parse_range_header); + if response.silent { + // 接了连接、读完请求头、一个字都不回:读方不设上界就永远挂在这里。 + tokio::time::sleep(HOLD).await; + return Ok(()); + } + let total = response.body.len() as u64; + // 超长包不能发 `Content-Length`,否则读方按声明截断、永远看不到多余字节。 + let with_length = response.trailing_extra.is_none(); let (status, headers, body) = match (range, response.range_support) { (Some((start, requested_end)), true) if start < total => { // 尊重请求里的结束偏移(分块下载必须逐段精确,否则组装后长度对不上)。 @@ -211,15 +277,14 @@ async fn serve(mut stream: TcpStream, response: MockResponse) -> std::io::Result } else { (start, end) }; - ( - 206, - vec![ - format!("Content-Range: bytes {claimed_start}-{claimed_end}/{total}"), - format!("Content-Length: {}", slice.len()), - "Accept-Ranges: bytes".to_string(), - ], - slice, - ) + let mut headers = vec![ + format!("Content-Range: bytes {claimed_start}-{claimed_end}/{total}"), + "Accept-Ranges: bytes".to_string(), + ]; + if with_length { + headers.push(format!("Content-Length: {}", slice.len())); + } + (206, headers, slice) } (Some(_), true) => ( 416, @@ -230,7 +295,10 @@ async fn serve(mut stream: TcpStream, response: MockResponse) -> std::io::Result Vec::new(), ), _ => { - let mut headers = vec![format!("Content-Length: {total}")]; + let mut headers = Vec::new(); + if with_length { + headers.push(format!("Content-Length: {total}")); + } if response.range_support { headers.push("Accept-Ranges: bytes".to_string()); } @@ -247,11 +315,53 @@ async fn serve(mut stream: TcpStream, response: MockResponse) -> std::io::Result } head.push_str("Connection: close\r\n\r\n"); stream.write_all(head.as_bytes()).await?; - stream.write_all(&body).await?; + write_body(&mut stream, &body, &response).await?; stream.flush().await?; Ok(()) } +/// 按夹具的口径写正文:正常 / 卡在中途 / 短包 / 超长尾巴 / 滴流。 +async fn write_body( + stream: &mut TcpStream, + body: &[u8], + response: &MockResponse, +) -> std::io::Result<()> { + if let Some(stall_at) = response.stall_at { + // 头部 + 这么多字节已发出,之后一个字不再写(连接不断):读方只能靠超时脱身。 + let sent = stall_at.min(body.len()); + stream.write_all(&body[..sent]).await?; + stream.flush().await?; + tokio::time::sleep(HOLD).await; + return Ok(()); + } + let limit = response.send_len.unwrap_or(body.len()).min(body.len()); + match response.drip { + Some((piece, gap)) => { + let piece = piece.max(1); + let mut offset = 0; + while offset < limit { + let end = (offset + piece).min(limit); + stream.write_all(&body[offset..end]).await?; + stream.flush().await?; + offset = end; + if offset < limit { + tokio::time::sleep(gap).await; + } + } + } + None => { + stream.write_all(&body[..limit]).await?; + stream.flush().await?; + } + } + if let Some(extra) = response.trailing_extra { + let filler = vec![0_u8; extra]; + stream.write_all(&filler).await?; + stream.flush().await?; + } + Ok(()) +} + /// 解析 `Range: bytes=-`(`end` 可缺省 = 到文件末尾)。 fn parse_range_header(value: &str) -> Option<(u64, Option)> { let (start, end) = value.split_once('-')?;