feat(download): 给每条下载 await 上界并在超时时报出「卡在哪一段」（M289 / P19）

全仓下载路径此前只有两处无界 await（分块与单连接的 `send()`），reqwest 自己的 `timeout` 又是 `None`：服务器接了连接不回一个字节 ⇒ future 永不结束， `ull java install` 那条 `silent()` 路径零事件零日志，用户只看到「卡住不动」。上界选「单次请求 + 读空闲」，**不是**「整作业总预算」： * `region.rs` 的 3s 是探测极小响应（整作业 = 一次 GET，两者同一件事）； * `font.rs` 的 ATTEMPT/TOTAL 是几百 KB 的字体（整作业几秒就完）； * 188MB 的 JDK 两者都不能照抄，所以分三段互不顶替的界，理由与取数**只写在** `crates/core/src/download/stall.rs` 模块头一处（含「别改成 reqwest 的 timeout，它覆盖整个响应体读取」）。改动： * 新增 `download::stall`：`StallStage`（连接 / 等响应头 / 等字节 / 等分块回收 / 校验）+ `StallReason`（期限用完 / 没有新字节）+ `StallReport`（阶段 + 已收 / 预期字节 + 无进展时长）+ `job_budget`（总期限按 32 KiB/s 平均吞吐反推）。 * `http::send_bounded` / `http::read_step`：两处 `send()` 与两处读循环的**唯一**构造点；超时错误带阶段与三个数，`message` 里也抄一份（整体失败时 `suppressed` 只搬 message）。 * `transfer_chunk::await_chunks`：`JoinSet::join_next` 加总期限与无进度窗口，两者靠 `StallReason` 分开报；窗口内有新字节就续期（慢但有效不误杀）。 * `verify_part` 挪到 blocking 池并套上界：校验阶段卡住也报得出来（阶段=校验）。 * `engine.rs`：整体失败时把**最后一次**尝试的失速现场带上去，并写进用户可见的那行字。 * 配置：`response_timeout` 60s / `chunk_stall_timeout` 120s / `chunk_job_floor` 10min / `verify_timeout` 120s（`EngineConfig`，未动设置 schema）。测试（全离线、`127.0.0.1`、<2s）：夹具新增 `silent()` / `stalling_after` / `short_body` / `oversize_body` / `dripping`；用例覆盖静默服务端、中途卡住、短包、超长包、慢但持续（4 倍于上界也要成功）、超时后引擎仍能下载（RAII 许可自还，不写手动归还）、以及分块作业的「无进度 vs 总期限」两条。另：`config/tests.rs` 的一处表达式在去缩进后行宽变化，随本 commit 由 rustfmt 合成一行（纯搬移 commit 提交时它仍是逐字节原文）。
2026-11-12 23:24:19 +08:00
1 parent 61c8a4eff9
commit cc0def3415
@@ -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<DownloadError> = 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<Result<u64, DownloadError>>,
+        counter: &AtomicU64,
+        total: u64,
+        url: &str,
+    ) -> Option<DownloadError> {
+        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<u64>,
        id: DownloadId,
    ) -> Result<(), DownloadError> {
        let Some(integrity) = integrity else {
@@ -145,15 +212,42 @@ impl DownloadEngine {
            algo: integrity.algo,
            expected_prefix: integrity.short_prefix().to_string(),
        });
-        let actual = verify::digest_file(part.path(), integrity.algo)
-            .map_err(|err| io_err("读取中间文件失败", &err))?;
-        if verify::matches(&integrity.hex, &actual) {
+        // 整文件重算是同步阻塞 IO：挪到 blocking 池并套上界，否则「校验阶段卡住」
+        // （网络盘掉线、杀软锁文件）照样是永久挂住，且它发生在字节全部收完之后，
+        // 从进度上看与「下载完成」一模一样。
+        let on_disk = part.len();
+        let path = part.path().to_path_buf();
+        let algo = integrity.algo;
+        let digest = tokio::time::timeout(
+            self.cfg.verify_timeout,
+            tokio::task::spawn_blocking(move || verify::digest_file(&path, algo)),
+        )
+        .await
+        .map_err(|_| {
+            DownloadError::stalled(
+                StallReport::deadline(
+                    StallStage::Verify,
+                    on_disk,
+                    expected,
+                    self.cfg.verify_timeout,
+                ),
+                format!(
+                    "整文件重算摘要超过 {}ms",
+                    self.cfg.verify_timeout.as_millis()
+                ),
+            )
+        })?
+        .map_err(|err| {
+            DownloadError::new(DownloadErrorKind::Io, format!("校验任务异常退出：{err}"))
+        })?
+        .map_err(|err| io_err("读取中间文件失败", &err))?;
+        if verify::matches(&integrity.hex, &digest) {
            return Ok(());
        }
        part.discard();
        Err(DownloadError::new(
            DownloadErrorKind::Integrity,
-            format!("校验不符：期望 {} 实得 {actual}", integrity.hex),
+            format!("校验不符：期望 {} 实得 {digest}", integrity.hex),
        ))
    }
 }