diff --git a/crates/core/src/download/bytes.rs b/crates/core/src/download/bytes.rs index f405402..9efb6f0 100644 --- a/crates/core/src/download/bytes.rs +++ b/crates/core/src/download/bytes.rs @@ -10,6 +10,7 @@ use super::engine::{millis, DownloadEngine}; use super::error::{leaf_of, DownloadError, DownloadErrorKind}; use super::http; use super::progress::ProgressEvent; +use super::stall::StallReport; use super::types::{Candidate, DownloadId, DownloadRequest}; use super::verify; @@ -31,10 +32,12 @@ impl DownloadEngine { )); } - let _permit = self.acquire().await?; + let _permit = self.acquire(req.expected_size).await?; let mut suppressed: Vec = Vec::new(); // 逐源的叶子分类(出去时汇总成一条,见 `DownloadError::leaf`)。 let mut kinds: Vec = Vec::new(); + // 与 `download()` 一致:最后一次尝试的失速现场带出去(含阶段与三个数)。 + let mut last_stall: Option = None; for candidate in &plan { // 与文件下载同一套规则:冷却中/刚出冷却的源只给一次探测机会。 @@ -59,13 +62,14 @@ impl DownloadEngine { return Ok(bytes); } Err(err) => { + last_stall = err.stall.as_deref().copied(); self.health.record_failure( &candidate.source, req.kind, &err.message, now_ms(), ); - let retry = err.kind.is_retryable() && attempt + 1 < max_attempts; + let retry = err.is_retryable() && attempt + 1 < max_attempts; let delay = self.next_delay(attempt); self.emit(ProgressEvent::SourceFailed { id, @@ -90,12 +94,17 @@ impl DownloadEngine { self.flush_health(); // 叶子成因要跟着出去:上层(在线内容源的 `Transport`)靠它区分「这个 URL 不存在」 // 与「没联网」,只留 `AllSourcesFailed` 的话那套分类在生产路径上永远不可达。 + let detail = match &last_stall { + Some(stall) => format!("(最后一次失速:{stall})"), + None => String::new(), + }; Err(DownloadError::new( DownloadErrorKind::AllSourcesFailed, - format!("所有候选源都取不到文本资源:{}", req.url), + format!("所有候选源都取不到文本资源:{}{detail}", req.url), ) .with_suppressed(suppressed) - .with_leaf(leaf_of(kinds))) + .with_leaf(leaf_of(kinds)) + .with_last_stall(last_stall)) } async fn fetch_bytes( @@ -103,18 +112,25 @@ impl DownloadEngine { req: &DownloadRequest, candidate: &Candidate, ) -> Result, DownloadError> { - let response = self - .http - .get(&candidate.url) - .send() - .await - .map_err(|err| http::classify_reqwest(&err, &candidate.url))?; + // 两处 await 都有界(复审 r1 p2-1):等响应头走 `send_bounded`(与文件下载同一构造点), + // 整包读改成流式累加 + `read_step` 的空闲判定。不能再退回 `response.bytes()`: + // 它是一次无界的整包读,`MAX_IN_MEMORY_BYTES` 只管大小不管时间。 + let deadlines = self.cfg.deadlines(); + let response = http::send_bounded( + self.http.get(&candidate.url), + &candidate.url, + deadlines, + req.expected_size, + ) + .await + .map_err(|err| err.with_source(candidate.source.as_str()))?; let status = response.status(); if !status.is_success() { return Err(http::classify_status(status.as_u16(), &candidate.url) .with_source(candidate.source.as_str())); } - if let Some(len) = http::content_length(response.headers()) { + let declared = http::content_length(response.headers()); + if let Some(len) = declared { if len > MAX_IN_MEMORY_BYTES { return Err(DownloadError::new( DownloadErrorKind::InvalidRequest, @@ -126,10 +142,42 @@ impl DownloadEngine { .with_source(candidate.source.as_str())); } } - let body = response - .bytes() + + let mut stream = response.bytes_stream(); + let mut body: Vec = Vec::with_capacity( + declared + .and_then(|len| usize::try_from(len).ok()) + .unwrap_or(0), + ); + loop { + let next = http::read_step( + futures_util::StreamExt::next(&mut stream), + deadlines.read_idle, + body.len() as u64, + declared, + &candidate.url, + ) .await - .map_err(|err| http::classify_reqwest(&err, &candidate.url))?; + .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).with_source(candidate.source.as_str()) + })?; + body.extend_from_slice(&chunk); + // `Content-Length` 可以撒谎、也可以缺失(靠关连接分帧):边读边卡,不能只看头。 + if body.len() as u64 > MAX_IN_MEMORY_BYTES { + return Err(DownloadError::new( + DownloadErrorKind::InvalidRequest, + format!( + "文本资源超过内存上限({} 字节),拒绝继续载入:{}", + body.len(), + candidate.url + ), + ) + .with_source(candidate.source.as_str())); + } + } + if let Some(integrity) = &req.integrity { let actual = verify::digest_bytes(integrity.algo, &body); if !verify::matches(&integrity.hex, &actual) { @@ -140,6 +188,6 @@ impl DownloadEngine { .with_source(candidate.source.as_str())); } } - Ok(body.to_vec()) + Ok(body) } } 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 20929a1..0f67ccb 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,26 @@ 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, + /// 等并发许可的上界,默认 120min。 + /// + /// 必须 ≥ 单个作业的总期限(最坏 ≈100 min 预算 + 校验),否则排队本身会被误杀; + /// 它挡的是「许可被永久占住」。取数见 [`super::stall`] 模块头。 + pub permit_timeout: Duration, /// User-Agent(唯一定义在 `http::USER_AGENT`;两条 HTTP 路径发的都是这一份,别在这里另拼)。 pub user_agent: String, /// 进度事件节流间隔,默认 250ms。 @@ -70,6 +91,11 @@ 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), + permit_timeout: Duration::from_secs(120 * 60), user_agent: super::http::USER_AGENT.to_string(), progress_throttle: Duration::from_millis(250), backoff: BackoffPolicy::default(), @@ -79,6 +105,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 { @@ -223,178 +268,5 @@ pub fn publish(cache_root: &Path, next: DownloadTuning) -> bool { } #[cfg(test)] -mod tests { - use super::*; - use crate::setting::keys::{launcher_schema, launcher_version}; - - /// **守门③**:schema 的默认值 / 区间 / 键名与 `EngineConfig::default()` / 调优默认值取同一组常量。 - /// 变异证明(只改实现侧):`EngineConfig::default()` 的并发写回 64、schema 默认值写成 16、 - /// `MAX_CONCURRENCY` 改了而 schema 仍写 64 ⇒ 各自红。 - #[test] - fn the_schema_contract_is_single_sourced() { - let schema = launcher_schema(); - assert_eq!(schema.version(), launcher_version()); - let cfg = EngineConfig::default(); - let tune = DownloadTuning::default(); - - // 并发 1~64(0 张许可没意义);尝试次数 **0**~10(P2-1:0 = 只试一次,旧配置里真有)。 - let contract: [(&str, i64, i64, usize); 2] = [ - ( - KEY_CONCURRENCY, - i64::try_from(DEFAULT_MAX_CONCURRENCY).expect("装得下"), - 1, - MAX_CONCURRENCY, - ), - ( - KEY_RETRY_LIMIT, - i64::from(DEFAULT_MAX_ATTEMPTS_PER_SOURCE), - 0, - usize::try_from(MAX_ATTEMPTS_PER_SOURCE).expect("装得下"), - ), - ]; - for (key, default, min, max) in contract { - let spec = schema.spec(key).expect("键已登记"); - assert_eq!(spec.default, serde_json::json!(default), "{key}"); - assert_eq!( - spec.kind, - crate::setting::spec::ValueKind::Int { - min, - max: i64::try_from(max).expect("装得下"), - }, - "{key}" - ); - } - assert_eq!(cfg.max_concurrency, DEFAULT_MAX_CONCURRENCY); - assert_eq!(tune.max_concurrency, DEFAULT_MAX_CONCURRENCY); - assert_eq!(cfg.max_attempts_per_source, DEFAULT_MAX_ATTEMPTS_PER_SOURCE); - assert_eq!( - tune.max_attempts_per_source, - DEFAULT_MAX_ATTEMPTS_PER_SOURCE - ); - } - - /// `SettingSpec.doc` 是 `&'static str`(`concat!` 插不进常量),描述里的数字只能手写——这条钉住它, - /// 改常量不改文案就红(`catalog.rs` 会把这些 doc 当 JSON schema 的 description 发出去)。 - #[test] - fn the_key_docs_quote_the_real_limits() { - let schema = launcher_schema(); - let doc = |key: &str| schema.spec(key).expect("键已登记").doc; - assert!(doc(KEY_CONCURRENCY).contains(&format!("1~{MAX_CONCURRENCY}"))); - assert!(doc(KEY_CONCURRENCY).contains(&format!("出厂 {DEFAULT_MAX_CONCURRENCY}"))); - assert!(doc(KEY_RETRY_LIMIT).contains(&format!("出厂 {DEFAULT_MAX_ATTEMPTS_PER_SOURCE}"))); - } - - /// 未填 / 越界回落默认(不阻断下载);**`retry_limit: 0` 是合法值**,语义 = 只试一次。 - /// - /// 变异证明:把并发的区间过滤删掉 ⇒ `0` / `9999` 两条红;把尝试次数的 `.max(1)` 删掉 ⇒ - /// `0 → 1` 那条红(引擎会拿到 0 次尝试);把 `0` 判成越界 ⇒ 同样红。 - #[test] - fn missing_or_illegal_values_fall_back_to_the_defaults() { - assert_eq!( - DownloadTuning::from_settings(None, None, None), - DownloadTuning::default() - ); - - let just_over = i64::try_from(MAX_CONCURRENCY).expect("上限装得下") + 1; - for bad in [0_i64, -1, 9_999, just_over] { - let tune = DownloadTuning::from_settings(Some(bad), None, None); - assert_eq!(tune.max_concurrency, DEFAULT_MAX_CONCURRENCY, "{bad}"); - } - // 0 与 1 都 = 只试一次:旧配置里的 0 不许回落成出厂 3(那正是「界面 0、引擎 3」)。 - for zero_or_one in [0_i64, 1] { - let tune = DownloadTuning::from_settings(None, Some(zero_or_one), None); - assert_eq!(tune.max_attempts_per_source, 1, "{zero_or_one}"); - } - for bad in [-1_i64, 9_999] { - let tune = DownloadTuning::from_settings(None, Some(bad), None); - assert_eq!( - tune.max_attempts_per_source, - DEFAULT_MAX_ATTEMPTS_PER_SOURCE - ); - } - - let tune = DownloadTuning::from_settings(Some(2), Some(2), None); - assert_eq!(tune.max_concurrency, 2); - assert_eq!(tune.max_attempts_per_source, 2); - } - - /// 三态:**空串 ≠ 关**(这条是「设置项说假话」的高发点,M197 的键语义同源)。 - #[test] - fn empty_gh_proxy_means_follow_builtin_not_off() { - assert_eq!(gh_proxy_override(""), GhProxyOverride::Builtin); - assert_eq!(gh_proxy_override(" "), GhProxyOverride::Builtin); - for off in [ - "off", "OFF", " none ", "direct", "disable", "disabled", "no", - ] { - assert_eq!(gh_proxy_override(off), GhProxyOverride::Off, "{off}"); - } - assert_eq!( - gh_proxy_override(" https://gh.example/ "), - GhProxyOverride::Prefix("https://gh.example/".to_string()) - ); - // 未填(键不存在)也是 Builtin,不是 Off。 - assert_eq!( - DownloadTuning::from_settings(None, None, None).gh_proxy, - GhProxyOverride::Builtin - ); - } - - /// `apply_gh_proxy` 的三档真的改到档案上、并且**改到候选链上**(`expand()` 读的就是档案里那个 - /// 前缀;自定义前缀那一档的链断言在 `shared.rs`)。变异证明:`Off` 分支改成原样返回 ⇒ `off` 红。 - #[test] - fn gh_proxy_override_reaches_the_profile_and_the_chain() { - use crate::download::mirrors::{MirrorTable, SOURCE_GH_PROXY}; - use crate::download::types::ResourceKind; - - let url = "https://github.com/o/r/releases/download/v1/a.jar"; - let table = MirrorTable::builtin(); - let sources = |profile: &RegionProfile| -> Vec { - table - .expand(url, ResourceKind::Other, profile) - .into_iter() - .map(|candidate| candidate.source) - .collect() - }; - - let mirror_first = RegionProfile::mirror_first(0); - let kept = DownloadTuning::default().apply_gh_proxy(mirror_first.clone()); - assert_eq!(kept.gh_proxy, mirror_first.gh_proxy, "Builtin 不该动档案"); - assert!( - sources(&kept).contains(&SOURCE_GH_PROXY.to_string()), - "出厂套前缀" - ); - - let off = DownloadTuning::from_settings(None, None, Some("off")) - .apply_gh_proxy(mirror_first.clone()); - assert!( - !sources(&off).contains(&SOURCE_GH_PROXY.to_string()), - "关了就没候选" - ); - - let prefix = DownloadTuning::from_settings(None, None, Some("https://gh.example/")) - .apply_gh_proxy(mirror_first); - assert_eq!(prefix.gh_proxy.as_deref(), Some("https://gh.example/")); - assert!(prefix.gh_proxy_enabled(), "前缀生效"); - } - - /// 每个数据根的调优互不串味(一个根改成 2 不该改另一个根的引擎)。 - /// 变异证明:把 `wanted`/`publish` 的键改回单一进程级槽位 ⇒ 第二个根读到 2,红。 - #[test] - fn tuning_is_per_cache_root_and_no_change_is_reported_as_such() { - let a = PathBuf::from("/tmp/ull-tuning-a"); - let b = PathBuf::from("/tmp/ull-tuning-b"); - assert_eq!(wanted(&a), DownloadTuning::default()); - - let changed = DownloadTuning::from_settings(Some(2), None, None); - assert!(publish(&a, changed.clone()), "真变了要报变更"); - assert_eq!(wanted(&a), changed); - assert!(!publish(&a, changed.clone()), "同值不该报变更"); - assert_eq!(wanted(&b), DownloadTuning::default(), "别的根不受影响"); - - assert!( - publish(&a, DownloadTuning::default()), - "恢复默认也是一次变更" - ); - assert_eq!(wanted(&a), DownloadTuning::default()); - } -} +#[path = "config/tests.rs"] +mod tests; diff --git a/crates/core/src/download/config/tests.rs b/crates/core/src/download/config/tests.rs new file mode 100644 index 0000000..1538681 --- /dev/null +++ b/crates/core/src/download/config/tests.rs @@ -0,0 +1,175 @@ +//! `download::config` 的单测:**从 `config.rs` 拆出只为守 400 行硬线**(M289 加超时字段前先纯搬移,逐字节未改)。 + +use super::*; +use crate::setting::keys::{launcher_schema, launcher_version}; + +/// **守门③**:schema 的默认值 / 区间 / 键名与 `EngineConfig::default()` / 调优默认值取同一组常量。 +/// 变异证明(只改实现侧):`EngineConfig::default()` 的并发写回 64、schema 默认值写成 16、 +/// `MAX_CONCURRENCY` 改了而 schema 仍写 64 ⇒ 各自红。 +#[test] +fn the_schema_contract_is_single_sourced() { + let schema = launcher_schema(); + assert_eq!(schema.version(), launcher_version()); + let cfg = EngineConfig::default(); + let tune = DownloadTuning::default(); + + // 并发 1~64(0 张许可没意义);尝试次数 **0**~10(P2-1:0 = 只试一次,旧配置里真有)。 + let contract: [(&str, i64, i64, usize); 2] = [ + ( + KEY_CONCURRENCY, + i64::try_from(DEFAULT_MAX_CONCURRENCY).expect("装得下"), + 1, + MAX_CONCURRENCY, + ), + ( + KEY_RETRY_LIMIT, + i64::from(DEFAULT_MAX_ATTEMPTS_PER_SOURCE), + 0, + usize::try_from(MAX_ATTEMPTS_PER_SOURCE).expect("装得下"), + ), + ]; + for (key, default, min, max) in contract { + let spec = schema.spec(key).expect("键已登记"); + assert_eq!(spec.default, serde_json::json!(default), "{key}"); + assert_eq!( + spec.kind, + crate::setting::spec::ValueKind::Int { + min, + max: i64::try_from(max).expect("装得下"), + }, + "{key}" + ); + } + assert_eq!(cfg.max_concurrency, DEFAULT_MAX_CONCURRENCY); + assert_eq!(tune.max_concurrency, DEFAULT_MAX_CONCURRENCY); + assert_eq!(cfg.max_attempts_per_source, DEFAULT_MAX_ATTEMPTS_PER_SOURCE); + assert_eq!( + tune.max_attempts_per_source, + DEFAULT_MAX_ATTEMPTS_PER_SOURCE + ); +} + +/// `SettingSpec.doc` 是 `&'static str`(`concat!` 插不进常量),描述里的数字只能手写——这条钉住它, +/// 改常量不改文案就红(`catalog.rs` 会把这些 doc 当 JSON schema 的 description 发出去)。 +#[test] +fn the_key_docs_quote_the_real_limits() { + let schema = launcher_schema(); + let doc = |key: &str| schema.spec(key).expect("键已登记").doc; + assert!(doc(KEY_CONCURRENCY).contains(&format!("1~{MAX_CONCURRENCY}"))); + assert!(doc(KEY_CONCURRENCY).contains(&format!("出厂 {DEFAULT_MAX_CONCURRENCY}"))); + assert!(doc(KEY_RETRY_LIMIT).contains(&format!("出厂 {DEFAULT_MAX_ATTEMPTS_PER_SOURCE}"))); +} + +/// 未填 / 越界回落默认(不阻断下载);**`retry_limit: 0` 是合法值**,语义 = 只试一次。 +/// +/// 变异证明:把并发的区间过滤删掉 ⇒ `0` / `9999` 两条红;把尝试次数的 `.max(1)` 删掉 ⇒ +/// `0 → 1` 那条红(引擎会拿到 0 次尝试);把 `0` 判成越界 ⇒ 同样红。 +#[test] +fn missing_or_illegal_values_fall_back_to_the_defaults() { + assert_eq!( + DownloadTuning::from_settings(None, None, None), + DownloadTuning::default() + ); + + let just_over = i64::try_from(MAX_CONCURRENCY).expect("上限装得下") + 1; + for bad in [0_i64, -1, 9_999, just_over] { + let tune = DownloadTuning::from_settings(Some(bad), None, None); + assert_eq!(tune.max_concurrency, DEFAULT_MAX_CONCURRENCY, "{bad}"); + } + // 0 与 1 都 = 只试一次:旧配置里的 0 不许回落成出厂 3(那正是「界面 0、引擎 3」)。 + for zero_or_one in [0_i64, 1] { + let tune = DownloadTuning::from_settings(None, Some(zero_or_one), None); + assert_eq!(tune.max_attempts_per_source, 1, "{zero_or_one}"); + } + for bad in [-1_i64, 9_999] { + let tune = DownloadTuning::from_settings(None, Some(bad), None); + assert_eq!( + tune.max_attempts_per_source, + DEFAULT_MAX_ATTEMPTS_PER_SOURCE + ); + } + + let tune = DownloadTuning::from_settings(Some(2), Some(2), None); + assert_eq!(tune.max_concurrency, 2); + assert_eq!(tune.max_attempts_per_source, 2); +} + +/// 三态:**空串 ≠ 关**(这条是「设置项说假话」的高发点,M197 的键语义同源)。 +#[test] +fn empty_gh_proxy_means_follow_builtin_not_off() { + assert_eq!(gh_proxy_override(""), GhProxyOverride::Builtin); + assert_eq!(gh_proxy_override(" "), GhProxyOverride::Builtin); + for off in [ + "off", "OFF", " none ", "direct", "disable", "disabled", "no", + ] { + assert_eq!(gh_proxy_override(off), GhProxyOverride::Off, "{off}"); + } + assert_eq!( + gh_proxy_override(" https://gh.example/ "), + GhProxyOverride::Prefix("https://gh.example/".to_string()) + ); + // 未填(键不存在)也是 Builtin,不是 Off。 + assert_eq!( + DownloadTuning::from_settings(None, None, None).gh_proxy, + GhProxyOverride::Builtin + ); +} + +/// `apply_gh_proxy` 的三档真的改到档案上、并且**改到候选链上**(`expand()` 读的就是档案里那个 +/// 前缀;自定义前缀那一档的链断言在 `shared.rs`)。变异证明:`Off` 分支改成原样返回 ⇒ `off` 红。 +#[test] +fn gh_proxy_override_reaches_the_profile_and_the_chain() { + use crate::download::mirrors::{MirrorTable, SOURCE_GH_PROXY}; + use crate::download::types::ResourceKind; + + let url = "https://github.com/o/r/releases/download/v1/a.jar"; + let table = MirrorTable::builtin(); + let sources = |profile: &RegionProfile| -> Vec { + table + .expand(url, ResourceKind::Other, profile) + .into_iter() + .map(|candidate| candidate.source) + .collect() + }; + + let mirror_first = RegionProfile::mirror_first(0); + let kept = DownloadTuning::default().apply_gh_proxy(mirror_first.clone()); + assert_eq!(kept.gh_proxy, mirror_first.gh_proxy, "Builtin 不该动档案"); + assert!( + sources(&kept).contains(&SOURCE_GH_PROXY.to_string()), + "出厂套前缀" + ); + + let off = + DownloadTuning::from_settings(None, None, Some("off")).apply_gh_proxy(mirror_first.clone()); + assert!( + !sources(&off).contains(&SOURCE_GH_PROXY.to_string()), + "关了就没候选" + ); + + let prefix = DownloadTuning::from_settings(None, None, Some("https://gh.example/")) + .apply_gh_proxy(mirror_first); + assert_eq!(prefix.gh_proxy.as_deref(), Some("https://gh.example/")); + assert!(prefix.gh_proxy_enabled(), "前缀生效"); +} + +/// 每个数据根的调优互不串味(一个根改成 2 不该改另一个根的引擎)。 +/// 变异证明:把 `wanted`/`publish` 的键改回单一进程级槽位 ⇒ 第二个根读到 2,红。 +#[test] +fn tuning_is_per_cache_root_and_no_change_is_reported_as_such() { + let a = PathBuf::from("/tmp/ull-tuning-a"); + let b = PathBuf::from("/tmp/ull-tuning-b"); + assert_eq!(wanted(&a), DownloadTuning::default()); + + let changed = DownloadTuning::from_settings(Some(2), None, None); + assert!(publish(&a, changed.clone()), "真变了要报变更"); + assert_eq!(wanted(&a), changed); + assert!(!publish(&a, changed.clone()), "同值不该报变更"); + assert_eq!(wanted(&b), DownloadTuning::default(), "别的根不受影响"); + + assert!( + publish(&a, DownloadTuning::default()), + "恢复默认也是一次变更" + ); + assert_eq!(wanted(&a), DownloadTuning::default()); +} diff --git a/crates/core/src/download/engine.rs b/crates/core/src/download/engine.rs index e782e78..02ae5eb 100644 --- a/crates/core/src/download/engine.rs +++ b/crates/core/src/download/engine.rs @@ -14,7 +14,6 @@ //! 中间文件是本会话命名的(`.ullpart--`),下次(甚至下次启动)由 `PartFile::for_dest` 认领并改名, //! 续传判定照旧(身份规则见 `atomic.rs` 模块头)。传输细节在 `transfer.rs`,错误分类在 `http.rs`,本文件只负责顺序与策略。 -use std::path::Path; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -30,11 +29,18 @@ use super::http; use super::mirrors::MirrorTable; use super::progress::{ProgressEmitter, ProgressEvent}; use super::region::RegionProfile; +use super::stall::{StallReport, StallStage}; +mod finish; + +pub(crate) use finish::millis; + use super::types::{ Candidate, DownloadId, DownloadOutcome, DownloadRequest, ResourceKind, ResumePolicy, SOURCE_OFFICIAL, }; +mod inspect; + /// 一次成功尝试的产物描述。 pub(crate) struct AttemptOk { /// 文件总字节数。 @@ -94,50 +100,6 @@ impl DownloadEngine { self } - /// 配置。 - #[must_use] - pub fn config(&self) -> &EngineConfig { - &self.cfg - } - - /// **真实并发闸门**此刻的许可数(探针,M193):读 [`Self::acquire`] 会去抢的那张 - /// `Semaphore`,**不是** `config().max_concurrency`。有下载在飞时偏小,空闲等于容量。 - #[must_use] - pub fn concurrency_probe(&self) -> usize { - self.sem.available_permits() - } - - /// HTTP 客户端(同一条连接池与超时配置):给「一锤子查询」用(如 Adoptium 版本查询), - /// 没必要为了一个 GET 再造一个 client。 - #[must_use] - pub fn client(&self) -> &reqwest::Client { - &self.http - } - - /// 当前地区档案(设置页展示)。 - #[must_use] - pub fn region(&self) -> &RegionProfile { - &self.region - } - - /// 镜像表(`game::java` 的 Oracle / MS / Zulu 分发表在这里查)。 - #[must_use] - pub fn mirrors(&self) -> &MirrorTable { - &self.mirrors - } - - /// 健康度注册表(诊断页)。 - #[must_use] - pub fn health(&self) -> &HealthRegistry { - &self.health - } - - /// 内容寻址缓存。 - #[must_use] - pub fn cache(&self) -> &ContentCache { - &self.cache - } - /// 候选规划:镜像展开 → 官方位置保持不变 → 其余按健康度排序。 #[must_use] pub fn plan(&self, req: &DownloadRequest, now_ms: u64) -> Vec { @@ -179,7 +141,7 @@ impl DownloadEngine { if let Some(integrity) = req.integrity.clone() { if let Some(cached) = self.cache.lookup(&integrity, now_ms()) { - let bytes = materialize(&cached, &req.dest).map_err(|err| { + let bytes = finish::materialize(&cached, &req.dest).map_err(|err| { DownloadError::new( DownloadErrorKind::Io, format!("缓存直取失败({}):{err}", req.dest.display()), @@ -213,10 +175,14 @@ impl DownloadEngine { )); } - let _permit = self.acquire().await?; + let _permit = self.acquire(req.expected_size).await?; 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; + // 整作业预算用完 ⇒ 不再试剩余候选源(见下面的 `job_budget_exhausted`)。 + let mut exhausted = false; for candidate in &plan { if let Some(from) = previous.take() { @@ -245,6 +211,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 { @@ -262,7 +229,13 @@ impl DownloadEngine { now_ms(), ); } - let retry = err.kind.is_retryable() && attempt + 1 < max_attempts; + // 整作业总期限用完是**链路级**判决(同一份字节、同一条链路): + // ① `err.is_retryable()` 让同源不再重试(每试一次都要再烧满一整个预算); + // ② 下面外循环的 `exhausted` 让它连剩余候选源都不试。 + // 复审 r1 p2-2 的算术:不这么做,188MB/20 KiB/s 的用户会从「1.6 小时 + // 慢慢下完」变成「每源 3 次 × 每个候选 × 100 min」的反复重下。 + exhausted = err.job_budget_exhausted(); + let retry = err.is_retryable() && attempt + 1 < max_attempts; let delay = self.next_delay(attempt); self.emit(ProgressEvent::SourceFailed { id, @@ -282,55 +255,26 @@ impl DownloadEngine { } } previous = Some(candidate.source.clone()); + if exhausted { + break; + } } // 全部候选失败:保留中间文件与元数据,下次(甚至下次启动)可续传。 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)) - } - - fn finish( - &self, - req: &DownloadRequest, - part: PartFile, - id: DownloadId, - candidate: &Candidate, - ok: AttemptOk, - started: Instant, - ) -> Result { - if let Some(integrity) = &req.integrity { - // 缓存写入失败不影响交付(缓存纯属加速)。 - let _ = self.cache.insert(part.path(), integrity, now_ms()); - let _ = self.cache.persist(); - } - part.commit(&req.dest).map_err(|err| { - DownloadError::new( - DownloadErrorKind::Io, - format!("原子落盘失败({}):{err}", req.dest.display()), - ) - })?; - let elapsed = started.elapsed(); - self.emit(ProgressEvent::Completed { - id, - bytes: ok.bytes, - from_cache: false, - source: candidate.source.clone(), - elapsed_ms: millis(elapsed), - }); - Ok(DownloadOutcome { - id, - path: req.dest.clone(), - bytes: ok.bytes, - from_cache: false, - source: candidate.source.clone(), - resumed_from: ok.resumed_from, - elapsed, - }) + .with_suppressed(suppressed) + .with_last_stall(last_stall)) } pub(crate) fn announce(&self, id: &DownloadId, req: &DownloadRequest, candidates: usize) { @@ -344,10 +288,41 @@ impl DownloadEngine { }); } - pub(crate) async fn acquire(&self) -> Result { - Arc::clone(&self.sem).acquire_owned().await.map_err(|_| { - DownloadError::new(DownloadErrorKind::Io, "并发闸门已关闭(引擎正在销毁)") - }) + /// 抢一张并发许可(**有界**:等许可也是一次 `await`)。 + /// + /// 为什么必须设界:M289 之前任何一处无界 `await` 都能让一个作业永久攥着许可,同缓存根 + /// 的其它下载就全堵在这行(复审 r1 p2-1 点名的连坐形态)。上界取 `permit_timeout` + /// (必须 ≥ 单个作业的总期限,见 `stall` 模块头),到点报阶段「等许可」—— + /// 不冒充「服务器不给我数据」。 + pub(crate) async fn acquire( + &self, + expected: Option, + ) -> Result { + match tokio::time::timeout( + self.cfg.permit_timeout, + Arc::clone(&self.sem).acquire_owned(), + ) + .await + { + Ok(Ok(permit)) => Ok(permit), + Ok(Err(_)) => Err(DownloadError::new( + DownloadErrorKind::Io, + "并发闸门已关闭(引擎正在销毁)", + )), + Err(_) => Err(DownloadError::stalled( + StallReport::deadline( + StallStage::AcquirePermit, + 0, + expected, + self.cfg.permit_timeout, + ), + format!( + "等并发许可超过 {}s(此刻可用 {} 张)", + self.cfg.permit_timeout.as_secs(), + self.sem.available_permits() + ), + )), + } } /// 该源此刻给几次尝试:冷却中 / 刚出冷却未洗白 → 1 次(那就是恢复探测,别让死源每次到期 @@ -378,23 +353,3 @@ impl DownloadEngine { self.progress.emit(event); } } - -/// 缓存命中时把文件变成工作副本(硬链接优先,跨卷退化为复制)。 -fn materialize(source: &Path, dest: &Path) -> std::io::Result { - if let Some(parent) = dest.parent() { - if !parent.as_os_str().is_empty() { - std::fs::create_dir_all(parent)?; - } - } - if dest.exists() { - let _ = std::fs::remove_file(dest); - } - if std::fs::hard_link(source, dest).is_err() { - std::fs::copy(source, dest)?; - } - Ok(std::fs::metadata(dest)?.len()) -} - -pub(crate) fn millis(duration: Duration) -> u64 { - u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) -} diff --git a/crates/core/src/download/engine/finish.rs b/crates/core/src/download/engine/finish.rs new file mode 100644 index 0000000..34491f3 --- /dev/null +++ b/crates/core/src/download/engine/finish.rs @@ -0,0 +1,76 @@ +//! 一次尝试成功后的收尾:缓存写入 → 原子落盘 → 完成事件;外加两个小工具。 +//! +//! 从 `engine.rs` 拆出只为守 400 行硬线(M289 r1 复审又给 engine 加了许可上界与 +//! 预算耗尽短路,贴线了)。搬出来的代码逐字节未改,只有可见性从私有放宽到 +//! `pub(super)` / `pub(crate)`(跨模块调用需要)。 + +use std::path::Path; +use std::time::{Duration, Instant}; + +use super::super::atomic::PartFile; +use super::super::cache::now_ms; +use super::super::error::{DownloadError, DownloadErrorKind}; +use super::super::progress::ProgressEvent; +use super::super::types::{Candidate, DownloadId, DownloadOutcome, DownloadRequest}; +use super::{AttemptOk, DownloadEngine}; + +impl DownloadEngine { + pub(super) fn finish( + &self, + req: &DownloadRequest, + part: PartFile, + id: DownloadId, + candidate: &Candidate, + ok: AttemptOk, + started: Instant, + ) -> Result { + if let Some(integrity) = &req.integrity { + // 缓存写入失败不影响交付(缓存纯属加速)。 + let _ = self.cache.insert(part.path(), integrity, now_ms()); + let _ = self.cache.persist(); + } + part.commit(&req.dest).map_err(|err| { + DownloadError::new( + DownloadErrorKind::Io, + format!("原子落盘失败({}):{err}", req.dest.display()), + ) + })?; + let elapsed = started.elapsed(); + self.emit(ProgressEvent::Completed { + id, + bytes: ok.bytes, + from_cache: false, + source: candidate.source.clone(), + elapsed_ms: millis(elapsed), + }); + Ok(DownloadOutcome { + id, + path: req.dest.clone(), + bytes: ok.bytes, + from_cache: false, + source: candidate.source.clone(), + resumed_from: ok.resumed_from, + elapsed, + }) + } +} + +/// 缓存命中时把文件变成工作副本(硬链接优先,跨卷退化为复制)。 +pub(super) fn materialize(source: &Path, dest: &Path) -> std::io::Result { + if let Some(parent) = dest.parent() { + if !parent.as_os_str().is_empty() { + std::fs::create_dir_all(parent)?; + } + } + if dest.exists() { + let _ = std::fs::remove_file(dest); + } + if std::fs::hard_link(source, dest).is_err() { + std::fs::copy(source, dest)?; + } + Ok(std::fs::metadata(dest)?.len()) +} + +pub(crate) fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} diff --git a/crates/core/src/download/engine/inspect.rs b/crates/core/src/download/engine/inspect.rs new file mode 100644 index 0000000..2cc9756 --- /dev/null +++ b/crates/core/src/download/engine/inspect.rs @@ -0,0 +1,57 @@ +//! 引擎的只读查询面:**从 `engine.rs` 拆出只为守 400 行硬线**(M289 加失速诊断前先纯搬移,逐字节未改)。 +//! +//! 这里只有「读」——配置、并发闸门探针、HTTP 客户端、地区档案、镜像表、健康度、内容缓存。 +//! 编排(候选循环、退避、落盘)仍在 [`super`]。 + +use super::super::cache::ContentCache; +use super::super::config::EngineConfig; +use super::super::health::HealthRegistry; +use super::super::mirrors::MirrorTable; +use super::super::region::RegionProfile; +use super::DownloadEngine; + +impl DownloadEngine { + /// 配置。 + #[must_use] + pub fn config(&self) -> &EngineConfig { + &self.cfg + } + + /// **真实并发闸门**此刻的许可数(探针,M193):读 [`Self::acquire`] 会去抢的那张 + /// `Semaphore`,**不是** `config().max_concurrency`。有下载在飞时偏小,空闲等于容量。 + #[must_use] + pub fn concurrency_probe(&self) -> usize { + self.sem.available_permits() + } + + /// HTTP 客户端(同一条连接池与超时配置):给「一锤子查询」用(如 Adoptium 版本查询), + /// 没必要为了一个 GET 再造一个 client。 + #[must_use] + pub fn client(&self) -> &reqwest::Client { + &self.http + } + + /// 当前地区档案(设置页展示)。 + #[must_use] + pub fn region(&self) -> &RegionProfile { + &self.region + } + + /// 镜像表(`game::java` 的 Oracle / MS / Zulu 分发表在这里查)。 + #[must_use] + pub fn mirrors(&self) -> &MirrorTable { + &self.mirrors + } + + /// 健康度注册表(诊断页)。 + #[must_use] + pub fn health(&self) -> &HealthRegistry { + &self.health + } + + /// 内容寻址缓存。 + #[must_use] + 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..8069940 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::{StallReason, StallReport, StallStage}; + /// 错误分类:决定「同源重试」「立刻换源」还是「整体失败」。 #[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 { @@ -137,6 +176,27 @@ impl DownloadError { } self.kind } + + /// **整作业总期限用完**:分块作业的平均吞吐掉到 [`super::stall::MIN_AVG_THROUGHPUT`] 以下。 + /// + /// 这是一个**链路级**判决(同一份字节、同一条链路),不是「这个源暂时不顺」。 + #[must_use] + pub fn job_budget_exhausted(&self) -> bool { + self.stall.as_deref().is_some_and(|stall| { + stall.stage == StallStage::AwaitChunkJoin && stall.reason == StallReason::Deadline + }) + } + + /// 这次失败值不值得再试一次。 + /// + /// 除了 `kind.is_retryable()`,**整作业总期限用完不再重试**:每个候选源(乃至同一源的 + /// 每一次重试)都要再烧满一整个预算,而它是链路级判决——复审 r1 p2-2 的算术:188MB 在 + /// 20 KiB/s 上要 160 min,预算 100.3 min,按默认「每源 3 次 × 多个候选」重试会让用户 + /// 从「1.6 小时慢慢下完」变成「几小时反复重下,最后全部源失败」。 + #[must_use] + pub fn is_retryable(&self) -> bool { + !self.job_budget_exhausted() && self.kind.is_retryable() + } } /// 从各候选源的失败里汇总出**最可诊断**的叶子成因。 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..dbcc218 --- /dev/null +++ b/crates/core/src/download/stall.rs @@ -0,0 +1,364 @@ +//! 失速诊断与下载的每一处上界(M289 / 台账 P19)。 +//! +//! # 事实:下载引擎曾是全仓唯一没有超时的那条路(r1 复审 p2-1 后修正) +//! +//! 「全仓下载路径只有两处无界 `await`」那句话**是错的**,被复审用一个 `silent()` 探针 +//! 当场打穿。真实的清单一共有**五处**,M289 修掉了其中五处: +//! +//! * 单连接请求的 `send()`(`transfer.rs`); +//! * 分块请求的 `send()`(`chunked.rs`); +//! * `download_bytes` 的 `send()`(`bytes.rs`,生产路径:版本清单 / 版本 JSON / +//! Modrinth 元数据)——整包读 `response.bytes()` 同样无界,而且它**只管大小不管时间** +//! (`MAX_IN_MEMORY_BYTES` 不是时间上界); +//! * `acquire_owned()`(`engine.rs`):等许可也是一次 `await`,卡死的作业攥着许可, +//! 同缓存根的 `download()` 就全堵在这里(P19 (c′) 的同族形态)。 +//! +//! `build_client` 只设了 `connect_timeout`,reqwest 自己的 `timeout` 默认 `None` —— +//! 于是「服务器接了连接、不回一个字节」= 这个 future 永不结束,而下 188MB 的作业 +//! 就在那里静静挂着(`ull java install` 走的是 `silent()` 引擎:零事件、零日志, +//! 用户只能看到一个不动的主界面)。 +//! +//! # 哪些入口有界、谁托管给谁 +//! +//! | 入口 | 上界 | 阶段 | +//! |------|------|------| +//! | `DownloadEngine::download` 单连接 | `http::send_bounded` + `http::read_step` | 等响应头 / 等字节 | +//! | `DownloadEngine::download` 分块 | 同上 + [`StallStage::AwaitChunkJoin`](无进度窗口与总期限) | 等分块回收 | +//! | `DownloadEngine::download_bytes` | `http::send_bounded` + `read_step` 流式累加(顺带用 `MAX_IN_MEMORY_BYTES` 卡住无 `Content-Length` 的洪流) | 等响应头 / 等字节 | +//! | `DownloadEngine::acquire`(许可) | `EngineConfig::permit_timeout` | [`StallStage::AcquirePermit`] | +//! | 校验(`verify_part`) | `EngineConfig::verify_timeout` | [`StallStage::Verify`] | +//! | `font::fetch_blocking` | 自带的整作业预算(外加上面全部) | —— | +//! | `region::probe_reachable` | 自带的 3s | —— | +//! +//! 因此**本模块内**裸 `.send()` 只剩两处:本模块的 [`super::http::send_bounded`](唯一构造点) +//! 与 `region.rs` 的 3s 探测(自带界)。新加下载入口时**先看这张表**,别再出现第二个 +//! `download_bytes`。 +//! +//! 模块外还有几处用**同一个** `reqwest::Client`(只设了 `connect_timeout`)的地方,M289 没碰 +//! (scope 是 `download/**`):`auth/http.rs` 自带 `.timeout(REQUEST_TIMEOUT)`;而 +//! `game/java/api.rs`、`game/java/net.rs`、`update/github.rs` 是裸 `send()`——**未审**, +//! 与这里修掉的是同族形态,已用 TowerFinding 报给塔台另立 mission。 +//! +//! # 取舍:三段可以共存的界,**不是**「整作业必须在 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`]),不是拍一个固定秒数。 +//! +//! **不要**把它改成 `reqwest::ClientBuilder::timeout(...)`:reqwest 的 `timeout` 覆盖 +//! **整个响应体**读取,那正好就是「慢但有效被判死」的那个坑(它的默认 `None` 也是 P19 +//! 记的另一半事实)。每个阶段各自的界才分得清「连不上」「不回话」「不回字节」。 +//! +//! # 「慢但有效」的适用边界(r1 复审 p2-2 要求写明) +//! +//! 第 3 条那句「慢链路照样活」有一个**前提**,等价说法是:**平均吞吐 ≥ 32 KiB/s 才活得下来**。 +//! 算术:188MB 在 20 KiB/s 上要 160.4 min,而预算只有 100.3 min ⇒ 必被杀。而且一旦被杀, +//! `DownloadError::job_budget_exhausted()` 让引擎**不再重试、也不再换源**(同一份字节、 +//! 同一条链路,候选源不可能更快)——所以在 <32 KiB/s 的链路上,这个文件大小是「下不完」, +//! 不是「慢一点」。要让用户自己调这几个数,得接 `download.*` 的 schema(`setting/keys.rs` +//! 不在 M289 的 scope,是另一个 mission);在那之前出厂值就是唯一值。 +//! +//! 这条边界的命中面不远:仓内就有 ≈67 KB/s 的真机观测(`font/tests/support.rs` 的文件头)。 +//! 真实公网慢链路的取值**没有真机验证**(M289 报告列为未验项)。 +//! +//! # 数字从哪来(全仓只写这一处) +//! +//! * `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 才活得下来」 +//! 的等价说法**,不是「慢一点的链路也没事」。 +//! * `permit_timeout` = 120min:等许可的上界。它必须 **≥ 单个作业的总期限** +//! (最坏 ≈ 100 min 预算 + 120s 校验),否则 `download.concurrency = 1` 时正常排队 +//! 会被误杀;它挡的是「许可被永久占住」(M289 之前每一处无界 `await` 都能造出来)。 +//! * `verify_timeout` = 120s:188MB 的 SHA-1 在正常盘上不到 1s,120s 只抓「整文件重算 +//! 摘要被卡住」(网络盘掉线、杀软锁文件),不参与慢链路判定。 +//! +//! # 口径(读这两个数之前先看这里) +//! +//! `received` / `expected` 是**该阶段自己的口径**:等字节 = 这条连接收到多少 / 该连接 +//! 被要求收多少(分块就是那一段的长度);等分块回收 = 整个作业收了多少 / 作业总长; +//! 等许可 = 都还没有(`received = 0`、`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, + /// 还没开始下载:在等引擎的并发许可(`Semaphore` 队列)。 + /// + /// 单独一档而不是塞进 `AwaitBytes`:这里上游根本还没轮到我们,「服务器不给数据」与 + /// 「引擎不给我们许可」是两种完全不同的故障(复审 r1 p2-1:挂死的作业攥着许可会把 + /// 同缓存根的 `download()` 全堵死)。 + AcquirePermit, + /// 分块作业在等分块任务回收(整作业的无进度 / 总期限上界)。 + 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::AcquirePermit => "acquire_permit", + 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::AcquirePermit => "等许可", + 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::AcquirePermit, "acquire_permit", "等许可"), + (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..e3d0a4a 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, @@ -115,6 +116,8 @@ impl DownloadEngine { cut: ResumeCut, ) -> Result { let integrity = req.integrity.as_ref(); + // 两段界都从这一处取(`EngineConfig::deadlines`):别在这里各读各的字段。 + let deadlines = self.cfg.deadlines(); let mut offset = cut.offset; let mut resumed_from = cut.resumed_from; let mut request = self.http.get(&candidate.url); @@ -133,15 +136,16 @@ 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, 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 +230,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), + deadlines.read_idle, + 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..11c1ee3 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,48 @@ 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 池并套上界,否则「校验阶段卡住」 + // (网络盘掉线、杀软锁文件)照样是永久挂住,且它发生在字节全部收完之后, + // 从进度上看与「下载完成」一模一样。 + // + // 代价(复审 r1 p3-1,未验):`spawn_blocking` 的线程**不可取消**——到点后它还在 + // 读文件、还持着句柄,重试会再堆一个(每源最多 3 个)。**假设未验**:Windows 上 + // 那个句柄可能让随后的同名重建(`attempt_chunked` 的 `OpenOptions::create(true)`) + // 拿到 `Access is denied`,表现为「预分配中间文件失败」而不是干净重试 + // (`PartFile::discard` 忽略错误)。仍然有界、有分类,只是诊断会拐一道。 + 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..c77598f --- /dev/null +++ b/crates/core/tests/download/chunked/timeout.rs @@ -0,0 +1,237 @@ +//! 分块作业的两道失速判据(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。 +/// 另外钉住 r1 p2-2 的第一条要求:预算用完**不重试**(`max_attempts_per_source = 3` 之下 +/// 仍然只发一轮分块请求,`hits` 停在 4)。变异:让 `DownloadError::is_retryable()` 忽略 +/// `job_budget_exhausted()` ⇒ `hits` 变成 8 / 12,本条红。 +#[tokio::test] +async fn a_chunked_job_that_is_too_slow_hits_the_total_budget_without_retrying() { + 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..12).map(dripping).collect()).await; + let (engine, dir) = harness( + "chunked-timeout-budget", + EngineConfig { + chunk_job_floor: Duration::from_millis(300), + max_attempts_per_source: 3, + ..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}"); + assert!( + !error.is_retryable(), + "整作业总期限用完是链路级判决,不许重试:{error}" + ); + assert_eq!( + server.hits(), + 4, + "预算用完只许发一轮分块请求(重试会再烧满一整个预算)" + ); + server.stop(); +} + +/// 预算耗尽**连候选源都不再换**(r1 p2-2 的「必然失败的形态」):镜像慢到撞预算, +/// 官方源一次都不该被碰——同一份字节、同一条链路,换源只会再烧一轮。 +/// +/// 变异:去掉 `engine.rs` 里的 `if exhausted { break; }` ⇒ `official.hits()` 变成 1,红。 +#[tokio::test] +async fn a_budget_exhaustion_does_not_try_the_next_source() { + use unlucky_core::download::mirrors::PrefixRule; + + let payload = vec![8_u8; 16 * 1024]; + let size = payload.len() as u64; + let official = MockServer::start(vec![MockResponse::ok(payload.clone())]).await; + let mirror = MockServer::start( + (0..12) + .map(|_| MockResponse::ok(payload.clone()).dripping(1, Duration::from_millis(20))) + .collect(), + ) + .await; + let mut table = MirrorTable::builtin(); + table.add_custom_rule(PrefixRule::new( + "mock-mirror", + format!("http://127.0.0.1:{}", official.port()), + format!("http://localhost:{}", mirror.port()), + )); + let dir = temp_dir("chunked-timeout-no-failover"); + let engine = DownloadEngine::new( + EngineConfig { + chunk_job_floor: Duration::from_millis(300), + max_attempts_per_source: 3, + ..config() + }, + Arc::new(table), + Arc::new(RegionProfile::mirror_first(0)), + Arc::new(HealthRegistry::in_memory()), + Arc::new(ContentCache::open(dir.join("cache"))), + ); + let url = official.url("/jdk.zip"); + + let error = tokio::time::timeout( + GATE, + engine.download(request(&url, dir.join("jdk.zip"), size)), + ) + .await + .expect("预算耗尽必须有界返回") + .expect_err("慢到没意义的镜像必须让整次下载失败"); + assert_eq!( + official.hits(), + 0, + "预算用完不该再换源(同一链路再烧一轮 100 分钟没有意义)" + ); + assert!(error.job_budget_exhausted(), "{error}"); + official.stop(); + mirror.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/bytes_timeout.rs b/crates/core/tests/download/engine_mock/bytes_timeout.rs new file mode 100644 index 0000000..8297d89 --- /dev/null +++ b/crates/core/tests/download/engine_mock/bytes_timeout.rs @@ -0,0 +1,97 @@ +//! `download_bytes` 的两处 `await` 也必须自己超时(M289 r1 复审 p2-1)。 +//! +//! 复审用一个 `silent()` 探针证明了同一个模块里**另一条下载入口**一模一样地还在挂: +//! `download_bytes` 的 `.send()` 是裸的、`response.bytes()` 是整包读(`MAX_IN_MEMORY_BYTES` +//! 只管大小不管时间)。生产调用点三个:`game/install/source.rs`(版本清单 / 版本 JSON)、 +//! `addon/online/transport.rs`(Modrinth 元数据)。 +//! +//! 现在两条都走与文件下载同一套构造点:`http::send_bounded` + `http::read_step` +//! (整包读换成流式累加,顺带卡住没有 `Content-Length` 的洪流)。 +//! +//! 变异:把 `fetch_bytes` 改回裸 `send()` ⇒ 第一条在外层 5s 处挂断而红; +//! 把流式读换回 `response.bytes()` ⇒ 第二条在外层 5s 处挂断而红。 + +use std::time::{Duration, Instant}; + +use unlucky_core::download::error::DownloadErrorKind; +use unlucky_core::download::stall::{StallReason, StallStage}; +use unlucky_core::download::types::{DownloadRequest, ResourceKind}; + +use super::harness::{config, engine, silent_responses, GATE}; +use super::support::{MockResponse, MockServer}; + +fn request(url: &str, dest: std::path::PathBuf, size: Option) -> DownloadRequest { + let request = DownloadRequest::new(ResourceKind::VersionManifest, url, &dest); + match size { + Some(size) => request.with_size(size), + None => request, + } +} + +/// 服务端接了连接、一个字都不回:`download_bytes` 必须自己在 2s 内返回,阶段=等响应头。 +#[tokio::test] +async fn download_bytes_against_a_silent_server_is_bounded_and_named() { + let server = MockServer::start(silent_responses(2)).await; + let (engine, dir) = engine(config(200, 200), "timeout-bytes-silent"); + let url = server.url("/manifest.json"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download_bytes(request(&url, dir.join("manifest.json"), Some(4096))), + ) + .await + .expect("download_bytes 的 send() 又无界了:外层 5s 挂断(与 `download()` 是同一类 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("`download_bytes` 的整体失败也要带最后一次失速现场"); + 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(4096), "{stall}"); + assert!( + error.to_string().contains("等响应头"), + "用户那行字要说清卡在哪:{error}" + ); + server.stop(); +} + +/// 头到手、正文写出一截就卡住:整包读也必须**空闲**有界(阶段=等字节),不是等它读完。 +#[tokio::test] +async fn download_bytes_body_read_is_idle_bounded() { + let payload = vec![6_u8; 32 * 1024]; + let server = MockServer::start(vec![MockResponse::ok(payload).stalling_after(16)]).await; + let (engine, dir) = engine(config(2_000, 300), "timeout-bytes-body"); + let url = server.url("/version.json"); + + let started = Instant::now(); + let error = tokio::time::timeout( + GATE, + engine.download_bytes(request(&url, dir.join("version.json"), None)), + ) + .await + .expect("正文读取又无界了:外层 5s 挂断(`response.bytes()` 是整包读,没有空闲判定)") + .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, 16, "已收字节是真实读到量:{stall}"); + // 夹具声明了 `Content-Length`(`ok()` 会带),所以预期字节是有的; + // 没有 `Content-Length` 的洪流由「边读边卡 MAX_IN_MEMORY_BYTES」兜(本文件不重复)。 + assert_eq!(stall.expected, Some(32 * 1024), "{stall}"); +} diff --git a/crates/core/tests/download/engine_mock/harness.rs b/crates/core/tests/download/engine_mock/harness.rs new file mode 100644 index 0000000..4376696 --- /dev/null +++ b/crates/core/tests/download/engine_mock/harness.rs @@ -0,0 +1,50 @@ +//! `engine_mock/timeout*` 两条用例文件共用的夹具(引擎构造 / 外层保险 / 静默响应脚本)。 +//! +//! 拆出来是因为 `bytes_timeout.rs` 与 `timeout.rs` 是兄弟模块,各写一份会变成「两份默认值」。 + +use std::sync::Arc; +use std::time::Duration; + +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::health::HealthRegistry; +use unlucky_core::download::mirrors::MirrorTable; +use unlucky_core::download::region::RegionProfile; + +use super::support::{temp_dir, MockResponse}; + +/// 外层保险:引擎若真的无界,用例在这里挂断并给出「这就是 bug」的说明,而不是跑不完。 +pub const GATE: Duration = Duration::from_secs(5); + +/// 只为超时用例准备的引擎(单连接路径;分块用例自己拼配置)。 +pub fn config(response_ms: u64, read_ms: u64) -> EngineConfig { + EngineConfig { + max_attempts_per_source: 1, + 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() + } +} + +pub fn engine(cfg: EngineConfig, tag: &str) -> (DownloadEngine, std::path::PathBuf) { + let dir = temp_dir(tag); + let engine = DownloadEngine::new( + cfg, + Arc::new(MirrorTable::builtin()), + Arc::new(RegionProfile::direct(0)), + Arc::new(HealthRegistry::in_memory()), + Arc::new(ContentCache::open(dir.join("cache"))), + ); + (engine, dir) +} + +/// `silent()` 服务的脚本(`count` 份:多候选/多重试时不会被脚本耗尽改成 500)。 +pub fn silent_responses(count: usize) -> Vec { + (0..count).map(|_| MockResponse::silent()).collect() +} 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..5db160d --- /dev/null +++ b/crates/core/tests/download/engine_mock/timeout.rs @@ -0,0 +1,330 @@ +//! 超时的**可诊断性**与**不误杀**(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::time::{Duration, Instant}; + +use unlucky_core::download::config::EngineConfig; +use unlucky_core::download::error::DownloadErrorKind; +use unlucky_core::download::stall::{StallReason, StallStage}; +use unlucky_core::download::types::{DownloadRequest, ResourceKind}; + +use super::harness::{config, engine, silent_responses, GATE}; +use super::support::{sha1_hex, MockResponse, MockServer}; + +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(); +} + +/// 等许可是第五处 `await`(r1 p2-1 的连坐形态):许可被占住时必须有界,且阶段是「等许可」。 +/// +/// `max_concurrency = 1` 让第一条下载攥住唯一许可;第二条堵在 `acquire()` 上。 +/// 变异:把 `acquire()` 的 `timeout` 去掉 ⇒ 第二条在外层 `GATE` 处挂断而红。 +#[tokio::test] +async fn waiting_for_a_permit_is_bounded_and_named_acquire_permit() { + let held = vec![4_u8; 8 * 1024]; + let holder = MockServer::start(vec![MockResponse::ok(held.clone()).stalling_after(0)]).await; + let good = MockServer::start(vec![MockResponse::ok(b"second".to_vec())]).await; + let (engine, dir) = engine( + EngineConfig { + max_concurrency: 1, + permit_timeout: Duration::from_millis(100), + ..config(2_000, 300) + }, + "timeout-permit-wait", + ); + + let (first, second) = tokio::time::timeout(GATE, async { + tokio::join!( + engine.download(request( + &holder.url("/hold.jar"), + dir.join("hold.jar"), + None + )), + engine.download(request(&good.url("/ok.jar"), dir.join("ok.jar"), None)), + ) + }) + .await + .expect("等许可仍然无界:两条 select 在外层 5s 处挂断"); + + assert!(first.is_err(), "第一条自己会读超时(占着许可 300ms)"); + let error = second.expect_err("许可被占住时第二条必须失败"); + let stall = error.stall.as_deref().expect("等许可失败要带现场"); + assert_eq!(stall.stage, StallStage::AcquirePermit, "{stall}"); + assert_eq!(stall.reason, StallReason::Deadline, "{stall}"); + assert!( + stall.idle >= Duration::from_millis(100), + "无进展时长不得小于上界:{stall}" + ); + assert!( + error.to_string().contains("等许可"), + "用户那行字要说清卡在哪(不许冒充「服务器不给数据」):{error}" + ); + assert_eq!(good.hits(), 0, "没拿到许可就不该发请求"); + holder.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..b15bdbd 100644 --- a/crates/core/tests/download/engine_mock_test.rs +++ b/crates/core/tests/download/engine_mock_test.rs @@ -362,3 +362,15 @@ 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; + +/// `download_bytes` 那条入口的两处 await(M289 r1 p2-1):同族但独立的用例文件。 +#[path = "engine_mock/bytes_timeout.rs"] +mod bytes_timeout; + +/// 上面两条共用的夹具(引擎构造 / 外层保险 / 静默脚本)。 +#[path = "engine_mock/harness.rs"] +mod harness; 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('-')?;