Merge branch 'feat/m293-chunkresume-chunked-download-writes'

crates/core/src/download/chunk_resume.rs

@@ -0,0 +1,295 @@
+//! 分块下载的**续传判定与旧字节自证**。
+//!
+//! 台账 P21 第 2 条 + P19(b)：分块路径此前**不写**续传元数据、每次进 `attempt_chunked` 先
+//! `part.discard()`，于是 188MB 的 JDK 任何一次失败都从 0 重下。这里补上「记住落了哪些段」，
+//! 但**不是**把旧字节当可信——分块路径的旧字节比单连接的更可疑，原因有三条：
+//!
+//! 1. 中间文件被 `set_len(total)` 预分配过，长度恒等于总长：`resume::decide` 的「长度 = 已下」
+//!    这条前提在这里根本不成立（它会判成 `Complete`，然后整文件校验把一份没下完的文件判死）；
+//! 2. 段是**乱序**写的，磁盘上「有字节」不等于「这一段是完整的」——进程被强杀时正写到一半的段
+//!    与写完了的段在文件里长得一模一样；
+//! 3. 同一个 URL 的响应可能在两次尝试之间换了内容（镜像回源、滚动发布）。
+//!
+//! 所以复用旧段要过两道门，缺一不可：
+//!
+//! * **身份**（[`decide`]，纯函数）：策略允许续传、元数据在、URL / 总长 / 摘要算法 / 期望摘要
+//!   全部与本次请求一致、有 ETag 或 Last-Modified 作校验器（续传请求会带 `If-Range`，让服务器
+//!   自己回答「内容变了没有」），并且每条记录都能**精确对上**本次分块规划里的某一段；
+//! * **自证**（[`verify_records`]）：按记录重读那段字节、重算摘要，与记录比对。
+//!
+//! 两道门任一不过 ⇒ 丢掉整份中间文件**从头再来**（不是「只补坏的那一段」）：
+//! 摘要对不上意味着这份元数据本身已经不可信（盘上数据在腐坏，或者元数据是崩溃前写的、
+//! 盘上并没有落那样），同一份元数据里的**其它**记录同样没有理由被信任。这条与
+//! [`super::resume`] 的保守立场一致——重下 100MB 也比交付一个坏文件便宜。
+//!
+//! 最后还有第三道保险：整文件摘要在传输结束后**整份重算**（`transfer_chunk::verify_part`），
+//! 覆盖新下的字节与复用的字节。前两道门只保证「旧字节是我们写的」，第三道门才保证
+//! 「我们写的是对的那份」。
+//!
+//! 写的这一侧是本文末的 [`ChunkJournal`]：每成功回收一段就落一次盘（原子写），失败时元数据
+//! 里已经有「哪几段、各自摘要是什么」，于是下一次尝试能接着下而不是从 0。
+//!
+//! # 校验器跟着**源**与**表示**走（复审 r1 + r2 的 p2-1）
+//!
+//! 校验器（ETag / Last-Modified）是「某个 URL 上**某一份表示**」的属性，不是这个文件的属性。
+//! 它有两个独立的失效维度，缺一不可：
+//!
+//! * **源**：镜像 / CDN / 对象存储各自生成，跨源几乎必然不同。把 A 的校验器带进源 B 的元数据，
+//!   身份门的「有校验器」就被一个不相干的值满足，于是拿 A 的 ETag 去问 B；
+//! * **表示**：同一个 URL 也可能换了一份表示（重新发布、CDN 回源到新对象）。`done` 为空表示这
+//!   一轮没有任何字节要接着用，此时上一份元数据里的校验器描述的不是这次要下的东西。
+//!
+//! 两条路的终点相同：按 RFC 7233 §3.2，服务器对不上的 `If-Range` 会回 200 + 完整正文，分块路径
+//! 拿不到 Range ⇒ `RangeUnsupported` ⇒ 整份丢掉退化单连接从头下（本 mission 要消灭的形态；r2 实测
+//! 同源换表示那条是 2.775×，比不续传的 1.775× 还费）。所以沿用旧校验器的条件只有一处判定
+//! （`transfer_chunk` 的 `carry` = 「同源 **且** `done` 非空」），写元数据与发 `If-Range` 都走它。
+//!
+//! # 已知代价（复审 r1 p3-3）
+//!
+//! * **每次续传都要重读「即将复用」的那些字节**（不是整份文件：复用 3/4 才读 3/4，1 段就只读
+//!   1 段）。这是本模块存在的意义本身——不复读就无法证明盘上旧字节还是当初那一段。188MB 文件
+//!   落完 3/4 时约 141MB 顺序读，本地 SSD 是毫秒级，网络盘上可能到几秒；它发生在
+//!   「进度从复用字节开始」之前，所以界面会看起来停在原地（未做专门进度事件，见 mission 报告
+//!   的跟进项）。要省掉这次读只有一条路：相信元数据里的摘要而不复核——那等于把「旧字节可信」
+//!   当前提，与上面的自证要求直接冲突，**明确不做**。
+//! * **请求没带期望摘要时，末尾那道整文件摘要空转**（`verify_part` 在 `integrity: None` 时直接
+//!   `Ok`）。此时「续传后重新校验」只剩两道：旧段自证（只能证明「这些字节是我们写下去的」）与
+//!   `If-Range`（只能证明「服务器那份表示没变」，而服务器忽略该头时就失效）。边界很清楚：
+//!   **长度相同、内容换了一份**的文件在这种配置下无法察觉（末尾没有摘要可比）。这不是本 mission
+//!   引入的（单连接路径一直如此，`transfer.rs` 零改动），但分块路径继承了它。现实入口是
+//!   `ModFile` 上游没给合法 sha1 的时候（`addon/online/download.rs` 只在 sha1 合法时才设
+//!   `integrity`，而 mod / 整合包正会走分块）；`Library` 元数据、`download.bin` 缓存也有无摘要
+//!   调用点。要收紧就得要求「分块路径必须带 integrity」，那会让这类下载退回单连接或直接失败——
+//!   属于产品决策，不在这里单方面改。
+//! * **超时后的读盘线程不可取消**：[`verify_records`] 跑在 `spawn_blocking` 上，调用方到达
+//!   `verify_timeout` 后会丢弃中间文件并重写同一路径，而那个线程可能还在读、还持着句柄
+//!   （与 `transfer_chunk::verify_part` 里记的那条同族）。假设未验：Windows 上可能让随后同名
+//!   重建拿到 `Access is denied`，表现为「预分配中间文件失败」而不是干净重试。有界、有分类，
+//!   但诊断会拐一道。
+
+use std::fs::File;
+use std::io::{self, Read, Seek, SeekFrom};
+use std::path::{Path, PathBuf};
+
+use super::cache::now_ms;
+use super::chunked::{self, ByteRange, ChunkFetch};
+use super::resume::{ChunkRecord, RestartReason, ResumeMeta};
+use super::types::{HashAlgo, ResumePolicy};
+use super::verify;
+
+/// 复用旧分块的判定结果。
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ChunkDecision {
+    /// 这些段可以复用——**但调用方仍须先跑 [`verify_records`]**，本判定只看不碰盘。
+    Reuse {
+        /// 已落盘且对得上本次规划的分块记录。
+        done: Vec<ChunkRecord>,
+    },
+    /// 从头下载（调用方要丢掉整份中间文件）。
+    Restart {
+        /// 原因（诊断用；与单连接路径共用同一个枚举）。
+        reason: RestartReason,
+    },
+}
+
+/// 判定能否复用旧分块（纯函数：不读盘、不看时钟）。
+///
+/// `part_len` 为 0 表示没有中间文件（或它是 0 字节）。`planned` 是本次要用的分块规划，
+/// 记录必须**逐一精确命中**其中的段：连接数变了、总长变了、或记录本身畸形，都判从头再来。
+#[must_use]
+pub fn decide(
+    policy: ResumePolicy,
+    part_len: u64,
+    meta: Option<&ResumeMeta>,
+    url: &str,
+    integrity: Option<(HashAlgo, &str)>,
+    total: u64,
+    planned: &[ByteRange],
+) -> ChunkDecision {
+    let restart = |reason| ChunkDecision::Restart { reason };
+    if policy == ResumePolicy::Never {
+        return restart(RestartReason::PolicyNever);
+    }
+    if part_len == 0 {
+        return restart(RestartReason::NoPart);
+    }
+    let Some(meta) = meta else {
+        return restart(RestartReason::NoMeta);
+    };
+    // 没有分块清单的元数据（单连接路径写的）证明不了「乱序段各自到哪」，一律从头来。
+    let Some(records) = meta.chunks.as_ref() else {
+        return restart(RestartReason::NoMeta);
+    };
+    if meta.url != url {
+        return restart(RestartReason::UrlChanged);
+    }
+    if let Some((algo, hash)) = integrity {
+        if meta.algo != algo || !meta.hash.eq_ignore_ascii_case(hash) {
+            return restart(RestartReason::HashChanged);
+        }
+    }
+    // 总长变了就是另一个文件：旧段的偏移全部作废（只比 URL 抓不住这个）。
+    if meta.total != Some(total) {
+        return restart(RestartReason::SizeChanged);
+    }
+    // 没有校验器就无法让服务器回答「内容变了没有」，只能从头下（与单连接同一立场）。
+    if meta.validator().if_range_value().is_none() {
+        return restart(RestartReason::NoValidator);
+    }
+    // 每条记录必须是规划里**原样**的一段，且不许重复：段边界一旦漂移，
+    // 「已落盘多少」就说不清了（长度恰好但位置错的文件会被整文件校验兜住，但那已经是白下）。
+    let mut done: Vec<ChunkRecord> = Vec::with_capacity(records.len());
+    for record in records {
+        if record.is_empty() || record.end >= total {
+            return restart(RestartReason::ChunkLayout);
+        }
+        if !planned
+            .iter()
+            .any(|range| range.start == record.start && range.end == record.end)
+        {
+            return restart(RestartReason::ChunkLayout);
+        }
+        if done
+            .iter()
+            .any(|seen| seen.start == record.start && seen.end == record.end)
+        {
+            return restart(RestartReason::ChunkLayout);
+        }
+        done.push(record.clone());
+    }
+    ChunkDecision::Reuse { done }
+}
+
+/// 本次还要抓的段 = 规划 − 已落盘（[`decide`] 已保证 `done ⊆ planned` 且边界精确）。
+#[must_use]
+pub fn remaining(planned: &[ByteRange], done: &[ChunkRecord]) -> Vec<ByteRange> {
+    planned
+        .iter()
+        .copied()
+        .filter(|range| {
+            !done
+                .iter()
+                .any(|record| record.start == range.start && record.end == range.end)
+        })
+        .collect()
+}
+
+/// 按记录逐段重读、重算摘要：返回**可复用的字节数**。
+///
+/// 任一记录对不上（摘要不符 / 读不满一段 / 读盘失败）就返回 `Err`，调用方必须整份丢掉。
+/// 同步阻塞 IO：异步路径要把它放进 `spawn_blocking` 并套上 `verify_timeout`
+/// （与 `transfer_chunk::verify_part` 同一套做法）。
+pub fn verify_records(path: &Path, records: &[ChunkRecord]) -> io::Result<u64> {
+    let mut file = File::open(path)?;
+    let mut buffer = vec![0_u8; 64 * 1024];
+    let mut verified = 0_u64;
+    for record in records {
+        let mut hasher = verify::Hasher::new(chunked::CHUNK_DIGEST);
+        file.seek(SeekFrom::Start(record.start))?;
+        let mut left = record.len();
+        while left > 0 {
+            let want = usize::try_from(left.min(buffer.len() as u64)).unwrap_or(buffer.len());
+            let read = file.read(&mut buffer[..want])?;
+            if read == 0 {
+                return Err(io::Error::new(
+                    io::ErrorKind::UnexpectedEof,
+                    format!(
+                        "分块 [{}..={}] 在盘上不够长（缺 {left} 字节）",
+                        record.start, record.end
+                    ),
+                ));
+            }
+            hasher.update(&buffer[..read]);
+            left -= read as u64;
+        }
+        let actual = hasher.finish();
+        if !verify::matches(&record.sha1, &actual) {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                format!(
+                    "分块 [{}..={}] 摘要不符：记录 {} 实得 {actual}",
+                    record.start, record.end, record.sha1
+                ),
+            ));
+        }
+        verified += record.len();
+    }
+    Ok(verified)
+}
+
+/// 续传元数据的增量写手：每落完一段就把「这一段已落盘、摘要是它」写进元数据。
+///
+/// 为什么不收工再写：进程被强杀 / 断电时收工那次写根本不会发生，而已经落盘的段是好几分钟到
+/// 几十分钟的字节。代价是每段一次小文件原子写（`plan` 最多切 `MAX_CONNECTIONS` = 16 段，
+/// 单个作业最多 16 次），而且**只有成功回收的段才记**——写到一半被 abort 的段不会被记，
+/// 下次整段重抓（于是「记录里的段」不会出现半截字节）。
+pub struct ChunkJournal {
+    /// 元数据本体（`chunks` 随进度增长）。
+    meta: ResumeMeta,
+    /// 落盘路径。
+    path: PathBuf,
+}
+
+impl ChunkJournal {
+    /// 建账并立刻落一次盘。
+    ///
+    /// 这一次写还担着第二个作用：元数据里 `chunks: Some(..)` 是「这个中间文件属于分块路径」的
+    /// 标记，下次尝试靠它躲开单连接路径「长度 = 已下满」的误判（预分配文件长度恒等于总长）。
+    pub fn begin(meta: ResumeMeta, path: PathBuf) -> Self {
+        let mut journal = Self { meta, path };
+        journal.save();
+        journal
+    }
+
+    /// 已落盘的段。
+    pub fn done(&self) -> &[ChunkRecord] {
+        self.meta.chunks.as_deref().unwrap_or_default()
+    }
+
+    /// 记一个已落盘段（顺带补上服务器校验器）。
+    ///
+    /// 校验器**只在还没有值时写入**，这是刻意的（复审 r1 问过）：同一个 URL 的续传里如果服务器
+    /// 换了 ETag，保留旧值反而是保守选择——下一次带着旧 ETag 发 `If-Range` 会被服务器判「变了」
+    /// 而回 200，于是干净重下；若改成覆盖成新值，下一次会把「旧字节 + 新表示」当成同一份续下去，
+    /// 只能等末尾整文件摘要才发现（那已经是白下）。跨源、跨表示的情况根本不走到这里：
+    /// `transfer_chunk` 的 `carry`（同源 **且** `done` 非空）已经让 `etag` 从 `None` 起步，
+    /// 这里填的只可能是**本源、本轮响应**带来的校验器。
+    pub fn record(&mut self, fetched: ChunkFetch) {
+        if self.meta.etag.is_none() {
+            self.meta.etag = fetched.etag;
+        }
+        if self.meta.last_modified.is_none() {
+            self.meta.last_modified = fetched.last_modified;
+        }
+        self.meta
+            .chunks
+            .get_or_insert_with(Vec::new)
+            .push(ChunkRecord {
+                start: fetched.range.start,
+                end: fetched.range.end,
+                sha1: fetched.sha1,
+            });
+        self.save();
+    }
+
+    /// 落盘（元数据写失败不影响下载本身：续传只是加速，不是正确性前提）。
+    ///
+    /// **同步写**（`write_atomic`：建 tmp → 写 → `sync_all` → rename）跑在 async 任务里。判断是
+    /// 「可以接受」，量级与理由（复审 r2 p3-2）：
+    /// * 一个作业最多 `MAX_CONNECTIONS`(16) 段 + 建账那次 = **≤17 次**，每次 1–3 KiB（一条记录
+    ///   约 80 字节）。本地 SSD 每次约 1–2 ms ⇒ 整作业 ≤ ~30 ms；网络盘上一次 rename 可能几十
+    ///   ms ⇒ ≤ ~1 s，摊在「几分钟到几小时」的作业上（落完一段本来就要几十秒到几分钟），占比
+    ///   1e-4 量级；
+    /// * 顺序有意义：必须**先**把「这一段已落盘 + 摘要」写到盘上，才谈得上「记账可信」
+    ///   （record-then-trust）。挪进 `spawn_blocking` 就得让 journal 跨 await 被拥有或加锁，
+    ///   为几毫秒的 executor 占用引入一处顺序风险，不划算；
+    /// * 与单连接路径同口径：那边也是在 async 里同步 `meta.save`（`transfer.rs`）。只改分块侧会
+    ///   让两条路径的行为分叉。
+    ///
+    /// 退路：真出现 executor 饥饿（并发下载很多 + 网络盘）时，把 `record()` 整体挪进
+    /// `spawn_blocking`、journal 变成 `Arc<Mutex<_>>` 是可行方向，那时再收敛。
+    pub fn save(&mut self) {
+        self.meta.saved_at_ms = now_ms();
+        let _ = self.meta.save(&self.path);
+    }
+}

