From ffca0f4454c89022fc868408a4396dbb286a3a6e Mon Sep 17 00:00:00 2001 From: AskaEth Date: Fri, 13 Nov 2026 06:43:39 +0800 Subject: [PATCH] =?UTF-8?q?fix(download):=20=E6=8D=A2=E6=BA=90=E5=90=8E?= =?UTF-8?q?=E4=B8=8D=E5=86=8D=E6=B2=BF=E7=94=A8=E6=97=A7=E6=BA=90=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C=E5=99=A8=20+=20=E4=B8=89=E6=9D=A1=20p3=20=E6=94=B6?= =?UTF-8?q?=E5=8F=A3=EF=BC=88M293=20r1=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复审 r1（p2-1）指出的链条：源 A 落 3/4 段失败 ⇒ 元数据 url=A + A 的 ETag；换源 B 时 `decide` 判 UrlChanged、删盘重来，但 `ChunkJournal::begin` 把 A 的校验器原样写进 url=B 的 新元数据，而 `record` 只在 `etag.is_none()` 时写入 ⇒ B 的真实校验器永远盖不上。于是拿 **A 的 ETag** 对 B 发 `If-Range`：按 RFC 7233 §3.2 服务器回 200 + 完整正文 ⇒ `fetch_chunk` 判 RangeUnsupported ⇒ 丢掉整份 ⇒ 单连接从头下 —— 债务以另一条路径复现。 修法（复审给的选项 A）：校验器与**记下它的 URL** 绑定，`same_source = previous.filter(|m| m.url == candidate.url)`，`etag` / `last_modified` / `If-Range` 三处全部走它，让「谁能当校验器」 只有一个判定点。不选 B（换源时清空）是因为 B 只是 A 在「URL 变了」这一特例上的写法，条件放在 读取处比散到写入处更难漏；不选 C（多源各记一份）是因为 `ResumeMeta` 本来就有 url，再加一个来源 字段会多一处可与 url 漂移的状态。 夹具此前不解析 `If-Range`，这一跳只能读代码推。现在 `support/` 记录每个请求的 `If-Range` 取值， 并按 RFC 7233 §3.2 处置（对不上就忽略 Range、回 200 + 完整正文），于是「真实服务器上的后果」也 进了用例。 判据（新用例 `chunked/switch.rs`，两个夹具服务器当一个官方源 + 一个镜像源）： * 源 B 收到的 `If-Range` 只许是它自己的 `"src-b"`（源 A 同源续传照旧带 `"src-a"` 作对照）； * 源 B 的字节数仍是 3 段 + 卡住的 1KiB + 补 1 段 = 41984（1.025×），换源没把同源续传打回整份重下； * 最终文件逐字节正确； * 第二条用例走「新源自己不发 ETag」那一支：`NoValidator` ⇒ 新源重试整份重下（`resumed_from=0`、 8 个请求、72704 字节），一个条件请求都不许发 —— 这是「无法证明就不复用」的明确取舍。 变异（去掉 URL 过滤）⇒ 两条都红，实测原文： assertion `left == right` failed: 源 B 收到了不属于它的校验器 left: ["\"src-a\""] right: ["\"src-b\""] 新源没有任何校验器，就不许发条件请求（带 A 的值更是串源）：["\"src-a\""] 同一变异下还实测到了后果：`b_bytes=113664 b_hits=6 resumed_from=0`（200 白下一份 + 单连接整份 重下 = 2.77×），即复审预测的那一跳确实存在。 三条 p3 一并收口： * p3-1：`abort_all()` 之前用 `try_join_next` 把「已落完但还没被收走」的段记进账 （`absorb_completed`，总长判据与正常回收同一道）。这是尽力减少浪费、不是正确性前提，且 它是「谁先完成」的竞态 ⇒ 没有确定性用例能钉住，已在文档里写明边界。 * p3-2：`verify_chunks` 改为返回 `Option`（可复用字节数），`AttemptOk.resumed_from` 直接取它，删掉原来「记录另算一遍」的双份口径。 * p3-3：`chunk_resume` 模块头新增「已知代价」：重读量正比于**复用的字节**（不是整份文件）、 为什么不能用「相信摘要」来省掉、以及超时后 `spawn_blocking` 线程不可取消在 Windows 上的 句柄风险。 拆文件守 400 行硬线：`await_chunks` + `absorb_completed` 移到 `transfer_chunk/join.rs` （逐字节搬移），`MockResponse` 移到 `support/response.rs`（逐字节搬移，只加 `with_etag` / `without_etag` 两个构造器）。 门禁：lib 903 / engine_mock 17 / chunked 20 / resume 12 / partfile 7 / cache 6 全绿， clippy -D warnings 0、fmt --check 0；字节判据复测未变差（resume 41984 = 1.025×， noresume 72704 = 1.775×）。 --- crates/core/src/download/chunk_resume.rs | 32 ++- crates/core/src/download/transfer_chunk.rs | 184 ++++---------- .../core/src/download/transfer_chunk/join.rs | 139 +++++++++++ crates/core/tests/download/chunked/switch.rs | 232 ++++++++++++++++++ crates/core/tests/download/chunked_test.rs | 4 + crates/core/tests/download/support/mod.rs | 5 +- .../core/tests/download/support/response.rs | 122 +++++++++ crates/core/tests/download/support/server.rs | 148 ++++------- 8 files changed, 628 insertions(+), 238 deletions(-) create mode 100644 crates/core/src/download/transfer_chunk/join.rs create mode 100644 crates/core/tests/download/chunked/switch.rs create mode 100644 crates/core/tests/download/support/response.rs diff --git a/crates/core/src/download/chunk_resume.rs b/crates/core/src/download/chunk_resume.rs index 64f75db..34dbada 100644 --- a/crates/core/src/download/chunk_resume.rs +++ b/crates/core/src/download/chunk_resume.rs @@ -28,6 +28,29 @@ //! //! 写的这一侧是本文末的 [`ChunkJournal`]：每成功回收一段就落一次盘（原子写），失败时元数据 //! 里已经有「哪几段、各自摘要是什么」，于是下一次尝试能接着下而不是从 0。 +//! +//! # 校验器跟着**源**走（复审 r1 p2-1） +//! +//! 校验器（ETag / Last-Modified）是「某个 URL 上那个表示」的属性，不是这个文件的属性：镜像 / +//! CDN / 对象存储各自生成，跨源几乎必然不同。把它当成文件的属性（例如换源后原样沿用）会让 +//! 身份门的「有校验器」被一个不相干的值满足，于是拿 A 的 ETag 去问 B —— 按 RFC 7233 服务器回 +//! 200 + 完整正文，分块路径拿不到 Range，整份丢掉退化单连接从头下（本 mission 要消灭的形态）。 +//! 所以沿用旧校验器的唯一条件是 `meta.url == 当前源 URL`，判定在口径上只有一处 +//! （`transfer_chunk` 的 `same_source`）。 +//! +//! # 已知代价（复审 r1 p3-3） +//! +//! * **每次续传都要重读「即将复用」的那些字节**（不是整份文件：复用 3/4 才读 3/4，1 段就只读 +//! 1 段）。这是本模块存在的意义本身——不复读就无法证明盘上旧字节还是当初那一段。188MB 文件 +//! 落完 3/4 时约 141MB 顺序读，本地 SSD 是毫秒级，网络盘上可能到几秒；它发生在 +//! 「进度从复用字节开始」之前，所以界面会看起来停在原地（未做专门进度事件，见 mission 报告 +//! 的跟进项）。要省掉这次读只有一条路：相信元数据里的摘要而不复核——那等于把「旧字节可信」 +//! 当前提，与上面的自证要求直接冲突，**明确不做**。 +//! * **超时后的读盘线程不可取消**：[`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}; @@ -208,7 +231,14 @@ impl ChunkJournal { self.meta.chunks.as_deref().unwrap_or_default() } - /// 记一个已落盘段（顺带补上服务器校验器：只有第一次拿到的算数，续传沿用旧的那个）。 + /// 记一个已落盘段（顺带补上服务器校验器）。 + /// + /// 校验器**只在还没有值时写入**，这是刻意的（复审 r1 问过）：同一个 URL 的续传里如果服务器 + /// 换了 ETag，保留旧值反而是保守选择——下一次带着旧 ETag 发 `If-Range` 会被服务器判「变了」 + /// 而回 200，于是干净重下；若改成覆盖成新值，下一次会把「旧字节 + 新表示」当成同一份续下去， + /// 只能等末尾整文件摘要才发现（那已经是白下）。跨源的情况根本不走到这里：换源时 + /// `transfer_chunk` 的 `same_source` 过滤已经让 `etag` 从 `None` 起步，这里填的是**本源的** + /// 校验器。 pub fn record(&mut self, fetched: ChunkFetch) { if self.meta.etag.is_none() { self.meta.etag = fetched.etag; diff --git a/crates/core/src/download/transfer_chunk.rs b/crates/core/src/download/transfer_chunk.rs index 476da7f..6a2c105 100644 --- a/crates/core/src/download/transfer_chunk.rs +++ b/crates/core/src/download/transfer_chunk.rs @@ -25,15 +25,18 @@ use std::time::Duration; use super::atomic::PartFile; use super::cache::now_ms; use super::chunk_resume::{self, ChunkDecision, ChunkJournal}; -use super::chunked::{self, ChunkFetch}; +use super::chunked; use super::engine::{AttemptOk, DownloadEngine}; use super::error::{DownloadError, DownloadErrorKind}; use super::progress::ProgressEvent; use super::resume::{ChunkRecord, ResumeMeta}; -use super::stall::{StallReason, StallReport, StallStage}; +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, @@ -73,24 +76,41 @@ impl DownloadEngine { // ② 自证：按记录重读那些字节、重算摘要。任何一段对不上就整份丢掉从头再来 // （理由见 `chunk_resume` 模块头：对不上说明这份元数据本身已经不可信）。 let mut done: Vec = Vec::new(); + let mut reused = 0_u64; match decision { ChunkDecision::Reuse { done: records } => { - if self.verify_chunks(part, &records).await { - done = records; - } else { - part.discard(); + // 复用字节数直接取自证的结果（唯一真相源：少读一段就少算一段）。 + match self.verify_chunks(part, &records).await { + Some(bytes) => { + done = records; + reused = bytes; + } + None => part.discard(), } } ChunkDecision::Restart { .. } => part.discard(), } - let reused: u64 = done.iter().map(ChunkRecord::len).sum(); - // 续传时把旧校验器带上：这是 `If-Range` 的取值来源，也是「远端内容没变」的协议层证据。 + // 校验器是**源**的属性（镜像/CDN/对象存储各自生成 ETag），不是这个文件的属性：只有 + // 「记下它的那个 URL 就是现在这个源」时才能沿用，否则当「没有校验器」。 + // + // 为什么必须挡（复审 r1 p2-1）：换源时 `decide` 判 `UrlChanged`、删盘重来，但 `previous` + // 还在内存里。把 A 的校验器写进 `url=B` 的新元数据之后，身份门的 `NoValidator` 会被一个 + // 不相干的校验器满足，于是拿 **A 的 ETag** 去对 B 发 `If-Range`：按 RFC 7233 §3.2， + // 服务器对不上就回 200 + 完整正文 ⇒ `fetch_chunk` 判 `RangeUnsupported` ⇒ 丢掉整份 ⇒ + // 单连接从头下——正是本 mission 要消灭的「从 0 重下」，只是换了条路径进来。 + // + // 三种修法里选「校验器与 URL 绑定」（复审的选项 A）而不是「换源时清空」（B）或「多源各记 + // 一份」（C）：`ResumeMeta` 本来就有 `url` 字段，绑定语义只需一个 `filter` 就能表达，且 + // 「这个校验器属于哪个源」在读取处即可判定，不必再加一个可能与 `url` 漂移的来源字段； + // 而 B 只是 A 在「URL 变了」这一特例上的写法，条件写在读取处比分散到写入处更难漏。 + let same_source = previous.as_ref().filter(|meta| meta.url == candidate.url); + // 续传时把**同源**的旧校验器带上：这是 `If-Range` 的取值来源，也是「远端内容没变」的 + // 协议层证据。（`done` 非空 ⇒ `decide` 已保证 `previous.url == candidate.url`， + // 这里仍走同一个 `same_source`，让「谁能当校验器」只有一个判定点。） let if_range = if done.is_empty() { None } else { - previous - .as_ref() - .and_then(|meta| meta.validator().if_range_value().map(str::to_string)) + same_source.and_then(|meta| meta.validator().if_range_value().map(str::to_string)) }; part.ensure_parent_async() @@ -117,11 +137,9 @@ impl DownloadEngine { let mut journal = ChunkJournal::begin( ResumeMeta { url: candidate.url.clone(), - // 续传沿用旧校验器：新响应可能不给 ETag，而 `If-Range` 要靠它。 - etag: previous.as_ref().and_then(|meta| meta.etag.clone()), - last_modified: previous - .as_ref() - .and_then(|meta| meta.last_modified.clone()), + // 只沿用**同一个源**的旧校验器；换了源就当没有，让下面每一段的响应自己去填。 + etag: same_source.and_then(|meta| meta.etag.clone()), + last_modified: same_source.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()), @@ -185,121 +203,6 @@ impl DownloadEngine { }) } - /// 回收分块任务，同时盯着「整个作业有没有在动」。 - /// - /// 两道判据必须分得开（另一处是单块自己的读空闲超时，阶段 = 等字节）： - /// * **无进度**：`counter` 连续 `chunk_stall_timeout` 没有新字节。分块任务卡在写盘上 - /// （杀软/网络盘）时单块读超时不会响，只有这道抓得住——现场留给 `join_next` 的正是 - /// 「188MB 落完 99% 之后永久挂住」那个形态； - /// * **总期限**：`chunk_job_budget`（按平均吞吐下限反推，见 `stall` 模块头）用完。 - /// - /// 两者都归到阶段 [`StallStage::AwaitChunkJoin`]，靠 [`StallReason`] 区分。 - /// - /// 成功的段在这里**顺手记账**（[`ChunkJournal::record`]）：这是唯一同时握着「段结果」与 - /// 「段区间」的地方，落到元数据之后，本作业后面怎么挂都不影响这几段的续传价值。 - async fn await_chunks( - &self, - set: &mut tokio::task::JoinSet>, - counter: &AtomicU64, - total: u64, - url: &str, - journal: &mut ChunkJournal, - ) -> Option { - let budget = self.cfg.chunk_job_budget(Some(total)); - let hard_deadline = tokio::time::Instant::now() + budget; - let mut last_seen = 0_u64; - let mut last_progress = tokio::time::Instant::now(); - loop { - let seen = counter.load(Ordering::Relaxed); - if seen > last_seen { - last_seen = seen; - last_progress = tokio::time::Instant::now(); - } - let wake = (last_progress + self.cfg.chunk_stall_timeout).min(hard_deadline); - match tokio::time::timeout_at(wake, set.join_next()).await { - Ok(Some(Ok(Ok(fetched)))) => { - // 「这个源给的还是不是同一个文件」：`Content-Range` 的总长必须与我们规划的 - // 一致，否则写下去的段根本拼不成目标文件（单连接路径同一处判据是收尾的 - // `written != total`）。 - if fetched.total.is_some_and(|server| server != 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)))) => { - 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)); - } - } - } - } - - /// 旧分块自证：按记录逐段重读、重算摘要。 - /// - /// `true` = 每一段都对得上，可以复用；`false` = 证明不了（摘要不符 / 盘上不够长 / 读盘失败 / - /// 超时），调用方必须整份丢掉。超时也归到「证明不了」：`spawn_blocking` 的线程不可取消， - /// 拿不到结果就不能拿旧字节去拼文件——宁可再生一遍字节。 - async fn verify_chunks(&self, part: &PartFile, records: &[ChunkRecord]) -> bool { - 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).is_ok() - }); - matches!( - tokio::time::timeout(self.cfg.verify_timeout, job).await, - Ok(Ok(true)) - ) - } - fn spawn_reporter( &self, id: DownloadId, @@ -322,6 +225,25 @@ impl DownloadEngine { }) } + /// 旧分块自证：按记录逐段重读、重算摘要。 + /// + /// `Some(bytes)` = 每一段都对得上，返回值就是**可复用的字节数**（调用方直接拿它当 + /// `resumed_from`，别再从记录另算一遍——两份口径会静默漂移）；`None` = 证明不了 + /// （摘要不符 / 盘上不够长 / 读盘失败 / 超时），调用方必须整份丢掉。 + /// + /// 超时也归到「证明不了」：`spawn_blocking` 的线程不可取消，到点后它还在读文件、还持着 + /// 句柄（与 `verify_part` 里那段注释同族，代价见 `chunk_resume` 模块头「已知代价」）。 + async fn verify_chunks(&self, part: &PartFile, records: &[ChunkRecord]) -> Option { + 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, + } + } + pub(crate) async fn verify_part( &self, part: &mut PartFile, diff --git a/crates/core/src/download/transfer_chunk/join.rs b/crates/core/src/download/transfer_chunk/join.rs new file mode 100644 index 0000000..9aef45d --- /dev/null +++ b/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>, + counter: &AtomicU64, + total: u64, + url: &str, + journal: &mut ChunkJournal, + ) -> Option { + let budget = self.cfg.chunk_job_budget(Some(total)); + let hard_deadline = tokio::time::Instant::now() + budget; + let mut last_seen = 0_u64; + let mut last_progress = tokio::time::Instant::now(); + loop { + let seen = counter.load(Ordering::Relaxed); + if seen > last_seen { + last_seen = seen; + last_progress = tokio::time::Instant::now(); + } + let wake = (last_progress + self.cfg.chunk_stall_timeout).min(hard_deadline); + match tokio::time::timeout_at(wake, set.join_next()).await { + Ok(Some(Ok(Ok(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>, + 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); + } + } + } +} diff --git a/crates/core/tests/download/chunked/switch.rs b/crates/core/tests/download/chunked/switch.rs new file mode 100644 index 0000000..4b4af39 --- /dev/null +++ b/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 { + (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 { + let mut script: Vec = (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 = (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(); +} diff --git a/crates/core/tests/download/chunked_test.rs b/crates/core/tests/download/chunked_test.rs index aadeb85..c38221b 100644 --- a/crates/core/tests/download/chunked_test.rs +++ b/crates/core/tests/download/chunked_test.rs @@ -381,3 +381,7 @@ mod timeout; /// 分块路径的续传（失败不重下、旧字节要自证）：同族用例单独一份。 #[path = "chunked/resume.rs"] mod resume; + +/// 换源后旧源校验器不许带过去（复审 r1 p2-1）：同样单独一份守 400 行硬线。 +#[path = "chunked/switch.rs"] +mod switch; diff --git a/crates/core/tests/download/support/mod.rs b/crates/core/tests/download/support/mod.rs index d10836c..e894c06 100644 --- a/crates/core/tests/download/support/mod.rs +++ b/crates/core/tests/download/support/mod.rs @@ -8,10 +8,13 @@ use std::sync::atomic::{AtomicUsize, Ordering}; +mod response; mod server; #[allow(unused_imports)] -pub use self::server::{MockResponse, MockServer}; +pub use self::response::MockResponse; +#[allow(unused_imports)] +pub use self::server::MockServer; /// 断言用的小工具：唯一临时目录（进程内计数 + 时间戳，避免并行测试互相踩）。 pub fn temp_dir(tag: &str) -> std::path::PathBuf { diff --git a/crates/core/tests/download/support/response.rs b/crates/core/tests/download/support/response.rs new file mode 100644 index 0000000..028c8e8 --- /dev/null +++ b/crates/core/tests/download/support/response.rs @@ -0,0 +1,122 @@ +//! 夹具的一次应答描述（状态码 / 正文 / 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, + /// 是否声明 `Accept-Ranges: bytes` 并在收到 `Range` 时真的切段。 + pub range_support: bool, + /// `ETag`（续传测试用）。 + pub etag: Option, + /// 返回 206 时故意把 `Content-Range` 写成从 0 开始（模拟段错位的坏服务器）。 + pub lie_content_range: bool, + /// 读完请求头后**一个字节都不写**（服务端接了连接但永不回话）。 + /// + /// 这是「无界 `await`」唯一能表达成用例的形态：读方不设上界就永远不返回。 + pub silent: bool, + /// 写出这么多字节正文后就**停住不再写**（中途卡住），连接保持打开。 + pub stall_at: Option, + /// 正文实际只发这么多字节（头部仍声称全量 = 短包）。 + pub send_len: Option, + /// 正文之后**多**发这么多字节，且故意不发 `Content-Length`（超长包：HTTP/1.1 靠关连接分帧）。 + pub trailing_extra: Option, + /// 正文按 `piece` 字节一片、片间睡 `gap` 滴流（慢但持续，用来证明超时不会误杀它）。 + pub drip: Option<(usize, Duration)>, +} + +impl MockResponse { + /// 200 + 完整内容（支持 Range）。 + pub fn ok(body: impl Into>) -> 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 + } + + /// 指定 `ETag`（换源用例要能造出「两台服务器各自不同的校验器」）。 + pub fn with_etag(mut self, etag: impl Into) -> Self { + self.etag = Some(etag.into()); + self + } + + /// 完全不发 `ETag`（模拟「自己没有任何校验器」的源）。 + pub fn without_etag(mut self) -> Self { + self.etag = None; + self + } + + /// 206 但 `Content-Range` 谎报（模拟段错位的坏 CDN）。 + pub fn with_wrong_content_range(mut self) -> Self { + self.lie_content_range = true; + self + } +} diff --git a/crates/core/tests/download/support/server.rs b/crates/core/tests/download/support/server.rs index 6558580..2891d72 100644 --- a/crates/core/tests/download/support/server.rs +++ b/crates/core/tests/download/support/server.rs @@ -16,119 +16,17 @@ use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; -/// mock 服务器给一次请求的应答。 -#[derive(Debug, Clone)] -pub struct MockResponse { - /// HTTP 状态码。 - pub status: u16, - /// 响应体。 - pub body: Vec, - /// 是否声明 `Accept-Ranges: bytes` 并在收到 `Range` 时真的切段。 - pub range_support: bool, - /// `ETag`（续传测试用）。 - pub etag: Option, - /// 返回 206 时故意把 `Content-Range` 写成从 0 开始（模拟段错位的坏服务器）。 - pub lie_content_range: bool, - /// 读完请求头后**一个字节都不写**（服务端接了连接但永不回话）。 - /// - /// 这是「无界 `await`」唯一能表达成用例的形态：读方不设上界就永远不返回。 - pub silent: bool, - /// 写出这么多字节正文后就**停住不再写**（中途卡住），连接保持打开。 - pub stall_at: Option, - /// 正文实际只发这么多字节（头部仍声称全量 = 短包）。 - pub send_len: Option, - /// 正文之后**多**发这么多字节，且故意不发 `Content-Length`（超长包：HTTP/1.1 靠关连接分帧）。 - pub trailing_extra: Option, - /// 正文按 `piece` 字节一片、片间睡 `gap` 滴流（慢但持续，用来证明超时不会误杀它）。 - pub drip: Option<(usize, Duration)>, -} +pub use super::response::MockResponse; /// 夹具「挂住」的时长：远大于用例自己的上界，保证读方只能靠超时脱身。 const HOLD: Duration = Duration::from_secs(30); -impl MockResponse { - /// 200 + 完整内容（支持 Range）。 - pub fn ok(body: impl Into>) -> 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, bytes: Arc, + if_ranges: Arc>>, handle: tokio::task::JoinHandle<()>, } @@ -141,9 +39,11 @@ impl MockServer { 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 { @@ -153,6 +53,7 @@ impl MockServer { 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 = { @@ -160,7 +61,7 @@ impl MockServer { script.get(index).cloned() }; let response = response.unwrap_or_else(|| MockResponse::status(500)); - let _ = serve(stream, response, &bytes).await; + let _ = serve(stream, response, &bytes, &seen).await; }); } }); @@ -169,6 +70,7 @@ impl MockServer { addr, hits, bytes, + if_ranges, handle, } } @@ -201,6 +103,17 @@ impl MockServer { self.bytes.load(Ordering::Relaxed) } + /// 收到过的 `If-Range` 取值（按到达顺序）。 + /// + /// 换源用例的判据：源 A 的校验器**不许**出现在源 B 的请求里。夹具照 RFC 7233 §3.2 处置 + /// （对不上就忽略 Range、回 200 + 完整正文），所以这条也顺带把「真实服务器上的后果」跑出来。 + pub fn if_range_values(&self) -> Vec { + self.if_ranges + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone() + } + /// 关停。 pub fn stop(self) { self.handle.abort(); @@ -211,6 +124,7 @@ async fn serve( mut stream: TcpStream, response: MockResponse, sent: &std::sync::atomic::AtomicU64, + seen_if_ranges: &std::sync::Mutex>, ) -> std::io::Result<()> { let mut buffer = Vec::new(); let mut chunk = [0_u8; 1024]; @@ -230,6 +144,30 @@ async fn serve( .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)) + }); + let range = if response.range_support && if_range.is_some() && !validator_ok { + None + } else { + range + }; if response.silent { // 接了连接、读完请求头、一个字都不回：读方不设上界就永远挂在这里。