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/config.rs b/crates/core/src/download/config.rs index 94232d8..0f67ccb 100644 --- a/crates/core/src/download/config.rs +++ b/crates/core/src/download/config.rs @@ -67,6 +67,11 @@ pub struct EngineConfig { /// /// 正常盘上 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。 @@ -90,6 +95,7 @@ impl Default for EngineConfig { 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(), diff --git a/crates/core/src/download/engine.rs b/crates/core/src/download/engine.rs index e3fe3b5..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,7 +29,11 @@ use super::http; use super::mirrors::MirrorTable; use super::progress::{ProgressEmitter, ProgressEvent}; use super::region::RegionProfile; -use super::stall::StallReport; +use super::stall::{StallReport, StallStage}; +mod finish; + +pub(crate) use finish::millis; + use super::types::{ Candidate, DownloadId, DownloadOutcome, DownloadRequest, ResourceKind, ResumePolicy, SOURCE_OFFICIAL, @@ -138,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()), @@ -172,12 +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() { @@ -224,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, @@ -244,6 +255,9 @@ impl DownloadEngine { } } previous = Some(candidate.source.clone()); + if exhausted { + break; + } } // 全部候选失败:保留中间文件与元数据,下次(甚至下次启动)可续传。 @@ -263,45 +277,6 @@ impl DownloadEngine { .with_last_stall(last_stall)) } - 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(crate) fn announce(&self, id: &DownloadId, req: &DownloadRequest, candidates: usize) { self.emit(ProgressEvent::Started { id: *id, @@ -313,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 次(那就是恢复探测,别让死源每次到期 @@ -347,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/error.rs b/crates/core/src/download/error.rs index afc8cf1..8069940 100644 --- a/crates/core/src/download/error.rs +++ b/crates/core/src/download/error.rs @@ -6,7 +6,7 @@ use std::fmt; use serde::{Deserialize, Serialize}; -use super::stall::StallReport; +use super::stall::{StallReason, StallReport, StallStage}; /// 错误分类:决定「同源重试」「立刻换源」还是「整体失败」。 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] @@ -176,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/stall.rs b/crates/core/src/download/stall.rs index 0334d8c..dbcc218 100644 --- a/crates/core/src/download/stall.rs +++ b/crates/core/src/download/stall.rs @@ -1,15 +1,43 @@ -//! 失速诊断与下载的三段上界(M289 / 台账 P19)。 +//! 失速诊断与下载的每一处上界(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′) 的同族形态)。 //! -//! 全仓下载路径此前只有两处**无界** `await`:分块请求与单连接请求的 `send()`。 //! `build_client` 只设了 `connect_timeout`,reqwest 自己的 `timeout` 默认 `None` —— //! 于是「服务器接了连接、不回一个字节」= 这个 future 永不结束,而下 188MB 的作业 //! 就在那里静静挂着(`ull java install` 走的是 `silent()` 引擎:零事件、零日志, //! 用户只能看到一个不动的主界面)。 //! -//! 同仓两个先例:`region.rs` 给探测的 `send()` 包了 3s,`font.rs` 给整个 -//! `download_bytes` 包了总预算——下载引擎自己反而是唯一没有界的那条路。 +//! # 哪些入口有界、谁托管给谁 +//! +//! | 入口 | 上界 | 阶段 | +//! |------|------|------| +//! | `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 秒内下完」 //! @@ -34,12 +62,24 @@ //! 字节一直在动就一直续期。每秒只挤几 KB 的链路也算「在动」,只有**真的不动**了才判失速; //! 3. [`StallStage::AwaitChunkJoin`](分块作业):一个分块卡住不许 4 条连接陪着挂到天荒地老。 //! 这里同时给「无进度窗口」与「总期限」,但总期限按**平均吞吐下限**反推 -//! ([`job_budget`]),不是拍一个固定秒数:能把 188MB 在期限内下完的慢链路照样活。 +//! ([`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 是**连接 @@ -49,16 +89,19 @@ //! * `chunk_stall_timeout` = 120s:4 条连接**整体**一个新字节都没有两分钟,只可能是 //! 分块任务卡住(写盘卡在杀软/网络盘上时,读超时不会响,只有这个会)。 //! * [`MIN_AVG_THROUGHPUT`] = 32 KiB/s:分块作业的总期限 = 总字节 / 32 KiB/s -//! (188MB ≈ 100 分钟)。即「整作业的平均速度掉到 32 KiB/s 以下才判死」;比这更快的 -//! 慢链路仍会下完。**这是平均吞吐门槛,不是慢链路的死刑**:真实公网慢链路的取值 -//! 没有真机验证过(M289 报告里列为未验项)。 +//! (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; @@ -75,6 +118,12 @@ pub enum StallStage { AwaitHeaders, /// 响应头已到,等下一块正文字节(读空闲判定的上界)。 AwaitBytes, + /// 还没开始下载:在等引擎的并发许可(`Semaphore` 队列)。 + /// + /// 单独一档而不是塞进 `AwaitBytes`:这里上游根本还没轮到我们,「服务器不给数据」与 + /// 「引擎不给我们许可」是两种完全不同的故障(复审 r1 p2-1:挂死的作业攥着许可会把 + /// 同缓存根的 `download()` 全堵死)。 + AcquirePermit, /// 分块作业在等分块任务回收(整作业的无进度 / 总期限上界)。 AwaitChunkJoin, /// 整文件重算摘要(校验阶段)。 @@ -89,6 +138,7 @@ impl StallStage { Self::Connect => "connect", Self::AwaitHeaders => "await_headers", Self::AwaitBytes => "await_bytes", + Self::AcquirePermit => "acquire_permit", Self::AwaitChunkJoin => "await_chunk_join", Self::Verify => "verify", } @@ -101,6 +151,7 @@ impl StallStage { Self::Connect => "建立连接", Self::AwaitHeaders => "等响应头", Self::AwaitBytes => "等字节", + Self::AcquirePermit => "等许可", Self::AwaitChunkJoin => "等分块回收", Self::Verify => "校验", } @@ -293,6 +344,7 @@ mod tests { (StallStage::Connect, "connect", "建立连接"), (StallStage::AwaitHeaders, "await_headers", "等响应头"), (StallStage::AwaitBytes, "await_bytes", "等字节"), + (StallStage::AcquirePermit, "acquire_permit", "等许可"), (StallStage::AwaitChunkJoin, "await_chunk_join", "等分块回收"), (StallStage::Verify, "verify", "校验"), ]; diff --git a/crates/core/src/download/transfer.rs b/crates/core/src/download/transfer.rs index c1b7aeb..e3d0a4a 100644 --- a/crates/core/src/download/transfer.rs +++ b/crates/core/src/download/transfer.rs @@ -116,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); @@ -134,14 +136,9 @@ impl DownloadEngine { } } - 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 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 { @@ -235,7 +232,7 @@ impl DownloadEngine { loop { let next = http::read_step( futures_util::StreamExt::next(&mut stream), - self.cfg.read_timeout, + deadlines.read_idle, written, total, &candidate.url, diff --git a/crates/core/src/download/transfer_chunk.rs b/crates/core/src/download/transfer_chunk.rs index bd13a29..11c1ee3 100644 --- a/crates/core/src/download/transfer_chunk.rs +++ b/crates/core/src/download/transfer_chunk.rs @@ -215,6 +215,12 @@ impl DownloadEngine { // 整文件重算是同步阻塞 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; diff --git a/crates/core/tests/download/chunked/timeout.rs b/crates/core/tests/download/chunked/timeout.rs index bf780e1..c77598f 100644 --- a/crates/core/tests/download/chunked/timeout.rs +++ b/crates/core/tests/download/chunked/timeout.rs @@ -124,17 +124,21 @@ async fn a_chunked_job_with_no_progress_is_named_await_chunk_join() { /// 分块作业的**总期限**:一直在动,但平均吞吐低到没意义 ⇒ 「等分块回收 + 期限用完」。 /// /// 与上一条的区别只有 `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() { +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..4).map(dripping).collect()).await; + 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() }, ); @@ -147,7 +151,7 @@ async fn a_chunked_job_that_is_too_slow_hits_the_total_budget() { ) .await .expect("分块作业总期限必须有界(外层挂断 = 无界)") - .expect_err("慢到没有意义的作业必须在总期限处失败"); + .expect_err("慢到没意义的作业必须在总期限处失败"); let elapsed = started.elapsed(); assert!( @@ -166,5 +170,68 @@ async fn a_chunked_job_that_is_too_slow_hits_the_total_budget() { "这条一直在动,已收字节必须 > 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/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 index ccb63b6..5db160d 100644 --- a/crates/core/tests/download/engine_mock/timeout.rs +++ b/crates/core/tests/download/engine_mock/timeout.rs @@ -17,53 +17,15 @@ //! 拆成子模块:`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() -} +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); @@ -317,3 +279,52 @@ async fn after_a_timeout_the_engine_still_downloads_again() { 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 eac51a3..b15bdbd 100644 --- a/crates/core/tests/download/engine_mock_test.rs +++ b/crates/core/tests/download/engine_mock_test.rs @@ -366,3 +366,11 @@ 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;