crates/core/src/download/chunked.rs

@@ -11,7 +11,14 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use super::error::{DownloadError, DownloadErrorKind};
 use super::http;
 use super::stall::Deadlines;
-use super::types::ResourceKind;
+use super::types::{HashAlgo, ResourceKind};
+use super::verify;
+
+/// 分块自证用的摘要算法（[`super::resume::ChunkRecord`] 里记的就是它）。
+///
+/// 固定 SHA-1 而不是跟随请求的校验算法：这条摘要是「这段字节是我们写下去的」的**归属证据**，
+/// 不对抗攻击者；整文件的期望摘要仍按请求里的 `algo` 校验。固定算法让记录不必再存一个算法字段。
+pub const CHUNK_DIGEST: HashAlgo = HashAlgo::Sha1;
 
 /// 默认分块连接数（设计文档 Q19 定案 N=4）。
 pub const DEFAULT_CONNECTIONS: usize = 4;
@@ -205,11 +212,35 @@ pub fn content_range_matches(requested: ByteRange, value: Option<&str>) -> bool
         .is_some_and(|parsed| parsed.start == requested.start && parsed.end == requested.end)
 }
 
+/// 一次成功分块抓取的产物：落盘的字节数 + **它的身份信息**。
+///
+/// 为什么把这些一起带回来：分块路径的续传元数据（`ResumeMeta::chunks`）要记「这一段是什么」
+/// ——段区间、这段字节的摘要（[`Self::sha1`]）、服务器的校验器与总长。它们只有响应头/读流这一处
+/// 拿得到，带回来由编排层决定怎么落盘，比在这里再读一遍盘便宜。
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ChunkFetch {
+    /// 本段区间。
+    pub range: ByteRange,
+    /// 实际落盘字节数（校验通过时恒等于 `range.len()`）。
+    pub written: u64,
+    /// 本段字节的摘要（边写边算，算法见 [`CHUNK_DIGEST`]）。
+    pub sha1: String,
+    /// 响应 `Content-Range` 里的总长（服务器没给就是 `None`）。
+    pub total: Option<u64>,
+    /// 响应 `ETag`。
+    pub etag: Option<String>,
+    /// 响应 `Last-Modified`。
+    pub last_modified: Option<String>,
+}
+
 /// 抓一个段，写到预分配文件的 `range.start` 偏移处，并把字节数累加进 `progress`。
 ///
 /// 服务器若对带 `Range` 的请求返回 200（无视 Range），立刻返回 `RangeUnsupported`
 /// —— 引擎据此退化单连接，而不是把整份文件重复写 N 遍。
 ///
+/// `if_range` 是续传时的校验器（ETag 优先 / Last-Modified 兜底）：服务器照办回 206，
+/// 内容变过回 200 —— 于是「远端换了内容」由协议层回答，不必等整文件摘要才发现。
+///
 /// 两处 `await` 都有界（M289）：等响应头用 [`http::send_bounded`]，等正文用
 /// [`http::read_step`]，上界来自 [`Deadlines`]。失速时错误里带阶段与本段口径的字节数。
 pub async fn fetch_chunk(
@@ -219,12 +250,16 @@ pub async fn fetch_chunk(
     part: &Path,
     deadlines: Deadlines,
     progress: &AtomicU64,
-) -> Result<u64, DownloadError> {
+    if_range: Option<&str>,
+) -> Result<ChunkFetch, DownloadError> {
     use tokio::io::{AsyncSeekExt, AsyncWriteExt};
 
-    let request = client
+    let mut request = client
         .get(url)
         .header(reqwest::header::RANGE, range.header_value());
+    if let Some(value) = if_range {
+        request = request.header(reqwest::header::IF_RANGE, value);
+    }
     let response = http::send_bounded(request, url, deadlines, Some(range.len())).await?;
 
     let status = response.status().as_u16();
@@ -252,6 +287,13 @@ pub async fn fetch_chunk(
         ));
     }
 
+    // 顺手取走总长：编排层要拿它比对「这个源给的还是不是同一个文件」。
+    let total = content_range
+        .and_then(parse_content_range)
+        .and_then(|parsed| parsed.total);
+    let etag = http::etag(response.headers());
+    let last_modified = http::last_modified(response.headers());
+
     let mut file = tokio::fs::OpenOptions::new()
         .write(true)
         .open(part)
