fix(download): 补齐 download_bytes 与等许可两处无界 await，预算耗尽不再重试（M289 r1）

r1 复审 p2-1：`stall.rs` 那句「全仓只有两处无界 await」是假话。真实清单五处，
M289 之前只修了两处；这次补齐：

* `bytes.rs::fetch_bytes` 的 `send()` 改走 `http::send_bounded`（与文件下载同一个构造点）；
  整包 `response.bytes()` 换成 `bytes_stream` + `http::read_step` 的空闲判定，并边读边卡
  `MAX_IN_MEMORY_BYTES`（`Content-Length` 可缺失/可撒谎）。生产调用点：版本清单、
  版本 JSON、Modrinth 元数据。
* `engine::acquire` 的 `acquire_owned()` 加上界（`EngineConfig::permit_timeout`，默认
  120min ≥ 单个作业总期限），到点报新阶段 `StallStage::AcquirePermit`（「等许可」），
  不冒充「服务器不给数据」。这就是「卡死作业攥着许可把队列堵死」的连坐形态。
* `stall.rs` 模块头改成一张「哪些入口有界、谁托管给谁」的表，并写清五处 await 的真实清单。

r1 复审 p2-2：总期限用完是链路级判决，不再按满预算重试/换源：
* `DownloadError::job_budget_exhausted()` + `DownloadError::is_retryable()`（覆盖
  `kind.is_retryable()`），引擎与 `download_bytes` 都改用它；
* 引擎在预算耗尽时直接结束候选循环，不再换源（同一份字节、同一条链路，换源只会再
  烧满一整个预算）。188MB/20 KiB/s 的用户不再从「1.6 小时慢慢下完」变成「几小时反复重下」。
* `stall.rs` 写明「慢链路照样活」的前提 = **平均吞吐 ≥ 32 KiB/s**，以及它挡掉哪类链路；
  `download.*` schema 接线留给另一个 mission（`setting/keys.rs` 不在 scope）。

顺带（复审 p3）：
* `transfer.rs` 的读空闲统一走 `EngineConfig::deadlines()`（不再各读字段）；
* `verify_part` 补一条代价说明：`spawn_blocking` 到点不可取消，可能让同名重建拿到
  `Access is denied`（**该假设未验**，注释里已标）；
* `engine.rs` 因这轮新增又贴到 410 行 ⇒ `finish` / `materialize` / `millis` 逐字节搬到
  `engine/finish.rs`（`engine::millis` 路径用 `pub(crate) use` 保持可用）。

新增用例（全离线、<2s）：`download_bytes` 静默服务端 / 正文中途卡住、等许可超时、
预算耗尽不重试（`max_attempts_per_source = 3` 下仍只发一轮）、预算耗尽不换源
（官方源 `hits == 0`）。

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();
+}

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<u64>) -> 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}");
+}

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<MockResponse> {
+    (0..count).map(|_| MockResponse::silent()).collect()
+}

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<MockResponse> {
-    (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<u64>) -> 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();
+}

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;