@@ -262,6 +304,7 @@ pub async fn fetch_chunk(
         .map_err(|err| io_error(&err, "定位分块偏移失败"))?;
 
     let mut stream = response.bytes_stream();
+    let mut hasher = verify::Hasher::new(CHUNK_DIGEST);
     let mut written = 0_u64;
     loop {
         let next = http::read_step(
@@ -277,6 +320,7 @@ pub async fn fetch_chunk(
         file.write_all(&chunk)
             .await
             .map_err(|err| io_error(&err, "写入分块失败"))?;
+        hasher.update(&chunk);
         let len = chunk.len() as u64;
         written += len;
         progress.fetch_add(len, Ordering::Relaxed);
@@ -291,7 +335,14 @@ pub async fn fetch_chunk(
             format!("分块长度不符：期望 {} 实得 {written}（{url}）", range.len()),
         ));
     }
-    Ok(written)
+    Ok(ChunkFetch {
+        range,
+        written,
+        sha1: hasher.finish(),
+        total,
+        etag,
+        last_modified,
+    })
 }
 
 fn io_error(err: &std::io::Error, context: &str) -> DownloadError {

crates/core/src/download/mod.rs

@@ -29,6 +29,7 @@
 //! health        源 × 资源类型的健康度记忆
 //! cache         内容寻址缓存（记录信任）      atomic    落盘守卫 / 原子写
 //! resume        保守续传判定                  verify    流式摘要
+//! chunk_resume  分块续传判定 + 旧字节自证
 //! progress      事件流                        error     错误分类
 //! shared        进程级共享：一个数据根一个引擎（跨请求复用连接池/限流/健康度）
 //! ```
@@ -49,6 +50,7 @@ pub mod adoptium;
 pub mod backoff;
 pub mod blocking;
 pub mod cache;
+pub mod chunk_resume;
 pub mod chunked;
 pub mod config;
 pub mod engine;
@@ -79,6 +81,7 @@ pub use self::atomic::{PartFile, ADOPT_MIN_AGE, META_SUFFIX, ORPHAN_MAX_AGE, PAR
 pub use self::backoff::{BackoffPolicy, Jitter};
 pub use self::blocking::block_on;
 pub use self::cache::{CacheRecord, CacheStats, ContentCache};
+pub use self::chunk_resume::ChunkDecision;
 pub use self::chunked::{ByteRange, ChunkedConfig, RangeSupport};
 pub use self::config::EngineConfig;
 pub use self::engine::DownloadEngine;
@@ -92,7 +95,7 @@ pub use self::mirrors::{direct_source_id, AdoptiumMirror, GhProxy, MirrorTable,
 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::resume::{ChunkRecord, 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,

crates/core/src/download/resume.rs

@@ -13,6 +13,14 @@
 //! 任何一条不满足就**从头下**——重下 100MB 也比交付一个坏文件便宜。
 //! 另外单独处理 416（`Requested Range Not Satisfiable`）：已下满则直接进入校验，
 //! 否则丢弃中间文件重来。
+//!
+//! # 分块路径（`chunks` 字段）
+//!
+//! 多连接分块写的是**乱序的段**，单连接那套「长度 = 偏移」的续传判定完全不适用：文件被
+//! `set_len(total)` 预分配过，长度永远等于总长，`Complete` 判定会被它骗到。所以分块路径把
+//! 「已落盘的分块清单」写进同一份元数据的 [`ResumeMeta::chunks`]，判定逻辑在
+//! [`super::chunk_resume`]：那边要求 URL / 总长 / 期望摘要 / 校验器全部对得上，**而且**
+//! 每条记录都要能重算出与记录一致的摘要，才允许复用旧字节。
 
 use std::io;
 use std::path::Path;
@@ -39,6 +47,49 @@ pub struct ResumeMeta {
     pub hash: String,
     /// 保存时间（unix 毫秒）。
     pub saved_at_ms: u64,
+    /// 分块路径已落盘的分块清单（单连接路径不写也不读，恒为 `None`）。
+    ///
+    /// `Some([])` 与 `None` 是两件事：前者表示「这个中间文件是分块路径预分配的、此刻一段都还没
+    /// 落完」，后者表示「这份元数据不是分块路径写的」。单连接路径要靠这个区别躲开
+    /// 「预分配全长 = 下满了」的误判（见 [`super::chunk_resume`]）。
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub chunks: Option<Vec<ChunkRecord>>,
+}
+
+/// 一个已落盘分块的**自证记录**：段区间 + 写盘时顺手算出的摘要。
+///
+/// 恢复时按这条记录重读那段字节、重算摘要并比对——这是「磁盘上这堆旧字节确实是我们上次
+/// 写下去的那一段」的唯一证据。只凭长度/URL/ETag 无法证明：同一个 URL 的响应可能换过内容，
+/// 进程被强杀也可能留下「元数据说写完了、盘上只写了一半」的段。
+///
+/// 摘要算法固定 SHA-1（见 [`super::chunked::CHUNK_DIGEST`]）：它在这里是**字节归属**的证据，
+/// 不是对抗性安全边界；整文件的期望摘要仍按请求里的算法校验。
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ChunkRecord {
+    /// 段起始偏移（含）。
+    pub start: u64,
+    /// 段结束偏移（**含**，与 [`super::chunked::ByteRange`] 同口径）。
+    pub end: u64,
+    /// 该段字节的小写十六进制摘要。
+    pub sha1: String,
+}
+
+impl ChunkRecord {
+    /// 段长度（字节；`end < start` 时为 0）。
+    #[must_use]
+    pub const fn len(&self) -> u64 {
+        if self.end < self.start {
+            0
+        } else {
+            self.end - self.start + 1
+        }
+    }
+
+    /// 空段。
+    #[must_use]
+    pub const fn is_empty(&self) -> bool {
+        self.end < self.start
+    }
 }
 
 impl ResumeMeta {
@@ -96,12 +147,18 @@ pub enum RestartReason {
     NoPart,
     /// 中间文件是 0 字节。
     EmptyPart,
+    /// 中间文件是分块路径预分配的全长文件：单连接的「长度 = 已下」不成立，交给分块路径判。
+    ChunkedPart,
     /// 没有元数据（无法证明中间文件是同一次下载的产物）。
     NoMeta,
     /// URL 变了（换源后偏移不可信）。
     UrlChanged,
     /// 期望摘要变了（目标文件已经不是同一个东西）。
     HashChanged,
+    /// 记录的总长与本次规划的总长不一致（远端换了文件，或换了个大小的目标）。
+    SizeChanged,
+    /// 已落盘分块的段区间与本次分块规划对不上（连接数/规划变了，旧段无法对号入座）。
+    ChunkLayout,
     /// 总长未知（无法判断是否已下满）。
     TotalUnknown,
     /// 已下字节超过总长（中间文件被污染）。
@@ -120,9 +177,12 @@ impl RestartReason {
             Self::PolicyNever => "policy_never",
             Self::NoPart => "no_part",
             Self::EmptyPart => "empty_part",
+            Self::ChunkedPart => "chunked_part",
             Self::NoMeta => "no_meta",
             Self::UrlChanged => "url_changed",
             Self::HashChanged => "hash_changed",
+            Self::SizeChanged => "size_changed",
+            Self::ChunkLayout => "chunk_layout",
             Self::TotalUnknown => "total_unknown",
             Self::Oversize => "oversize",
             Self::NoValidator => "no_validator",

crates/core/src/download/transfer.rs

@@ -50,8 +50,19 @@ impl DownloadEngine {
         let integrity = req.integrity.as_ref();
         let meta_path = part.meta_path();
         let previous = ResumeMeta::load(&meta_path);
+        // 分块路径的中间文件是**预分配整长**的：长度等于总长不代表下满了。元数据里带分块清单
+        // （`chunks`）就说明这份文件是分块路径写的，单连接的 `Complete` 判定在这里必然判错
+        // （它会把「3/4 块完成」当成「已下满」，然后整文件校验白读一遍再判死）。
+        // 这种文件一律交给分块路径：它能按段对号入座，且逐段重算摘要后才敢复用。
+        let chunked_part = previous
+            .as_ref()
+            .is_some_and(|meta| meta.chunks.is_some() && part.exists());
         // 「没有中间文件」和「有但 0 字节」是两种不同的原因，分开报出来便于诊断。
-        let decision = if !part.exists() {
+        let decision = if chunked_part {
+            ResumeDecision::Restart {
+                reason: RestartReason::ChunkedPart,
+            }
+        } else if !part.exists() {
             ResumeDecision::Restart {
                 reason: RestartReason::NoPart,
             }
@@ -207,6 +218,8 @@ impl DownloadEngine {
             algo: integrity.map_or(HashAlgo::Sha1, |item| item.algo),
             hash: integrity.map_or_else(String::new, |item| item.hex.clone()),
             saved_at_ms: now_ms(),
+            // 单连接路径不写分块清单：这份文件是「从头顺序追加」的，长度就是已下字节。
+            chunks: None,
         };
         let _ = meta.save(&part.meta_path());
 

crates/core/src/download/transfer_chunk.rs

@@ -1,21 +1,42 @@
 //! 大文件多连接分块的编排与校验。
 //!
 //! 分块本身由 `chunked.rs` 负责（纯规划 + 单块抓取），这里负责并发、进度上报、
-//! 失败清理与「整文件重算一次摘要」（分块是乱序写，只能整文件校验）。
+//! 续传元数据的读写与「整文件重算一次摘要」（分块是乱序写，只能整文件校验）。
+//!
+//! # 续传（M293：台账 P21 第 2 条 + P19(b)）
+//!
+//! 此前这里开头无条件 `part.discard()`、失败也 `discard()`，于是 188MB 的 JDK 任何一次失败
+//! 都从 0 重下；M289 又给「预算耗尽」加了「不再重试、不再换源」的短路，慢链路上一次晚到的
+//! 失败就变成「这个文件下不来」。现在：
+//!
+//! * 每落完一段就把它连同**这段字节的摘要**写进 `ResumeMeta::chunks`（`ChunkJournal`）；
+//! * 失败**不再**丢中间文件——已完成的段是几十分钟的字节，元数据里带着它们的自证记录；
+//! * 下次进本函数先过 [`super::chunk_resume::decide`]（身份）与 [`Self::verify_chunks`]
+//!   （旧字节自证），两道门都过才复用；任何一道不过就整份丢掉从头再来；
+//! * 整文件摘要照旧在传输结束后整份重算，覆盖新下 + 复用的全部字节。
+//!
+//! 唯一仍然主动 `discard()` 的失败是 `RangeUnsupported`：服务器不支持 Range 时单连接路径
+//! 接不住「预分配 + 乱序段」这种形状的文件（它只会截断重写），留着没意义。
 
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
 use super::atomic::PartFile;
+use super::cache::now_ms;
+use super::chunk_resume::{self, ChunkDecision, ChunkJournal};
 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::resume::{ChunkRecord, ResumeMeta};
+use super::stall::{StallReport, StallStage};
+use super::types::{Candidate, DownloadId, DownloadRequest, HashAlgo, Integrity};
 use super::verify;
 
+/// 分块任务的回收（失速判据 + 已完成段记账）：拆出守 400 行硬线。
+mod join;
+
 impl DownloadEngine {
     pub(crate) async fn attempt_chunked(
         &self,
@@ -38,53 +59,141 @@ impl DownloadEngine {
                 "可切分段不足，退化为单连接",
             ));
         }
+        let integrity = req.integrity.as_ref();
+
+        // ① 身份：旧分块是不是**这次**下载的产物（纯判定，不碰盘）。
+        let meta_path = part.meta_path();
+        let previous = ResumeMeta::load(&meta_path);
+        let decision = chunk_resume::decide(
+            req.resume,
+            part.len(),
+            previous.as_ref(),
+            &candidate.url,
+            integrity.map(|item| (item.algo, item.hex.as_str())),
+            total,
+            &ranges,
+        );
+        // ② 自证：按记录重读那些字节、重算摘要。任何一段对不上就整份丢掉从头再来
+        //    （理由见 `chunk_resume` 模块头：对不上说明这份元数据本身已经不可信）。
+        let mut done: Vec<ChunkRecord> = Vec::new();
+        let mut reused = 0_u64;
+        match decision {
+            ChunkDecision::Reuse { done: records } => {
+                // 复用字节数直接取自证的结果（唯一真相源：少读一段就少算一段）。
+                match self.verify_chunks(part, &records).await {
+                    Some(bytes) => {
+                        done = records;
+                        reused = bytes;
+                    }
+                    None => part.discard(),
+                }
+            }
+            ChunkDecision::Restart { .. } => part.discard(),
+        }
+        // 校验器既是**源**的属性（镜像/CDN/对象存储各自生成 ETag），也是**一份表示**的属性——
+        // 两件事缺一不可，所以「谁能当校验器」只有 `carry` 这一个判定点：
+        //
+        // * **源**（复审 r1 p2-1）：`previous.url` 不是当前源就当没有。把 A 的校验器写进 `url=B`
+        //   的新元数据之后，身份门的 `NoValidator` 会被一个不相干的值满足，于是拿 **A 的 ETag**
+        //   去对 B 发 `If-Range`：按 RFC 7233 §3.2，服务器对不上就回 200 + 完整正文 ⇒
+        //   `fetch_chunk` 判 `RangeUnsupported` ⇒ 丢掉整份 ⇒ 单连接从头下。
+        // * **表示**（复审 r2 p2-1）：`done` 为空表示这一轮没有任何字节要接着用（`Restart`，或
+        //   第一次跑），那上一份元数据里的校验器描述的是**另一份**表示。留着它这一轮看不出问题
+        //   （`if_range` 本来就不发），但它会被原样写进新元数据，而 `record` 只在缺值时写 ⇒ 本轮
+        //   响应的真实校验器永远盖不上它 ⇒ 下一次续传拿一个已知可能不匹配的值去问服务器 ⇒ 同上
+        //   那条 200/丢整份的路。实测那是 2.775×，比「干脆不续传」的 1.775× 还费一整份。
+        //   清掉之后，本轮响应里的真实校验器由 `ChunkJournal::record` 填进来；最坏也只是
+        //   `NoValidator` ⇒ 整份重下（1.775× 那条基线），不会更坏。
+        //
+        // 为什么不选复审给的另两种修法：「换源时清空」（B）只是「源」这一维的特例写法，条件放在
+        // 读取处比散到写入处更难漏；「校验器加来源字段」（C）会在 `ResumeMeta` 里多一份可与
+        // `url` 漂移的真相，而当前模型是「一个中间文件 = 一个目标」，没有并存多源的形态。
+        let same_source = previous.as_ref().filter(|meta| meta.url == candidate.url);
+        let carry = if done.is_empty() { None } else { same_source };
+        // 续传时把**同源、同一份表示**的旧校验器带上：这是 `If-Range` 的取值来源，也是「远端
+        // 内容没变」的协议层证据。
+        let if_range = carry.and_then(|meta| meta.validator().if_range_value().map(str::to_string));
 
-        part.discard();
         part.ensure_parent_async()
             .await
             .map_err(|err| io_err("创建下载目录失败", &err))?;
         {
-            let file = tokio::fs::OpenOptions::new()
-                .create(true)
-                .write(true)
-                .truncate(true)
+            let mut options = tokio::fs::OpenOptions::new();
+            options.create(true).write(true);
+            if done.is_empty() {
+                // 从头：先把可能存在的旧内容截掉，再预分配整长。
+                options.truncate(true);
+            }
+            let file = options
                 .open(part.path())
                 .await
                 .map_err(|err| io_err("预分配中间文件失败", &err))?;
+            // 续传时这一行是「归一化」：分块写是随机偏移，长度本该已经是 total；
+            // 若被外部截短/加长过，先摆正，缺的段下面的补抓会填满。
             file.set_len(total)
                 .await
                 .map_err(|err| io_err("预分配长度失败", &err))?;
         }
 
-        let counter = Arc::new(AtomicU64::new(0));
+        let mut journal = ChunkJournal::begin(
+            ResumeMeta {
+                url: candidate.url.clone(),
+                // 只沿用「同一个源 + 同一份表示」的旧校验器（见上面的 `carry`）；其余情况当没有，
+                // 让下面每一段的响应自己去填。
+                etag: carry.and_then(|meta| meta.etag.clone()),
+                last_modified: carry.and_then(|meta| meta.last_modified.clone()),
+                total: Some(total),
+                algo: integrity.map_or(HashAlgo::Sha1, |item| item.algo),
+                hash: integrity.map_or_else(String::new, |item| item.hex.clone()),
+                saved_at_ms: now_ms(),
+                chunks: Some(done),
+            },
+            meta_path,
+        );
+
+        let counter = Arc::new(AtomicU64::new(reused));
         let reporter = self.spawn_reporter(id, Arc::clone(&counter), total);
 
         let deadlines = self.cfg.deadlines();
+        let todo = chunk_resume::remaining(&ranges, journal.done());
         let mut set = tokio::task::JoinSet::new();
-        for range in ranges {
+        for range in todo {
             let client = self.http.clone();
             let url = candidate.url.clone();
             let path = part.path().to_path_buf();
             let progress = Arc::clone(&counter);
+            let if_range = if_range.clone();
             set.spawn(async move {
-                chunked::fetch_chunk(&client, &url, range, &path, deadlines, &progress).await
+                chunked::fetch_chunk(
+                    &client,
+                    &url,
+                    range,
+                    &path,
+                    deadlines,
+                    &progress,
+                    if_range.as_deref(),
+                )
+                .await
             });
         }
 
         let failure = self
-            .await_chunks(&mut set, &counter, total, &candidate.url)
-            .await
-            .map(|err| err.with_source(candidate.source.as_str()));
+            .await_chunks(&mut set, &counter, total, &candidate.url, &mut journal)
+            .await;
         reporter.abort();
 
         if let Some(err) = failure {
+            // 已完成的段连同自证摘要都在元数据里（`ChunkJournal` 每段落一次盘），中间文件留着，
+            // 下一次（甚至下次启动）接着下。只有「这个源不支持 Range」才把文件丢掉：
+            // 单连接路径接不住预分配 + 乱序段的形状。
+            if err.kind == DownloadErrorKind::RangeUnsupported {
                 part.discard();
-            return Err(err);
+            }
+            return Err(err.with_source(candidate.source.as_str()));
         }
 
-        // 分块是乱序写，只能整文件重算一次摘要。
-        self.verify_part(part, req.integrity.as_ref(), Some(total), id)
-            .await?;
+        // 分块是乱序写，只能整文件重算一次摘要——这一步覆盖新下的字节**与**复用的旧字节。
+        self.verify_part(part, integrity, Some(total), id).await?;
         self.emit(ProgressEvent::Progress {
             id,
             downloaded: total,
@@ -92,89 +201,10 @@ impl DownloadEngine {
         });
         Ok(AttemptOk {
             bytes: total,
-            resumed_from: 0,
+            resumed_from: reused,
         })
     }
 
-    /// 回收分块任务，同时盯着「整个作业有没有在动」。
-    ///
-    /// 两道判据必须分得开（另一处是单块自己的读空闲超时，阶段 = 等字节）：
-    /// * **无进度**：`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,
@@ -197,6 +227,31 @@ impl DownloadEngine {
         })
     }
 
+    /// 旧分块自证：按记录逐段重读、重算摘要。
+    ///
+    /// `Some(bytes)` = 每一段都对得上，返回值就是**可复用的字节数**（调用方直接拿它当
+    /// `resumed_from`，别再从记录另算一遍——两份口径会静默漂移）；`None` = 证明不了
+    /// （摘要不符 / 盘上不够长 / 读盘失败 / 超时），调用方必须整份丢掉。
+    ///
+    /// 超时也归到「证明不了」：`spawn_blocking` 的线程不可取消，到点后它还在读文件、还持着
+    /// 句柄（与 `verify_part` 里那段注释同族，代价见 `chunk_resume` 模块头「已知代价」）。
+    async fn verify_chunks(&self, part: &PartFile, records: &[ChunkRecord]) -> Option<u64> {
+        let path = part.path().to_path_buf();
+        let records = records.to_vec();
+        let job =
+            tokio::task::spawn_blocking(move || chunk_resume::verify_records(&path, &records));
+        match tokio::time::timeout(self.cfg.verify_timeout, job).await {
+            Ok(Ok(Ok(bytes))) => Some(bytes),
+            _ => None,
+        }
+    }
+
+    /// 整文件摘要（有界、分类见下）。
+    ///
+    /// `integrity: None` 时**直接返回 `Ok`**：这类请求没有任何端到端的摘要可比，末尾这一步
+    /// 空转。两条路径都如此（单连接一直如此，本 mission 未改）。此时「续传后重新校验」只剩
+    /// 旧段自证 + `If-Range` 两道，`chunk_resume` 模块头的「已知代价」写了这个一致性边界
+    /// （长度相同、内容换了一份的文件不可察觉；现实入口是没有上游 sha1 的 `ModFile`）。
     pub(crate) async fn verify_part(
         &self,
         part: &mut PartFile,

crates/core/src/download/transfer_chunk/join.rs

@@ -0,0 +1,139 @@
+//! 分块任务的回收：两道失速判据 + 已完成段的记账。
+//!
+//! 从 `transfer_chunk.rs` 拆出只为守 400 行硬线（M293 r1：换源校验器修复 + 两条 p3 收口之后）。
+//! 搬出来的代码逐字节未改，仍挂在同一个 `impl DownloadEngine` 上。
+
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use super::super::chunk_resume::ChunkJournal;
+use super::super::chunked::ChunkFetch;
+use super::super::engine::DownloadEngine;
+use super::super::error::{DownloadError, DownloadErrorKind};
+use super::super::stall::{StallReason, StallReport, StallStage};
+
+impl DownloadEngine {
+    /// 回收分块任务，同时盯着「整个作业有没有在动」。
+    ///
+    /// 两道判据必须分得开（另一处是单块自己的读空闲超时，阶段 = 等字节）：
+    /// * **无进度**：`counter` 连续 `chunk_stall_timeout` 没有新字节。分块任务卡在写盘上
+    ///   （杀软/网络盘）时单块读超时不会响，只有这道抓得住——现场留给 `join_next` 的正是
+    ///   「188MB 落完 99% 之后永久挂住」那个形态；
+    /// * **总期限**：`chunk_job_budget`（按平均吞吐下限反推，见 `stall` 模块头）用完。
+    ///
+    /// 两者都归到阶段 [`StallStage::AwaitChunkJoin`]，靠 [`StallReason`] 区分。
+    ///
+    /// 成功的段在这里**顺手记账**（[`ChunkJournal::record`]）：这是唯一同时握着「段结果」与
+    /// 「段区间」的地方，落到元数据之后，本作业后面怎么挂都不影响这几段的续传价值。
+    pub(super) async fn await_chunks(
+        &self,
+        set: &mut tokio::task::JoinSet<Result<ChunkFetch, DownloadError>>,
+        counter: &AtomicU64,
+        total: u64,
+        url: &str,
+        journal: &mut ChunkJournal,
+    ) -> 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(fetched)))) => {
+                    // 「这个源给的还是不是同一个文件」：`Content-Range` 的总长必须与我们规划的
+                    // 一致，否则写下去的段根本拼不成目标文件（单连接路径同一处判据是收尾的
+                    // `written != total`）。
+                    if fetched.total.is_some_and(|server| server != total) {
+                        absorb_completed(journal, set, total);
+                        set.abort_all();
+                        return Some(DownloadError::new(
+                            DownloadErrorKind::Network,
+                            format!(
+                                "分块总长不符：响应声明 {}，本次规划 {total}（{url}）",
+                                fetched.total.unwrap_or_default()
+                            ),
+                        ));
+                    }
+                    journal.record(fetched);
+                }
+                Ok(Some(Ok(Err(err)))) => {
+                    absorb_completed(journal, set, total);
+                    set.abort_all();
+                    return Some(err);
+                }
+                Ok(Some(Err(err))) => {
+                    absorb_completed(journal, set, total);
+                    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,
+                        )
+                    };
+                    absorb_completed(journal, set, total);
+                    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));
+                }
+            }
+        }
+    }
+}
+
+/// 把「已经落完、但还没被 `join_next` 收走」的段也记进账（复审 r1 p3-1）。
+///
+/// `abort_all()` 只中止任务、不替我们保存已完成的结果；失败分支直接 `return` 等于把那些**字节
+/// 已经完整落盘**的段扔掉，下次整段重抓（4 连接时最多白丢 3 段）。这里在 abort 之前用
+/// `try_join_next` 把就绪结果收干净，总长判据与正常回收同一道。
+///
+/// 边界：这是**尽力减少浪费**，不是正确性前提——没收走的段只会让下次多下，不会被误用
+/// （记录带段摘要自证）。也正因为它是「谁先完成」的竞态，没有确定性的用例能钉住它：
+/// 用例只能断言「失败后已完成的段被记账」，而失败与完成的先后顺序由调度器决定。
+fn absorb_completed(
+    journal: &mut ChunkJournal,
+    set: &mut tokio::task::JoinSet<Result<ChunkFetch, DownloadError>>,
+    total: u64,
+) {
+    while let Some(joined) = set.try_join_next() {
+        if let Ok(Ok(fetched)) = joined {
+            if fetched.total.is_none_or(|server| server == total) {
+                journal.record(fetched);
+            }
+        }
+    }
+}

crates/core/tests/download/chunked/last_modified.rs

@@ -0,0 +1,121 @@
+//! 弱校验器（`Last-Modified`）驱动 `If-Range` 的那条支路（复审 r2 p3-1）。
+//!
+//! 分块路径此前只在 **ETag** 上端到端覆盖过条件请求；`Validator::if_range_value()` 的兜底
+//! （ETag 缺失 ⇒ 用 Last-Modified）与 `ChunkJournal::record` 写 `last_modified` 这两条路，只有
+//! 单连接的纯函数用例覆盖（`resume_test.rs` 的 `missing_validator_restarts`），分块侧零覆盖。
+//!
+//! 这里补上：服务器**只给** `Last-Modified`（不给 ETag），3/4 段失败后重试必须带这个 HTTP-date
+//! 发 `If-Range`，夹具按弱校验器比对相符 ⇒ 206 ⇒ 只补 1 段（41984 = 1.025×），而不是整份重下。
+//!
+//! 抓得住：分块路径「只有弱校验器时也能续传」这条端到端行为（含 `record` 写 `last_modified`、
+//! `validator()` 兜底、夹具的 HTTP-date 比对）。
+//! 抓不住：真实服务器对 HTTP-date 型 `If-Range` 的**强/弱比较语义**（RFC 7233 §3.2 规定只能用
+//! 强校验器做 If-Range，HTTP-date 允许以 60s 粒度做「没变过」判定）——夹具只做字符串相等，这里
+//! 证明的是「我们自己这条链路会把日期原样送进 If-Range 并接受 206」，不是「所有真实服务器都会」。
+//! 若服务器判「不符」，按既有路径会回 200 ⇒ `RangeUnsupported` ⇒ 干净重下（正确性不受影响）。
+
+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 unlucky_core::download::types::{DownloadRequest, ResourceKind};
+
+use super::support::{sha1_hex, temp_dir, MockResponse, MockServer};
+
+/// 4 段 × 10 KiB。
+const SIZE: usize = 40 * 1024;
+/// 卡住的那一段吐这么多字节就停。
+const STALL_TAIL: usize = 1024;
+/// 服务器唯一的校验器（HTTP-date 原文，客户端会原样放进 `If-Range`）。
+const LAST_MODIFIED: &str = "Wed, 21 Oct 2015 07:28:00 GMT";
+
+fn payload() -> Vec<u8> {
+    (0..SIZE).map(|index| (index % 251) as u8).collect()
+}
+
+fn engine(dir: &std::path::Path) -> DownloadEngine {
+    DownloadEngine::new(
+        EngineConfig {
+            max_attempts_per_source: 2,
+            backoff: BackoffPolicy::none(),
+            connect_timeout: Duration::from_secs(2),
+            response_timeout: Duration::from_secs(2),
+            read_timeout: Duration::from_millis(300),
+            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()
+        },
+        Arc::new(MirrorTable::builtin()),
+        Arc::new(RegionProfile::direct(0)),
+        Arc::new(HealthRegistry::in_memory()),
+        Arc::new(ContentCache::open(dir.join("cache"))),
+    )
+}
+
+#[tokio::test]
+async fn last_modified_alone_can_drive_if_range_on_the_chunked_path() {
+    let payload = payload();
+    let size = payload.len() as u64;
+    // 只给弱校验器：没有 ETag，只有 Last-Modified。
+    let response = |body: Vec<u8>| {
+        MockResponse::ok(body)
+            .without_etag()
+            .with_last_modified(LAST_MODIFIED)
+    };
+    // 4 段里 3 段成功、1 段卡住；重试只该补 1 段；后面 4 条给「退化成整份重下」兜底。
+    let mut script: Vec<MockResponse> = (0..3).map(|_| response(payload.clone())).collect();
+    script.push(response(payload.clone()).stalling_after(STALL_TAIL));
+    script.push(response(payload.clone()));
+    script.extend((0..4).map(|_| response(payload.clone())));
+    let server = MockServer::start(script).await;
+    let url = server.url("/jdk.zip");
+    let dir = temp_dir("chunked-last-modified");
+    let engine = engine(&dir);
+    let dest = dir.join("jdk.zip");
+
+    let outcome = engine
+        .download(
+            DownloadRequest::new(ResourceKind::JavaRuntime, url.as_str(), &dest)
+                .with_sha1(sha1_hex(&payload))
+                .with_size(size),
+        )
+        .await
+        .expect("只有弱校验器时也要能续传完成");
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+
+    // 主判据：重试那一次带的是 **Last-Modified**（HTTP-date），而且只有那一次有条件请求。
+    let seen = server.if_range_values();
+    assert_eq!(
+        seen.len(),
+        1,
+        "只该在重试那一次发条件请求（首轮没有可续的前缀）：{seen:?}"
+    );
+    assert_eq!(
+        seen[0].to_ascii_lowercase(),
+        LAST_MODIFIED.to_ascii_lowercase(),
+        "ETag 缺失时 `If-Range` 必须兜底用 Last-Modified，实得 {:?}",
+        seen[0]
+    );
+    // 3 段 + 卡住的 1KiB + 补 1 段 = 整份 + 1KiB = 41984（1.025×）。
+    assert_eq!(
+        server.bytes_sent(),
+        size + STALL_TAIL as u64,
+        "弱校验器那一路也该只补 1 段（1.025×），不是整份重下"
+    );
+    assert_eq!(server.hits(), 5, "第一次 4 段 + 第二次补 1 段");
+    assert_eq!(outcome.resumed_from, size * 3 / 4, "复用了 3 段");
+    server.stop();
+}

crates/core/tests/download/chunked/resume.rs

@@ -0,0 +1,361 @@
+//! 分块路径的续传（M293 / 台账 P21 第 2 条 + P19(b)）：真跑两次尝试，**数服务端吐了多少字节**。
+//!
+//! 判据刻意不用「最终 hash 对」：hash 对只说明拼出来的文件没错，抓不住「其实整份又下了一遍」——
+//! 那正是 P21 那条债的形态（188MB 的 JDK 一次失败就从 0 重下）。所以每个用例都用夹具统计
+//! 服务端实际发出的正文字节数（[`MockServer::bytes_sent`]）：
+//!
+//! * `resumes_the_finished_chunks_after_a_partial_failure`：第一次在 3/4 处失败、第二次续传完成
+//!   ⇒ 两次总字节必须 **< 1.5×文件大小**。变异：把 `attempt_chunked` 改回「进来先 `discard()`、
+//!   失败也 `discard()`」⇒ 第二次必然整份重下，该断言红（1.75×）。
+//! * `without_resume_the_same_failure_downloads_again`：同一份夹具、同一个失败点，只把续传策略
+//!   换成 `Never` ⇒ 服务端确实吐出 ~1.75×。这条是上一条的**可执行变异证明**：`< 1.5×` 这个阈值
+//!   真的把「续传」和「全重下」分开了，不是随便定的一个宽阈值。
+//! * `corrupted_old_chunks_are_never_stitched_in`：盘上的旧段被改过一个字节 ⇒ 一段都不许复用
+//!   （摘要自证不过就整份丢掉从头再来），服务端要重新吐全部字节。
+//! * `changed_size_invalidates_old_chunk_layout`：总长换了（同一个目标文件，大小变了）⇒ 旧段一段
+//!   都不许用。这条钉的是「旧元数据不得被误用」。
+
+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::{plan, ByteRange, 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 unlucky_core::download::resume::{ChunkRecord, ResumeMeta};
+use unlucky_core::download::types::{DownloadRequest, HashAlgo, ResourceKind, ResumePolicy};
+use unlucky_core::download::verify::digest_bytes;
+
+use super::support::{sha1_hex, temp_dir, MockResponse, MockServer};
+
+/// 4 段 × 10 KiB。
+const SIZE: usize = 40 * 1024;
+/// 卡住的那一段吐这么多字节就停（这部分字节落盘了但不会被记账）。
+const STALL_TAIL: usize = 1024;
+
+fn payload() -> Vec<u8> {
+    (0..SIZE).map(|index| (index % 251) as u8).collect()
+}
+
+/// 一次尝试允许 2 次（第一次故意失败、第二次续传）；读空闲 300ms：卡住的那一段自己报错，
+/// 那时其余三段早就落完并被回收了（作业级的无进度窗口是 10s，不会抢在它前面）。
+fn config() -> EngineConfig {
+    EngineConfig {
+        max_attempts_per_source: 2,
+        backoff: BackoffPolicy::none(),
+        connect_timeout: Duration::from_secs(2),
+        response_timeout: Duration::from_secs(2),
+        read_timeout: Duration::from_millis(300),
+        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 engine(dir: &std::path::Path) -> DownloadEngine {
+    DownloadEngine::new(
+        config(),
+        Arc::new(MirrorTable::builtin()),
+        Arc::new(RegionProfile::direct(0)),
+        Arc::new(HealthRegistry::in_memory()),
+        Arc::new(ContentCache::open(dir.join("cache"))),
+    )
+}
+
+fn request(url: &str, dest: &std::path::Path, payload: &[u8]) -> DownloadRequest {
+    DownloadRequest::new(ResourceKind::JavaRuntime, url, dest)
+        .with_sha1(sha1_hex(payload))
+        .with_size(payload.len() as u64)
+}
+
+/// 第一次尝试的脚本：4 段里 3 段正常落完，第 4 个连接拿到「吐一点就卡住」的那份。
+fn partial_script(payload: &[u8]) -> Vec<MockResponse> {
+    let mut script: Vec<MockResponse> =
+        (0..3).map(|_| MockResponse::ok(payload.to_vec())).collect();
+    script.push(MockResponse::ok(payload.to_vec()).stalling_after(STALL_TAIL));
+    script
+}
+
+fn record_of(payload: &[u8], range: ByteRange) -> ChunkRecord {
+    ChunkRecord {
+        start: range.start,
+        end: range.end,
+        sha1: digest_bytes(
+            HashAlgo::Sha1,
+            &payload[range.start as usize..=range.end as usize],
+        ),
+    }
+}
+
+/// 分块路径的元数据（`chunks: Some(..)` 是「这个中间文件属于分块路径」的标记）。
+fn chunk_meta(url: &str, total: u64, hash: &str, chunks: Vec<ChunkRecord>) -> ResumeMeta {
+    ResumeMeta {
+        url: url.to_string(),
+        etag: Some("\"v1\"".to_string()),
+        last_modified: None,
+        total: Some(total),
+        algo: HashAlgo::Sha1,
+        hash: hash.to_string(),
+        saved_at_ms: 1,
+        chunks: Some(chunks),
+    }
+}
+
+/// 3/4 段落完就失败，第二次续传完成 ⇒ 服务端总共只该吐 ≈1.0×，而不是 ≈1.75×。
+#[tokio::test]
+async fn resumes_the_finished_chunks_after_a_partial_failure() {
+    let payload = payload();
+    let size = payload.len() as u64;
+    let mut script = partial_script(&payload);
+    // 第二次尝试只该有 1 段要补。脚本仍给足 4 条：万一实现退化成「从头下」，
+    // 那一次也要能下完——这样红的就是**字节判据**本身，而不是「脚本用尽 ⇒ 500」。
+    script.extend((0..4).map(|_| MockResponse::ok(payload.clone())));
+    let server = MockServer::start(script).await;
+    let url = server.url("/jdk.zip");
+    let dir = temp_dir("chunked-resume");
+    let engine = engine(&dir);
+    let dest = dir.join("jdk.zip");
+
+    let outcome = engine
+        .download(request(&url, &dest, &payload))
+        .await
+        .expect("续传后应当成功");
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+
+    // 主判据：服务端总共吐了多少字节（「最终 hash 对」抓不到「其实全重下了」）。
+    let sent = server.bytes_sent();
+    assert!(
+        sent < size * 3 / 2,
+        "两次尝试总共只该吐 ≈1.0× 字节，实得 {sent}（1.5×{size} = {}）：\
+         超过它说明第二次又把整份下了一遍",
+        size * 3 / 2
+    );
+    assert_eq!(server.hits(), 5, "第一次 4 段 + 第二次补 1 段");
+    assert_eq!(
+        outcome.resumed_from,
+        size * 3 / 4,
+        "3/4 的段应当从磁盘复用（剩下 1 段重抓）"
+    );
+    server.stop();
+}
+
+/// 上一条的变异证明：同一个失败点，只把续传关掉 ⇒ 两次尝试一共吐 ~1.75×。
+///
+/// 把实现改回「失败就 discard、下次从头下」时，上一条的 `< 1.5×` 会落到本用例的区间 ⇒ 必红。
+#[tokio::test]
+async fn without_resume_the_same_failure_downloads_again() {
+    let payload = payload();
+    let size = payload.len() as u64;
+    let mut script = partial_script(&payload);
+    // 第二次尝试要是从头下：4 段全部重抓。
+    script.extend((0..4).map(|_| MockResponse::ok(payload.clone())));
+    let server = MockServer::start(script).await;
+    let url = server.url("/jdk.zip");
+    let dir = temp_dir("chunked-noresume");
+    let engine = engine(&dir);
+    let dest = dir.join("jdk.zip");
+
+    let outcome = engine
+        .download(request(&url, &dest, &payload).with_resume(ResumePolicy::Never))
+        .await
+        .expect("不续传也要能下完");
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+    assert_eq!(outcome.resumed_from, 0, "策略 Never ⇒ 一个字节都不复用");
+
+    let sent = server.bytes_sent();
+    assert!(
+        sent > size * 3 / 2,
+        "不续传时服务端要吐出两遍的字节（实测 {sent}，应当 > {}）——\
+         这正是 P21 那条债的形态",
+        size * 3 / 2
+    );
+    assert_eq!(server.hits(), 8, "第一次 4 段 + 第二次整份 4 段");
+    server.stop();
+}
+
+/// 盘上的旧段被改过：不许把「记录里说好的字节」当成「盘上现在就是好的」直接拼进结果。
+#[tokio::test]
+async fn corrupted_old_chunks_are_never_stitched_in() {
+    let payload = payload();
+    let size = payload.len() as u64;
+    let mut script: Vec<MockResponse> = (0..4).map(|_| MockResponse::ok(payload.clone())).collect();
+    // 万一实现真去复用坏段，它还要再整份重下一遍才有机会成功（8 条够两种世界都下完）。
+    script.extend((0..4).map(|_| MockResponse::ok(payload.clone())));
+    let server = MockServer::start(script).await;
+    let url = server.url("/jdk.zip");
+    let dir = temp_dir("chunked-broken-part");
+    let engine = engine(&dir);
+    let dest = dir.join("jdk.zip");
+    let ranges = plan(size, 4);
+
+    // 旧版无主命名的中间文件（`PartFile::for_dest` 会收养它）。
+    let part = dir.join("jdk.zip.ullpart");
+    let meta_path = dir.join("jdk.zip.ullpart.ullmeta.json");
+    let mut on_disk = payload.clone();
+    on_disk[ranges[1].start as usize + 5] ^= 0xff;
+    std::fs::write(&part, &on_disk).expect("写旧中间文件失败");
+    // 元数据声称前 3 段都落完了（记的是**原始**字节的摘要），只有第 4 段没记。
+    chunk_meta(
+        &url,
+        size,
+        &sha1_hex(&payload),
+        ranges[..3]
+            .iter()
+            .map(|range| record_of(&payload, *range))
+            .collect(),
+    )
+    .save(&meta_path)
+    .expect("写旧元数据失败");
+
+    let outcome = engine
+        .download(request(&url, &dest, &payload))
+        .await
+        .expect("坏旧段被丢掉后应当重下成功");
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+    // 主判据：坏段自证不过 ⇒ 整份丢掉从头来，服务端只该吐这一份文件的字节。
+    assert_eq!(
+        server.bytes_sent(),
+        size,
+        "全部字节都要重新抓一遍（复用坏字节 = 交付坏文件）"
+    );
+    assert_eq!(server.hits(), 4, "从头下 = 4 个分块请求");
+    assert_eq!(
+        outcome.resumed_from, 0,
+        "任何一段对不上就整份丢掉，一段都不复用"
+    );
+    server.stop();
+}
+
+/// 同一个目标文件换了大小：旧的 2 段元数据**恰好**对得上新规划的前两段，只有「总长变了」能拦住它。
+#[tokio::test]
+async fn changed_size_invalidates_old_chunk_layout() {
+    let payload = payload();
+    let size = payload.len() as u64;
+    let mut script: Vec<MockResponse> = (0..4).map(|_| MockResponse::ok(payload.clone())).collect();
+    // 真去复用旧段时，它还要再整份重下一遍才有机会成功（8 条够两种世界都下完）。
+    script.extend((0..4).map(|_| MockResponse::ok(payload.clone())));
+    let server = MockServer::start(script).await;
+    let url = server.url("/jdk.zip");
+    let dir = temp_dir("chunked-size-change");
+    let engine = engine(&dir);
+    let dest = dir.join("jdk.zip");
+
+    // 旧中间文件 = 上一次「同一个目标、一半大小」的下载残留（内容是**另一个版本**的字节）。
+    let old_size = SIZE / 2;
+    let old_bytes = vec![0xEE_u8; old_size];
+    let part = dir.join("jdk.zip.ullpart");
+    let meta_path = dir.join("jdk.zip.ullpart.ullmeta.json");
+    std::fs::write(&part, &old_bytes).expect("写旧中间文件失败");
+    // 2 段 × 10KiB：段区间与新规划的 [0-10239] / [10240-20479] 完全重合，所以
+    // 「段对不上号」这条拦不住它，只有总长比较能拦（变异：删掉总长比较 ⇒ 本用例红：
+    // 复用 20480 字节 + 整文件校验失败 ⇒ 白下一次 20480 字节）。
+    chunk_meta(
+        &url,
+        old_size as u64,
+        &sha1_hex(&payload),
+        plan(old_size as u64, 2)
+            .iter()
+            .map(|range| record_of(&old_bytes, *range))
+            .collect(),
+    )
+    .save(&meta_path)
+    .expect("写旧元数据失败");
+
+    let outcome = engine
+        .download(request(&url, &dest, &payload))
+        .await
+        .expect("大小变了以后必须从头重下成功");
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+    // 主判据：旧文件的那 20480 字节一个都不许用（误用 ⇒ 这次会吐出 1.5×）。
+    assert_eq!(
+        server.bytes_sent(),
+        size,
+        "旧的那 20480 字节一个都不许用：只允许下这次的 {size} 字节"
+    );
+    assert_eq!(outcome.resumed_from, 0, "总长变了 ⇒ 旧段偏移全部作废");
+    server.stop();
+}
+
+/// 同一个 URL、**换了一份表示**（复审 r2 p2-1）：上一份表示的校验器不许搬进这一轮。
+///
+/// 链条（全走既有代码）：上一轮留下 `url=U, 旧总长/旧段区间, etag="v1"`（`chunk_meta` 固定写
+/// `"v1"`）⇒ 本轮因 `HashChanged` 判 `Restart`、`part.discard()`，但 `previous` 还在内存里 ⇒
+/// 若把 `"v1"` 写进**新一轮**元数据，而 `record` 又只在缺值时写，那本轮落的 3 段（**新**内容）
+/// 配的是上一份表示的校验器 ⇒ 重试拿 `"v1"` 发 `If-Range` ⇒ 此刻服务器是 `"v2"`，按 RFC 7233
+/// 回 200 + 完整正文 ⇒ `RangeUnsupported` ⇒ 丢整份 ⇒ 单连接从头下。
+/// 实测那是 113664 字节（2.775×），比「干脆不续传」的 72704（1.775×）还多一整份。
+///
+/// 修法：`carry = if done.is_empty() { None } else { same_source }` —— `done` 为空就没有任何
+/// 字节要接着用，旧校验器描述的不是这次要下的东西；本轮响应里的真实校验器由 `record` 填进来。
+///
+/// 断言：重试带的 `If-Range` 必须是本轮响应给的 `"v2"`（出现 `"v1"` 即串台），且字节数落回
+/// 41984（1.025×）而不是 113664。
+#[tokio::test]
+async fn stale_validator_from_the_previous_representation_is_not_carried() {
+    /// 服务器此刻那份表示的 ETag（旧元数据里的 `"v1"` 是上一份表示的）。
+    const ETAG_NEW: &str = "\"v2\"";
+
+    let payload = payload();
+    let size = payload.len() as u64;
+    // 4 段：3 段成功 + 1 段卡住；重试只该补 1 段；后面 4 条给「退化成单连接整份重下」兜底
+    // （那种坏结局也要能下完，红的是判据不是脚本用尽）。
+    let fresh = |body: Vec<u8>| MockResponse::ok(body).with_etag(ETAG_NEW);
+    let mut script: Vec<MockResponse> = (0..3).map(|_| fresh(payload.clone())).collect();
+    script.push(fresh(payload.clone()).stalling_after(STALL_TAIL));
+    script.push(fresh(payload.clone()));
+    script.extend((0..4).map(|_| fresh(payload.clone())));
+    let server = MockServer::start(script).await;
+    let url = server.url("/jdk.zip");
+    let dir = temp_dir("chunked-stale-validator");
+    let engine = engine(&dir);
+    let dest = dir.join("jdk.zip");
+
+    // 上一轮留下的中间文件 + 元数据：同一个 URL，但描述的是另一份表示（旧总长 / 旧段区间 /
+    // 旧 ETag / 旧摘要）。
+    let old_size = SIZE / 2;
+    let old_bytes = vec![0xEE_u8; old_size];
+    let part = dir.join("jdk.zip.ullpart");
+    let meta_path = dir.join("jdk.zip.ullpart.ullmeta.json");
+    std::fs::write(&part, &old_bytes).expect("写旧中间文件失败");
+    chunk_meta(
+        &url,
+        old_size as u64,
+        &sha1_hex(&old_bytes),
+        plan(old_size as u64, 2)
+            .iter()
+            .map(|range| record_of(&old_bytes, *range))
+            .collect(),
+    )
+    .save(&meta_path)
+    .expect("写旧元数据失败");
+
+    let outcome = engine
+        .download(request(&url, &dest, &payload))
+        .await
+        .expect("换了表示之后仍应能下完");
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+    assert_eq!(
+        server.if_range_values(),
+        vec![ETAG_NEW.to_string()],
+        "重试只许带本轮响应给的校验器；出现 \"v1\" 说明上一份表示的校验器被搬进了这一轮"
+    );
+    // 3 段 + 卡住的 1KiB + 补 1 段 = 整份文件 + 1KiB = 41984（1.025×）。
+    assert_eq!(
+        server.bytes_sent(),
+        size + STALL_TAIL as u64,
+        "应当是 41984（1.025×）；113664（2.775×）说明陈旧校验器把整份顶回来又重下了一遍"
+    );
+    assert_eq!(server.hits(), 5, "第一次 4 段 + 第二次补 1 段");
+    assert_eq!(outcome.resumed_from, size * 3 / 4, "复用了本轮的 3 段");
+    server.stop();
+}

crates/core/tests/download/chunked/switch.rs

@@ -0,0 +1,232 @@
+//! 换源（复审 r1 **p2-1**）：源 A 的校验器**不许**被带到源 B 上。
+//!
+//! 复审发现的链条：A 落 3/4 段后失败 ⇒ 元数据 = `url=A` + A 的 ETag；引擎换到 B ⇒ `decide` 判
+//! `UrlChanged`、删盘重来，但 `ChunkJournal::begin` 把 **A 的校验器**原样写进 `url=B` 的新元数据；
+//! `record` 又只在 `etag.is_none()` 时写入 ⇒ B 的真实校验器永远盖不上。于是 B 的重试拿 A 的 ETag
+//! 发 `If-Range`：按 RFC 7233 §3.2，服务器对不上就回 200 + 完整正文 ⇒ 分块路径判 `RangeUnsupported`
+//! ⇒ 丢掉整份 ⇒ 单连接从头下。债务以另一条路径复现，正是本 mission 要消灭的「从 0 重下」。
+//!
+//! 修法 = **校验器与记下它的 URL 绑定**（复审给的选项 A）：只有 `previous.url == candidate.url`
+//! 才沿用，否则当「没有校验器」，让新源自己的响应去填（`ChunkJournal::record`）。
+//!
+//! 判据都看**线上行为**，不看实现细节：
+//! * 源 B 收到的每一个 `If-Range` 都必须是 B 自己的 ETag（A 的一个都不许出现）；
+//! * 源 B 自己的字节数仍是 1.025×（换源之后 B 的同源续传照走，没被这一修打回「整份重下」）；
+//! * 最终文件逐字节正确。
+//!
+//! 变异（把 URL 过滤去掉）⇒ 第一条红：`if_range_values` 变成 A 的 `"src-a"`，且夹具按 RFC 7233
+//! 回 200 ⇒ 分块判 `RangeUnsupported` ⇒ 退化单连接 ⇒ 字节数也涨（红原文见 mission 报告）。
+
+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, PrefixRule};
+use unlucky_core::download::region::RegionProfile;
+use unlucky_core::download::types::{DownloadRequest, ResourceKind};
+
+use super::support::{sha1_hex, temp_dir, MockResponse, MockServer};
+
+/// 4 段 × 10 KiB。
+const SIZE: usize = 40 * 1024;
+const CHUNK: u64 = 10 * 1024;
+/// 卡住的那一段吐这么多字节就停（落盘但不会被记账）。
+const STALL_TAIL: usize = 1024;
+/// 两台「服务器」各自的校验器：故意取不同的值，任何串台都能一眼看出来。
+const ETAG_A: &str = "\"src-a\"";
+const ETAG_B: &str = "\"src-b\"";
+
+fn payload() -> Vec<u8> {
+    (0..SIZE).map(|index| (index % 251) as u8).collect()
+}
+
+/// 一次尝试允许 2 次：源 A 用完 2 次出局 → 换源 B；源 B 也是 2 次（失败一次 + 续传一次）。
+fn config() -> EngineConfig {
+    EngineConfig {
+        max_attempts_per_source: 2,
+        backoff: BackoffPolicy::none(),
+        connect_timeout: Duration::from_secs(2),
+        response_timeout: Duration::from_secs(2),
+        read_timeout: Duration::from_millis(300),
+        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()
+    }
+}
+
+/// 「3 段成功 + 1 段卡住」的 4 条脚本（等于一次部分失败的分块尝试）。
+fn partial(etag: &str, payload: &[u8]) -> Vec<MockResponse> {
+    let mut script: Vec<MockResponse> = (0..3)
+        .map(|_| MockResponse::ok(payload.to_vec()).with_etag(etag))
+        .collect();
+    script.push(
+        MockResponse::ok(payload.to_vec())
+            .with_etag(etag)
+            .stalling_after(STALL_TAIL),
+    );
+    script
+}
+
+#[tokio::test]
+async fn switching_source_never_reuses_the_previous_source_validator() {
+    let payload = payload();
+    let size = payload.len() as u64;
+
+    // 源 A：部分失败一次，同源续传那一次直接 500 ⇒ 源 A 出局（引擎按候选顺序换到 B）。
+    let mut script_a = partial(ETAG_A, &payload);
+    script_a.push(MockResponse::status(500));
+    let server_a = MockServer::start(script_a).await;
+
+    // 源 B：同样部分失败一次（于是 B 有自己的已落盘分块），续传那一段就该带 **B 自己的** 校验器。
+    // 后面多给 4 条：万一实现退化成「整份重下」，那次也要能下完——红的是判据，而不是脚本用尽。
+    let mut script_b = partial(ETAG_B, &payload);
+    script_b.push(MockResponse::ok(payload.clone()).with_etag(ETAG_B));
+    script_b.extend((0..4).map(|_| MockResponse::ok(payload.clone()).with_etag(ETAG_B)));
+    let server_b = MockServer::start(script_b).await;
+
+    // 两台服务器当两个候选源：官方 = A，镜像规则把前缀改写成 B。
+    let url = server_a.url("/jdk.zip");
+    let mut table = MirrorTable::builtin();
+    table.add_custom_rule(PrefixRule::new(
+        "test-mirror",
+        format!("{}/", server_a.url("")),
+        format!("{}/", server_b.url("")),
+    ));
+    let dir = temp_dir("chunked-switch");
+    let engine = DownloadEngine::new(
+        config(),
+        Arc::new(table),
+        Arc::new(RegionProfile::direct(0)),
+        Arc::new(HealthRegistry::in_memory()),
+        Arc::new(ContentCache::open(dir.join("cache"))),
+    );
+    let dest = dir.join("jdk.zip");
+
+    let outcome = engine
+        .download(
+            DownloadRequest::new(ResourceKind::JavaRuntime, url.as_str(), &dest)
+                .with_sha1(sha1_hex(&payload))
+                .with_size(size),
+        )
+        .await
+        .expect("换源后应当能下完");
+
+    assert_eq!(
+        std::fs::read(&dest).expect("读回落盘文件失败"),
+        payload,
+        "换源 + 续传之后文件必须逐字节正确"
+    );
+    // 主判据：源 B 只许看到**它自己**的校验器。若这里出现 `"src-a"`，就是复审 r1 p2-1 的债
+    // 从另一条路回来了（真实服务器会对不上回 200 ⇒ 丢掉整份 ⇒ 单连接从头下）。
+    assert_eq!(
+        server_b.if_range_values(),
+        vec![ETAG_B.to_string()],
+        "源 B 收到了不属于它的校验器：换源后旧源的 ETag 被带过来了"
+    );
+    // 对照：源 A 的同源续传本来就该带自己的校验器（证明这个观测点不是恒空）。
+    assert_eq!(
+        server_a.if_range_values(),
+        vec![ETAG_A.to_string()],
+        "源 A 的同源续传应当带 A 自己的校验器"
+    );
+    // 换源之后 B 自己的那一轮仍是「同源续传」：3 段 + 1KiB + 补 1 段 = 1.025×。
+    assert_eq!(
+        server_b.bytes_sent(),
+        3 * CHUNK + STALL_TAIL as u64 + CHUNK,
+        "源 B 的字节数应当是 1.025×（3 段 + 卡住的 1KiB + 补 1 段），不是整份重下"
+    );
+    assert_eq!(server_b.hits(), 5, "B：一次部分失败（4 请求）+ 一次补 1 段");
+    assert_eq!(
+        outcome.resumed_from,
+        3 * CHUNK,
+        "最终那一次复用了 3 段（补抓 1 段），不是从头下"
+    );
+
+    server_a.stop();
+    server_b.stop();
+}
+
+/// 同一个换源场景，但**新源自己没有任何校验器**（复审 r1 建议的那条用例）。
+///
+/// 语义是刻意的保守：换了源、新源又不给 ETag/Last-Modified ⇒ 我们无法让服务器回答「内容变过
+/// 没有」，`chunk_resume::decide` 判 `NoValidator` ⇒ 新源那次重试**整份重下**，不复用自己刚
+/// 落的 3 段。代价（多下一份）换的是「绝不拿无法证明的旧字节去拼文件」，与本模块的立场一致。
+///
+/// 这条同时是「修法不要退化成沿用旧校验器」的守卫：只要有人把 `same_source` 过滤去掉、或把
+/// `record` 改成无条件覆盖，B 的元数据就会带上 A 的 `"src-a"` ⇒ 这里会出现 `If-Range` ⇒ 红。
+#[tokio::test]
+async fn switched_source_without_its_own_validator_restarts_from_zero() {
+    let payload = payload();
+    let size = payload.len() as u64;
+
+    // 源 A：有 ETag，部分失败后同源续传吃 500 ⇒ 出局。
+    let mut script_a = partial(ETAG_A, &payload);
+    script_a.push(MockResponse::status(500));
+    let server_a = MockServer::start(script_a).await;
+
+    // 源 B：**完全不发 ETag**。部分失败一次（于是 B 有 3 段已落盘），重试时不许复用。
+    let mut script_b: Vec<MockResponse> = (0..3)
+        .map(|_| MockResponse::ok(payload.clone()).without_etag())
+        .collect();
+    script_b.push(
+        MockResponse::ok(payload.clone())
+            .without_etag()
+            .stalling_after(STALL_TAIL),
+    );
+    script_b.extend((0..4).map(|_| MockResponse::ok(payload.clone()).without_etag()));
+    let server_b = MockServer::start(script_b).await;
+
+    let url = server_a.url("/jdk.zip");
+    let mut table = MirrorTable::builtin();
+    table.add_custom_rule(PrefixRule::new(
+        "test-mirror",
+        format!("{}/", server_a.url("")),
+        format!("{}/", server_b.url("")),
+    ));
+    let dir = temp_dir("chunked-switch-novalidator");
+    let engine = DownloadEngine::new(
+        config(),
+        Arc::new(table),
+        Arc::new(RegionProfile::direct(0)),
+        Arc::new(HealthRegistry::in_memory()),
+        Arc::new(ContentCache::open(dir.join("cache"))),
+    );
+    let dest = dir.join("jdk.zip");
+
+    let outcome = engine
+        .download(
+            DownloadRequest::new(ResourceKind::JavaRuntime, url.as_str(), &dest)
+                .with_sha1(sha1_hex(&payload))
+                .with_size(size),
+        )
+        .await
+        .expect("新源没有校验器时也要能下完（宁可重下）");
+
+    assert_eq!(std::fs::read(&dest).expect("读回落盘文件失败"), payload);
+    assert!(
+        server_b.if_range_values().is_empty(),
+        "新源没有任何校验器，就不许发条件请求（带 A 的值更是串源）：{:?}",
+        server_b.if_range_values()
+    );
+    assert_eq!(
+        server_b.bytes_sent(),
+        3 * CHUNK + STALL_TAIL as u64 + 4 * CHUNK,
+        "B 的第二次尝试是整份重下（无法证明旧字节 ⇒ 不复用）"
+    );
+    assert_eq!(server_b.hits(), 8, "B：部分失败 4 请求 + 整份重下 4 请求");
+    assert_eq!(outcome.resumed_from, 0, "最终那一次从头下");
+
+    server_a.stop();
+    server_b.stop();
+}

crates/core/tests/download/chunked_test.rs

@@ -377,3 +377,15 @@ async fn rejects_lying_content_range_and_falls_back() {
 /// 分块作业的两道失速判据（无进度 / 总期限）：拆到子模块守 400 行硬线。
 #[path = "chunked/timeout.rs"]
 mod timeout;
+
+/// 分块路径的续传（失败不重下、旧字节要自证）：同族用例单独一份。
+#[path = "chunked/resume.rs"]
+mod resume;
+
+/// 换源后旧源校验器不许带过去（复审 r1 p2-1）：同样单独一份守 400 行硬线。
+#[path = "chunked/switch.rs"]
+mod switch;
+
+/// 弱校验器（Last-Modified）驱动 `If-Range` 的支路（复审 r2 p3-1）：单独一份。
+#[path = "chunked/last_modified.rs"]
+mod last_modified;

crates/core/tests/download/engine_mock_test.rs

@@ -278,6 +278,7 @@ async fn resumes_from_existing_part_file() {
         algo: HashAlgo::Sha1,
         hash: sha1_hex(&payload),
         saved_at_ms: 1,
+        chunks: None,
     }
     .save(&meta_path)
     .expect("写续传元数据失败");

crates/core/tests/download/resume/chunk.rs

@@ -0,0 +1,242 @@
+//! 分块路径的续传判定（`chunk_resume`，M293）。
+//!
+//! 分块路径不能复用单连接那套「长度 = 已下」的判定：它的中间文件被 `set_len(total)` 预分配过，
+//! 长度恒等于总长（`resume::decide` 会判成 `Complete`，然后整文件校验把一份没下完的文件判死）。
+//! 这里钉住的是**另一套条件**：URL / 总长 / 摘要 / 校验器全部对上，而且每条记录都要精确命中
+//! 本次分块规划里的一段；过了判定还要能按记录重算出同样的摘要（`verify_records`），
+//! 否则一律从头再来。
+
+use unlucky_core::download::chunk_resume::{
+    decide as chunk_decide, remaining, verify_records, ChunkDecision,
+};
+use unlucky_core::download::chunked::{plan, ByteRange};
+use unlucky_core::download::resume::{ChunkRecord, RestartReason, ResumeMeta};
+use unlucky_core::download::types::{HashAlgo, ResumePolicy};
+use unlucky_core::download::verify::digest_bytes;
+
+use super::{meta, HASH, URL};
+
+fn chunk_payload(len: usize) -> Vec<u8> {
+    (0..len).map(|index| (index % 251) as u8).collect()
+}
+
+fn record(payload: &[u8], range: ByteRange) -> ChunkRecord {
+    let bytes = &payload[range.start as usize..=range.end as usize];
+    ChunkRecord {
+        start: range.start,
+        end: range.end,
+        sha1: digest_bytes(HashAlgo::Sha1, bytes),
+    }
+}
+
+fn with_chunks(base: ResumeMeta, chunks: Vec<ChunkRecord>) -> ResumeMeta {
+    ResumeMeta {
+        chunks: Some(chunks),
+        ..base
+    }
+}
+
+/// 判定「应当从头再来」，返回原因。
+fn chunk_restart(
+    part_len: u64,
+    meta: Option<&ResumeMeta>,
+    url: &str,
+    total: u64,
+    planned: &[ByteRange],
+) -> RestartReason {
+    match chunk_decide(
+        ResumePolicy::Conservative,
+        part_len,
+        meta,
+        url,
+        Some((HashAlgo::Sha1, HASH)),
+        total,
+        planned,
+    ) {
+        ChunkDecision::Restart { reason } => reason,
+        other => panic!("应当判从头再来，实得 {other:?}"),
+    }
+}
+
+#[test]
+fn chunk_resume_reuses_only_segments_that_match_the_plan() {
+    let payload = chunk_payload(1_000);
+    let planned = plan(1_000, 4);
+    let done = vec![record(&payload, planned[0]), record(&payload, planned[2])];
+    let meta = with_chunks(meta(URL), done.clone());
+
+    let decision = chunk_decide(
+        ResumePolicy::Conservative,
+        1_000,
+        Some(&meta),
+        URL,
+        Some((HashAlgo::Sha1, HASH)),
+        1_000,
+        &planned,
+    );
+    assert_eq!(
+        decision,
+        ChunkDecision::Reuse { done: done.clone() },
+        "身份全部对上时应复用，且只复用记录过的段"
+    );
+    assert_eq!(
+        remaining(&planned, &done),
+        vec![planned[1], planned[3]],
+        "本次要抓的应当是规划里没落盘的那几段"
+    );
+
+    // 全部落完 ⇒ 没有要抓的段（下一次尝试只做整文件校验）。
+    let all: Vec<ChunkRecord> = planned
+        .iter()
+        .map(|range| record(&payload, *range))
+        .collect();
+    assert!(remaining(&planned, &all).is_empty());
+}
+
+#[test]
+fn chunk_resume_restart_matrix() {
+    let planned = plan(1_000, 4);
+    let good = with_chunks(
+        meta(URL),
+        vec![ChunkRecord {
+            start: planned[0].start,
+            end: planned[0].end,
+            sha1: HASH.to_string(),
+        }],
+    );
+
+    // 策略禁止续传。
+    assert_eq!(
+        match chunk_decide(
+            ResumePolicy::Never,
+            1_000,
+            Some(&good),
+            URL,
+            Some((HashAlgo::Sha1, HASH)),
+            1_000,
+            &planned
+        ) {
+            ChunkDecision::Restart { reason } => reason,
+            other => panic!("实得 {other:?}"),
+        },
+        RestartReason::PolicyNever
+    );
+    // 没有中间文件 / 没有元数据 / 元数据里没有分块清单（单连接写的那种）。
+    assert_eq!(
+        chunk_restart(0, Some(&good), URL, 1_000, &planned),
+        RestartReason::NoPart
+    );
+    assert_eq!(
+        chunk_restart(1_000, None, URL, 1_000, &planned),
+        RestartReason::NoMeta
+    );
+    assert_eq!(
+        chunk_restart(1_000, Some(&meta(URL)), URL, 1_000, &planned),
+        RestartReason::NoMeta,
+        "没有分块清单的元数据证明不了乱序段各自到哪"
+    );
+    // 换了源 / 换了目标文件（摘要变）/ 换了大小。
+    assert_eq!(
+        chunk_restart(
+            1_000,
+            Some(&good),
+            "https://other.example/x.jar",
+            1_000,
+            &planned
+        ),
+        RestartReason::UrlChanged
+    );
+    let changed_hash = ResumeMeta {
+        hash: "f".repeat(40),
+        ..good.clone()
+    };
+    assert_eq!(
+        chunk_restart(1_000, Some(&changed_hash), URL, 1_000, &planned),
+        RestartReason::HashChanged
+    );
+    assert_eq!(
+        chunk_restart(1_000, Some(&good), URL, 2_000, &plan(2_000, 4)),
+        RestartReason::SizeChanged,
+        "总长变了就是另一个文件，旧段的偏移全部作废"
+    );
+    // 没有校验器：服务器无法回答「内容变了没有」。
+    let weak = ResumeMeta {
+        etag: None,
+        last_modified: None,
+        ..good.clone()
+    };
+    assert_eq!(
+        chunk_restart(1_000, Some(&weak), URL, 1_000, &planned),
+        RestartReason::NoValidator
+    );
+    // 段边界漂移：半段、重复、越界都不行。
+    let half = with_chunks(
+        meta(URL),
+        vec![ChunkRecord {
+            start: 0,
+            end: 100,
+            sha1: HASH.to_string(),
+        }],
+    );
+    assert_eq!(
+        chunk_restart(1_000, Some(&half), URL, 1_000, &planned),
+        RestartReason::ChunkLayout,
+        "记录必须是规划里原样的一段"
+    );
+    let first = ChunkRecord {
+        start: planned[0].start,
+        end: planned[0].end,
+        sha1: HASH.to_string(),
+    };
+    let duplicated = with_chunks(meta(URL), vec![first.clone(), first]);
+    assert_eq!(
+        chunk_restart(1_000, Some(&duplicated), URL, 1_000, &planned),
+        RestartReason::ChunkLayout
+    );
+    // 换了连接数 ⇒ 规划变了 ⇒ 旧段对不上号（从头再来，而不是硬拼）。
+    let reconfigured = with_chunks(
+        meta(URL),
+        vec![ChunkRecord {
+            start: planned[0].start,
+            end: planned[0].end,
+            sha1: HASH.to_string(),
+        }],
+    );
+    assert_eq!(
+        chunk_restart(1_000, Some(&reconfigured), URL, 1_000, &plan(1_000, 8)),
+        RestartReason::ChunkLayout
+    );
+}
+
+#[test]
+fn chunk_verify_rejects_bytes_that_do_not_match_the_record() {
+    let dir = std::env::temp_dir().join(format!("ull-chunk-verify-{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).unwrap();
+    let path = dir.join("jdk.zip.ullpart");
+
+    let payload = chunk_payload(1_000);
+    let planned = plan(1_000, 4);
+    let records = vec![record(&payload, planned[0]), record(&payload, planned[1])];
+    std::fs::write(&path, &payload).expect("写中间文件失败");
+    assert_eq!(
+        verify_records(&path, &records).expect("字节与记录一致时必须通过"),
+        planned[0].len() + planned[1].len(),
+        "返回的是可复用的字节数"
+    );
+
+    // 盘上那一段被改过（断电半截写 / 坏盘 / 别的源写进来的字节）。
+    let mut broken = payload.clone();
+    broken[planned[1].start as usize + 3] ^= 0xff;
+    std::fs::write(&path, &broken).expect("写中间文件失败");
+    assert!(
+        verify_records(&path, &records).is_err(),
+        "摘要对不上必须判「证明不了」，不能拿旧字节当好的"
+    );
+
+    // 盘上根本不够长（记录说写完了，其实只有开头几个字节）。
+    std::fs::write(&path, &payload[..10]).expect("写中间文件失败");
+    assert!(verify_records(&path, &records).is_err());
+
+    let _ = std::fs::remove_dir_all(&dir);
+}

crates/core/tests/download/resume_test.rs

@@ -5,6 +5,10 @@ use unlucky_core::download::resume::{
 };
 use unlucky_core::download::types::{HashAlgo, ResumePolicy};
 
+/// 分块路径的续传判定与旧字节自证（M293）：拆到子模块守 400 行硬线。
+#[path = "resume/chunk.rs"]
+mod chunk;
+
 const URL: &str = "https://bmclapi2.bangbang93.com/versions/1.20.1/client.jar";
 const HASH: &str = "0123456789abcdef0123456789abcdef01234567";
 
@@ -17,6 +21,7 @@ fn meta(url: &str) -> ResumeMeta {
         algo: HashAlgo::Sha1,
         hash: HASH.to_string(),
         saved_at_ms: 1,
+        chunks: None,
     }
 }
 

crates/core/tests/download/support/mod.rs

@@ -1,17 +1,20 @@
 //! 测试支撑：临时目录、摘要工具、进程内极简 HTTP 服务器（零额外依赖）。
 //!
 //! 为什么自己写 mock server 而不用 wiremock：P0 阶段控制依赖面（Win7 Tier-3 构建 +
-//! 单文件体积），一个 60 行的 `TcpListener` 足够表达「成功 / 500 / 校验不符 / 无视 Range」。
+//! 单文件体积），一个 `TcpListener` + 脚本队列足够表达「成功 / 500 / 校验不符 / 无视 Range」。
+//! 服务器本体拆在 [`server`] 里（守 400 行硬线），这里只做再导出，调用方路径不变。
 
 #![allow(dead_code)]
 
-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};
+mod response;
+mod server;
+
+#[allow(unused_imports)]
+pub use self::response::MockResponse;
+#[allow(unused_imports)]
+pub use self::server::MockServer;
 
 /// 断言用的小工具：唯一临时目录（进程内计数 + 时间戳，避免并行测试互相踩）。
 pub fn temp_dir(tag: &str) -> std::path::PathBuf {
@@ -61,328 +64,3 @@ pub fn age_file(path: &std::path::Path, age: std::time::Duration) {
         .set_times(times)
         .expect("改 mtime 失败");
 }
-
-/// mock 服务器给一次请求的应答。
-#[derive(Debug, Clone)]
-pub struct MockResponse {
-    /// HTTP 状态码。
-    pub status: u16,
-    /// 响应体。
-    pub body: Vec<u8>,
-    /// 是否声明 `Accept-Ranges: bytes` 并在收到 `Range` 时真的切段。
-    pub range_support: bool,
-    /// `ETag`（续传测试用）。
-    pub etag: Option<String>,
-    /// 返回 206 时故意把 `Content-Range` 写成从 0 开始（模拟段错位的坏服务器）。
-    pub lie_content_range: bool,
-    /// 读完请求头后**一个字节都不写**（服务端接了连接但永不回话）。
-    ///
-    /// 这是「无界 `await`」唯一能表达成用例的形态：读方不设上界就永远不返回。
-    pub silent: bool,
-    /// 写出这么多字节正文后就**停住不再写**（中途卡住），连接保持打开。
-    pub stall_at: Option<usize>,
-    /// 正文实际只发这么多字节（头部仍声称全量 = 短包）。
-    pub send_len: Option<usize>,
-    /// 正文之后**多**发这么多字节，且故意不发 `Content-Length`（超长包：HTTP/1.1 靠关连接分帧）。
-    pub trailing_extra: Option<usize>,
-    /// 正文按 `piece` 字节一片、片间睡 `gap` 滴流（慢但持续，用来证明超时不会误杀它）。
-    pub drip: Option<(usize, Duration)>,
-}
-
-/// 夹具「挂住」的时长：远大于用例自己的上界，保证读方只能靠超时脱身。
-const HOLD: Duration = Duration::from_secs(30);
-
-impl MockResponse {
-    /// 200 + 完整内容（支持 Range）。
-    pub fn ok(body: impl Into<Vec<u8>>) -> Self {
-        Self {
-            status: 200,
-            body: body.into(),
-            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,
-        }
-    }
-
-    /// 固定状态码 + 空体。
-    pub fn status(status: u16) -> Self {
-        Self {
-            status,
-            body: Vec::new(),
-            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;
-        self
-    }
-
-    /// 206 但 `Content-Range` 谎报（模拟段错位的坏 CDN）。
-    pub fn with_wrong_content_range(mut self) -> Self {
-        self.lie_content_range = true;
-        self
-    }
-}
-
-/// 进程内极简 HTTP/1.1 服务器：按脚本顺序返回响应，脚本用尽后一律 500。
-pub struct MockServer {
-    addr: SocketAddr,
-    hits: Arc<AtomicUsize>,
-    handle: tokio::task::JoinHandle<()>,
-}
-
-impl MockServer {
-    /// 启动服务器（脚本是「每次请求一个响应」的队列）。
-    pub async fn start(script: Vec<MockResponse>) -> Self {
-        let listener = TcpListener::bind("127.0.0.1:0")
-            .await
-            .expect("绑定端口失败");
-        let addr = listener.local_addr().expect("取本地地址失败");
-        let hits = Arc::new(AtomicUsize::new(0));
-        let script = Arc::new(std::sync::Mutex::new(script));
-        let counter = Arc::clone(&hits);
-
-        let handle = tokio::spawn(async move {
-            loop {
-                let Ok((stream, _)) = listener.accept().await else {
-                    break;
-                };
-                let script = Arc::clone(&script);
-                let counter = Arc::clone(&counter);
-                tokio::spawn(async move {
-                    let index = counter.fetch_add(1, Ordering::Relaxed);
-                    let response = {
-                        let script = script.lock().unwrap_or_else(|err| err.into_inner());
-                        script.get(index).cloned()
-                    };
-                    let response = response.unwrap_or_else(|| MockResponse::status(500));
-                    let _ = serve(stream, response).await;
-                });
-            }
-        });
-
-        Self { addr, hits, handle }
-    }
-
-    /// 第 `index` 个请求该用的 URL（换成不同路径可模拟「不同源」）。
-    pub fn url(&self, path: &str) -> String {
-        format!("http://{}{}", self.addr, path)
-    }
-
-    /// 换 host 拼 URL（用 `localhost` vs `127.0.0.1` 造出两个不同源）。
-    pub fn url_with(&self, host: &str, path: &str) -> String {
-        format!("http://{host}:{}{}", self.addr.port(), path)
-    }
-
-    /// 端口号。
-    pub fn port(&self) -> u16 {
-        self.addr.port()
-    }
-
-    /// 已收到的请求数。
-    pub fn hits(&self) -> usize {
-        self.hits.load(Ordering::Relaxed)
-    }
-
-    /// 关停。
-    pub fn stop(self) {
-        self.handle.abort();
-    }
-}
-
-async fn serve(mut stream: TcpStream, response: MockResponse) -> std::io::Result<()> {
-    let mut buffer = Vec::new();
-    let mut chunk = [0_u8; 1024];
-    loop {
-        let read = stream.read(&mut chunk).await?;
-        if read == 0 {
-            break;
-        }
-        buffer.extend_from_slice(&chunk[..read]);
-        if buffer.windows(4).any(|window| window == b"\r\n\r\n") {
-            break;
-        }
-    }
-    let request = String::from_utf8_lossy(&buffer).to_ascii_lowercase();
-    let range = request
-        .lines()
-        .find_map(|line| line.strip_prefix("range: bytes="))
-        .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 => {
-            // 尊重请求里的结束偏移（分块下载必须逐段精确，否则组装后长度对不上）。
-            let end = requested_end.unwrap_or(total - 1).min(total - 1);
-            let slice = response.body[start as usize..=end as usize].to_vec();
-            // 谎报模式：body 仍是正确的那一段，但头部声称从 0 开始（长度恰好 → 只有
-            // 严格比对 Content-Range 才抓得住）。
-            let (claimed_start, claimed_end) = if response.lie_content_range {
-                (0, slice.len().saturating_sub(1) as u64)
-            } else {
-                (start, end)
-            };
-            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,
-            vec![
-                format!("Content-Range: bytes */{total}"),
-                "Content-Length: 0".to_string(),
-            ],
-            Vec::new(),
-        ),
-        _ => {
-            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());
-            }
-            (response.status, headers, response.body.clone())
-        }
-    };
-    let mut head = format!("HTTP/1.1 {status} {}\r\n", reason(status));
-    if let Some(etag) = &response.etag {
-        head.push_str(&format!("ETag: {etag}\r\n"));
-    }
-    for header in headers {
-        head.push_str(&header);
-        head.push_str("\r\n");
-    }
-    head.push_str("Connection: close\r\n\r\n");
-    stream.write_all(head.as_bytes()).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=<start>-<end>`（`end` 可缺省 = 到文件末尾）。
-fn parse_range_header(value: &str) -> Option<(u64, Option<u64>)> {
-    let (start, end) = value.split_once('-')?;
-    let start = start.trim().parse::<u64>().ok()?;
-    let end = end.trim();
-    let end = if end.is_empty() {
-        None
-    } else {
-        Some(end.parse::<u64>().ok()?)
-    };
-    Some((start, end))
-}
-
-fn reason(status: u16) -> &'static str {
-    match status {
-        200 => "OK",
-        206 => "Partial Content",
-        404 => "Not Found",
-        416 => "Range Not Satisfiable",
-        500 => "Internal Server Error",
-        503 => "Service Unavailable",
-        _ => "Status",
-    }
-}

crates/core/tests/download/support/response.rs

@@ -0,0 +1,132 @@
+//! 夹具的一次应答描述（状态码 / 正文 / Range 支持 / 校验器 / 各种坏形态）。
+//!
+//! 从 `support/server.rs` 拆出只为守 400 行硬线（M293 r1：给夹具加上 `If-Range` 观测之后）。
+
+use std::time::Duration;
+
+/// mock 服务器给一次请求的应答。
+#[derive(Debug, Clone)]
+pub struct MockResponse {
+    /// HTTP 状态码。
+    pub status: u16,
+    /// 响应体。
+    pub body: Vec<u8>,
+    /// 是否声明 `Accept-Ranges: bytes` 并在收到 `Range` 时真的切段。
+    pub range_support: bool,
+    /// `ETag`（续传测试用）。
+    pub etag: Option<String>,
+    /// `Last-Modified`（`ETag` 缺失时的弱校验器；HTTP-date 字样原样发出）。
+    pub last_modified: Option<String>,
+    /// 返回 206 时故意把 `Content-Range` 写成从 0 开始（模拟段错位的坏服务器）。
+    pub lie_content_range: bool,
+    /// 读完请求头后**一个字节都不写**（服务端接了连接但永不回话）。
+    ///
+    /// 这是「无界 `await`」唯一能表达成用例的形态：读方不设上界就永远不返回。
+    pub silent: bool,
+    /// 写出这么多字节正文后就**停住不再写**（中途卡住），连接保持打开。
+    pub stall_at: Option<usize>,
+    /// 正文实际只发这么多字节（头部仍声称全量 = 短包）。
+    pub send_len: Option<usize>,
+    /// 正文之后**多**发这么多字节，且故意不发 `Content-Length`（超长包：HTTP/1.1 靠关连接分帧）。
+    pub trailing_extra: Option<usize>,
+    /// 正文按 `piece` 字节一片、片间睡 `gap` 滴流（慢但持续，用来证明超时不会误杀它）。
+    pub drip: Option<(usize, Duration)>,
+}
+
+impl MockResponse {
+    /// 200 + 完整内容（支持 Range）。
+    pub fn ok(body: impl Into<Vec<u8>>) -> Self {
+        Self {
+            status: 200,
+            body: body.into(),
+            range_support: true,
+            etag: Some("\"v1\"".to_string()),
+            last_modified: None,
+            lie_content_range: false,
+            silent: false,
+            stall_at: None,
+            send_len: None,
+            trailing_extra: None,
+            drip: None,
+        }
+    }
+
+    /// 固定状态码 + 空体。
+    pub fn status(status: u16) -> Self {
+        Self {
+            status,
+            body: Vec::new(),
+            range_support: false,
+            etag: None,
+            last_modified: 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;
+        self
+    }
+
+    /// 指定 `ETag`（换源用例要能造出「两台服务器各自不同的校验器」）。
+    pub fn with_etag(mut self, etag: impl Into<String>) -> Self {
+        self.etag = Some(etag.into());
+        self
+    }
+
+    /// 完全不发 `ETag`（模拟「自己没有强校验器」的源）。
+    pub fn without_etag(mut self) -> Self {
+        self.etag = None;
+        self
+    }
+
+    /// 指定 `Last-Modified`（`ETag` 缺失时 `If-Range` 的兜底取值）。
+    pub fn with_last_modified(mut self, value: impl Into<String>) -> Self {
+        self.last_modified = Some(value.into());
+        self
+    }
+
+    /// 206 但 `Content-Range` 谎报（模拟段错位的坏 CDN）。
+    pub fn with_wrong_content_range(mut self) -> Self {
+        self.lie_content_range = true;
+        self
+    }
+}

crates/core/tests/download/support/server.rs

@@ -0,0 +1,314 @@
+//! 进程内极简 HTTP/1.1 夹具服务器（零额外依赖）。
+//!
+//! 为什么自己写而不用 wiremock：P0 阶段控制依赖面（Win7 Tier-3 构建 + 单文件体积），一个
+//! `TcpListener` + 脚本队列足够表达「成功 / 500 / 校验不符 / 无视 Range / 中途卡住」。
+//!
+//! 正文写出的字节数记在 [`MockServer::bytes_sent`] 里：续传用例的判据是「服务端到底吐了
+//! 多少字节」，只看最终 hash 抓不到「其实整份又下了一遍」（见 `chunked/resume.rs`）。
+//!
+//! 从 `support/mod.rs` 拆出只为守 400 行硬线（M293）。
+
+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};
+
+pub use super::response::MockResponse;
+
+/// 夹具「挂住」的时长：远大于用例自己的上界，保证读方只能靠超时脱身。
+const HOLD: Duration = Duration::from_secs(30);
+
+/// 进程内极简 HTTP/1.1 服务器：按脚本顺序返回响应，脚本用尽后一律 500。
+pub struct MockServer {
+    addr: SocketAddr,
+    hits: Arc<AtomicUsize>,
+    bytes: Arc<std::sync::atomic::AtomicU64>,
+    if_ranges: Arc<std::sync::Mutex<Vec<String>>>,
+    handle: tokio::task::JoinHandle<()>,
+}
+
+impl MockServer {
+    /// 启动服务器（脚本是「每次请求一个响应」的队列）。
+    pub async fn start(script: Vec<MockResponse>) -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("绑定端口失败");
+        let addr = listener.local_addr().expect("取本地地址失败");
+        let hits = Arc::new(AtomicUsize::new(0));
+        let bytes = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        let if_ranges = Arc::new(std::sync::Mutex::new(Vec::new()));
+        let script = Arc::new(std::sync::Mutex::new(script));
+        let counter = Arc::clone(&hits);
+        let body_bytes = Arc::clone(&bytes);
+        let seen_if_ranges = Arc::clone(&if_ranges);
+
+        let handle = tokio::spawn(async move {
+            loop {
+                let Ok((stream, _)) = listener.accept().await else {
+                    break;
+                };
+                let script = Arc::clone(&script);
+                let counter = Arc::clone(&counter);
+                let bytes = Arc::clone(&body_bytes);
+                let seen = Arc::clone(&seen_if_ranges);
+                tokio::spawn(async move {
+                    let index = counter.fetch_add(1, Ordering::Relaxed);
+                    let response = {
+                        let script = script.lock().unwrap_or_else(|err| err.into_inner());
+                        script.get(index).cloned()
+                    };
+                    let response = response.unwrap_or_else(|| MockResponse::status(500));
+                    let _ = serve(stream, response, &bytes, &seen).await;
+                });
+            }
+        });
+
+        Self {
+            addr,
+            hits,
+            bytes,
+            if_ranges,
+            handle,
+        }
+    }
+
+    /// 第 `index` 个请求该用的 URL（换成不同路径可模拟「不同源」）。
+    pub fn url(&self, path: &str) -> String {
+        format!("http://{}{}", self.addr, path)
+    }
+
+    /// 换 host 拼 URL（用 `localhost` vs `127.0.0.1` 造出两个不同源）。
+    pub fn url_with(&self, host: &str, path: &str) -> String {
+        format!("http://{host}:{}{}", self.addr.port(), path)
+    }
+
+    /// 端口号。
+    pub fn port(&self) -> u16 {
+        self.addr.port()
+    }
+
+    /// 已收到的请求数。
+    pub fn hits(&self) -> usize {
+        self.hits.load(Ordering::Relaxed)
+    }
+
+    /// 已发出的**正文字节**总数（不含响应头）。
+    ///
+    /// 续传用例的判据靠它：「最终 hash 对」证明不了没白下字节，只有统计服务端真正吐了多少
+    /// 才知道是不是又整份下了一遍（见 `chunked/resume.rs`）。
+    pub fn bytes_sent(&self) -> u64 {
+        self.bytes.load(Ordering::Relaxed)
+    }
+
+    /// 收到过的 `If-Range` 取值（按到达顺序）。
+    ///
+    /// 换源用例的判据：源 A 的校验器**不许**出现在源 B 的请求里。夹具照 RFC 7233 §3.2 处置
+    /// （对不上就忽略 Range、回 200 + 完整正文），所以这条也顺带把「真实服务器上的后果」跑出来。
+    pub fn if_range_values(&self) -> Vec<String> {
+        self.if_ranges
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .clone()
+    }
+
+    /// 关停。
+    pub fn stop(self) {
+        self.handle.abort();
+    }
+}
+
+async fn serve(
+    mut stream: TcpStream,
+    response: MockResponse,
+    sent: &std::sync::atomic::AtomicU64,
+    seen_if_ranges: &std::sync::Mutex<Vec<String>>,
+) -> std::io::Result<()> {
+    let mut buffer = Vec::new();
+    let mut chunk = [0_u8; 1024];
+    loop {
+        let read = stream.read(&mut chunk).await?;
+        if read == 0 {
+            break;
+        }
+        buffer.extend_from_slice(&chunk[..read]);
+        if buffer.windows(4).any(|window| window == b"\r\n\r\n") {
+            break;
+        }
+    }
+    let request = String::from_utf8_lossy(&buffer).to_ascii_lowercase();
+    let range = request
+        .lines()
+        .find_map(|line| line.strip_prefix("range: bytes="))
+        .map(str::trim)
+        .and_then(parse_range_header);
+    let if_range = request
+        .lines()
+        .find_map(|line| line.strip_prefix("if-range:"))
+        .map(|value| value.trim().to_string());
+    if let Some(value) = &if_range {
+        seen_if_ranges
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+            .push(value.clone());
+    }
+    // RFC 7233 §3.2：`If-Range` 与当前表示对不上（含「服务器根本没有校验器」）⇒ **忽略 Range**，
+    // 回 200 + 完整正文。夹具照此处置，于是「拿别的源的校验器去问」会真的走到
+    // `fetch_chunk` 的 `RangeUnsupported` 分支（那就是真实服务器上退化成整份重下的那一跳）。
+    let validator_ok = if_range.as_deref().is_some_and(|sent_value| {
+        response
+            .etag
+            .as_deref()
+            .is_some_and(|etag| sent_value.eq_ignore_ascii_case(etag))
+            // ETag 缺失时按弱校验器（HTTP-date）比：这是 `If-Range` 的兜底支路。
+            || response
+                .last_modified
+                .as_deref()
+                .is_some_and(|date| sent_value.eq_ignore_ascii_case(date))
+    });
+    let range = if response.range_support && if_range.is_some() && !validator_ok {
+        None
+    } else {
+        range
+    };
+
+    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 => {
+            // 尊重请求里的结束偏移（分块下载必须逐段精确，否则组装后长度对不上）。
+            let end = requested_end.unwrap_or(total - 1).min(total - 1);
+            let slice = response.body[start as usize..=end as usize].to_vec();
+            // 谎报模式：body 仍是正确的那一段，但头部声称从 0 开始（长度恰好 → 只有
+            // 严格比对 Content-Range 才抓得住）。
+            let (claimed_start, claimed_end) = if response.lie_content_range {
+                (0, slice.len().saturating_sub(1) as u64)
+            } else {
+                (start, end)
+            };
+            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,
+            vec![
+                format!("Content-Range: bytes */{total}"),
+                "Content-Length: 0".to_string(),
+            ],
+            Vec::new(),
+        ),
+        _ => {
+            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());
+            }
+            (response.status, headers, response.body.clone())
+        }
+    };
+    let mut head = format!("HTTP/1.1 {status} {}\r\n", reason(status));
+    if let Some(etag) = &response.etag {
+        head.push_str(&format!("ETag: {etag}\r\n"));
+    }
+    if let Some(date) = &response.last_modified {
+        head.push_str(&format!("Last-Modified: {date}\r\n"));
+    }
+    for header in headers {
+        head.push_str(&header);
+        head.push_str("\r\n");
+    }
+    head.push_str("Connection: close\r\n\r\n");
+    stream.write_all(head.as_bytes()).await?;
+    write_body(&mut stream, &body, &response, sent).await?;
+    stream.flush().await?;
+    Ok(())
+}
+
+/// 按夹具的口径写正文：正常 / 卡在中途 / 短包 / 超长尾巴 / 滴流。发出的正文逐块计数。
+async fn write_body(
+    stream: &mut TcpStream,
+    body: &[u8],
+    response: &MockResponse,
+    sent: &std::sync::atomic::AtomicU64,
+) -> std::io::Result<()> {
+    if let Some(stall_at) = response.stall_at {
+        // 头部 + 这么多字节已发出，之后一个字不再写（连接不断）：读方只能靠超时脱身。
+        let sent_len = stall_at.min(body.len());
+        stream.write_all(&body[..sent_len]).await?;
+        stream.flush().await?;
+        sent.fetch_add(sent_len as u64, Ordering::Relaxed);
+        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?;
+                sent.fetch_add((end - offset) as u64, Ordering::Relaxed);
+                offset = end;
+                if offset < limit {
+                    tokio::time::sleep(gap).await;
+                }
+            }
+        }
+        None => {
+            stream.write_all(&body[..limit]).await?;
+            stream.flush().await?;
+            sent.fetch_add(limit as u64, Ordering::Relaxed);
+        }
+    }
+    if let Some(extra) = response.trailing_extra {
+        let filler = vec![0_u8; extra];
+        stream.write_all(&filler).await?;
+        stream.flush().await?;
+        sent.fetch_add(extra as u64, Ordering::Relaxed);
+    }
+    Ok(())
+}
+
+/// 解析 `Range: bytes=<start>-<end>`（`end` 可缺省 = 到文件末尾）。
+fn parse_range_header(value: &str) -> Option<(u64, Option<u64>)> {
+    let (start, end) = value.split_once('-')?;
+    let start = start.trim().parse::<u64>().ok()?;
+    let end = end.trim();
+    let end = if end.is_empty() {
+        None
+    } else {
+        Some(end.parse::<u64>().ok()?)
+    };
+    Some((start, end))
+}
+
+fn reason(status: u16) -> &'static str {
+    match status {
+        200 => "OK",
+        206 => "Partial Content",
+        404 => "Not Found",
+        416 => "Range Not Satisfiable",
+        500 => "Internal Server Error",
+        503 => "Service Unavailable",
+        _ => "Status",
+    }
+}