fix(download): 换源后不再沿用旧源校验器 + 三条 p3 收口(M293 r1)
复审 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<u64>`(可复用字节数),`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×)。
This commit is contained in:
1 parent
3ade2d14a8
commit
ffca0f4454
8 files changed
+627
-237
No files matched your search
@@ -28,6 +28,29 @@
|
|||||||
//!
|
//!
|
||||||
//! 写的这一侧是本文末的 [`ChunkJournal`]:每成功回收一段就落一次盘(原子写),失败时元数据
|
//! 写的这一侧是本文末的 [`ChunkJournal`]:每成功回收一段就落一次盘(原子写),失败时元数据
|
||||||
//! 里已经有「哪几段、各自摘要是什么」,于是下一次尝试能接着下而不是从 0。
|
//! 里已经有「哪几段、各自摘要是什么」,于是下一次尝试能接着下而不是从 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::fs::File;
|
||||||
use std::io::{self, Read, Seek, SeekFrom};
|
use std::io::{self, Read, Seek, SeekFrom};
|
||||||
@@ -208,7 +231,14 @@ impl ChunkJournal {
|
|||||||
self.meta.chunks.as_deref().unwrap_or_default()
|
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) {
|
pub fn record(&mut self, fetched: ChunkFetch) {
|
||||||
if self.meta.etag.is_none() {
|
if self.meta.etag.is_none() {
|
||||||
self.meta.etag = fetched.etag;
|
self.meta.etag = fetched.etag;
|
||||||
|
|||||||
@@ -25,15 +25,18 @@ use std::time::Duration;
|
|||||||
use super::atomic::PartFile;
|
use super::atomic::PartFile;
|
||||||
use super::cache::now_ms;
|
use super::cache::now_ms;
|
||||||
use super::chunk_resume::{self, ChunkDecision, ChunkJournal};
|
use super::chunk_resume::{self, ChunkDecision, ChunkJournal};
|
||||||
use super::chunked::{self, ChunkFetch};
|
use super::chunked;
|
||||||
use super::engine::{AttemptOk, DownloadEngine};
|
use super::engine::{AttemptOk, DownloadEngine};
|
||||||
use super::error::{DownloadError, DownloadErrorKind};
|
use super::error::{DownloadError, DownloadErrorKind};
|
||||||
use super::progress::ProgressEvent;
|
use super::progress::ProgressEvent;
|
||||||
use super::resume::{ChunkRecord, ResumeMeta};
|
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::types::{Candidate, DownloadId, DownloadRequest, HashAlgo, Integrity};
|
||||||
use super::verify;
|
use super::verify;
|
||||||
|
|
||||||
|
/// 分块任务的回收(失速判据 + 已完成段记账):拆出守 400 行硬线。
|
||||||
|
mod join;
|
||||||
|
|
||||||
impl DownloadEngine {
|
impl DownloadEngine {
|
||||||
pub(crate) async fn attempt_chunked(
|
pub(crate) async fn attempt_chunked(
|
||||||
&self,
|
&self,
|
||||||
@@ -73,24 +76,41 @@ impl DownloadEngine {
|
|||||||
// ② 自证:按记录重读那些字节、重算摘要。任何一段对不上就整份丢掉从头再来
|
// ② 自证:按记录重读那些字节、重算摘要。任何一段对不上就整份丢掉从头再来
|
||||||
// (理由见 `chunk_resume` 模块头:对不上说明这份元数据本身已经不可信)。
|
// (理由见 `chunk_resume` 模块头:对不上说明这份元数据本身已经不可信)。
|
||||||
let mut done: Vec<ChunkRecord> = Vec::new();
|
let mut done: Vec<ChunkRecord> = Vec::new();
|
||||||
|
let mut reused = 0_u64;
|
||||||
match decision {
|
match decision {
|
||||||
ChunkDecision::Reuse { done: records } => {
|
ChunkDecision::Reuse { done: records } => {
|
||||||
if self.verify_chunks(part, &records).await {
|
// 复用字节数直接取自证的结果(唯一真相源:少读一段就少算一段)。
|
||||||
|
match self.verify_chunks(part, &records).await {
|
||||||
|
Some(bytes) => {
|
||||||
done = records;
|
done = records;
|
||||||
} else {
|
reused = bytes;
|
||||||
part.discard();
|
}
|
||||||
|
None => part.discard(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
ChunkDecision::Restart { .. } => part.discard(),
|
ChunkDecision::Restart { .. } => part.discard(),
|
||||||
}
|
}
|
||||||
let reused: u64 = done.iter().map(ChunkRecord::len).sum();
|
// 校验器是**源**的属性(镜像/CDN/对象存储各自生成 ETag),不是这个文件的属性:只有
|
||||||
// 续传时把旧校验器带上:这是 `If-Range` 的取值来源,也是「远端内容没变」的协议层证据。
|
// 「记下它的那个 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() {
|
let if_range = if done.is_empty() {
|
||||||
None
|
None
|
||||||
} else {
|
} else {
|
||||||
previous
|
same_source.and_then(|meta| meta.validator().if_range_value().map(str::to_string))
|
||||||
.as_ref()
|
|
||||||
.and_then(|meta| meta.validator().if_range_value().map(str::to_string))
|
|
||||||
};
|
};
|
||||||
|
|
||||||
part.ensure_parent_async()
|
part.ensure_parent_async()
|
||||||
@@ -117,11 +137,9 @@ impl DownloadEngine {
|
|||||||
let mut journal = ChunkJournal::begin(
|
let mut journal = ChunkJournal::begin(
|
||||||
ResumeMeta {
|
ResumeMeta {
|
||||||
url: candidate.url.clone(),
|
url: candidate.url.clone(),
|
||||||
// 续传沿用旧校验器:新响应可能不给 ETag,而 `If-Range` 要靠它。
|
// 只沿用**同一个源**的旧校验器;换了源就当没有,让下面每一段的响应自己去填。
|
||||||
etag: previous.as_ref().and_then(|meta| meta.etag.clone()),
|
etag: same_source.and_then(|meta| meta.etag.clone()),
|
||||||
last_modified: previous
|
last_modified: same_source.and_then(|meta| meta.last_modified.clone()),
|
||||||
.as_ref()
|
|
||||||
.and_then(|meta| meta.last_modified.clone()),
|
|
||||||
total: Some(total),
|
total: Some(total),
|
||||||
algo: integrity.map_or(HashAlgo::Sha1, |item| item.algo),
|
algo: integrity.map_or(HashAlgo::Sha1, |item| item.algo),
|
||||||
hash: integrity.map_or_else(String::new, |item| item.hex.clone()),
|
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<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) {
|
|
||||||
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(
|
fn spawn_reporter(
|
||||||
&self,
|
&self,
|
||||||
id: DownloadId,
|
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<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,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) async fn verify_part(
|
pub(crate) async fn verify_part(
|
||||||
&self,
|
&self,
|
||||||
part: &mut PartFile,
|
part: &mut PartFile,
|
||||||
|
|||||||
@@ -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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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();
|
||||||
|
}
|
||||||
@@ -381,3 +381,7 @@ mod timeout;
|
|||||||
/// 分块路径的续传(失败不重下、旧字节要自证):同族用例单独一份。
|
/// 分块路径的续传(失败不重下、旧字节要自证):同族用例单独一份。
|
||||||
#[path = "chunked/resume.rs"]
|
#[path = "chunked/resume.rs"]
|
||||||
mod resume;
|
mod resume;
|
||||||
|
|
||||||
|
/// 换源后旧源校验器不许带过去(复审 r1 p2-1):同样单独一份守 400 行硬线。
|
||||||
|
#[path = "chunked/switch.rs"]
|
||||||
|
mod switch;
|
||||||
@@ -8,10 +8,13 @@
|
|||||||
|
|
||||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||||
|
|
||||||
|
mod response;
|
||||||
mod server;
|
mod server;
|
||||||
|
|
||||||
#[allow(unused_imports)]
|
#[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 {
|
pub fn temp_dir(tag: &str) -> std::path::PathBuf {
|
||||||
|
|||||||
@@ -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<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)>,
|
||||||
|
}
|
||||||
|
|
||||||
|
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
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 指定 `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
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 206 但 `Content-Range` 谎报(模拟段错位的坏 CDN)。
|
||||||
|
pub fn with_wrong_content_range(mut self) -> Self {
|
||||||
|
self.lie_content_range = true;
|
||||||
|
self
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -16,119 +16,17 @@ use std::time::Duration;
|
|||||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||||
use tokio::net::{TcpListener, TcpStream};
|
use tokio::net::{TcpListener, TcpStream};
|
||||||
|
|
||||||
/// mock 服务器给一次请求的应答。
|
pub use super::response::MockResponse;
|
||||||
#[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);
|
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。
|
/// 进程内极简 HTTP/1.1 服务器:按脚本顺序返回响应,脚本用尽后一律 500。
|
||||||
pub struct MockServer {
|
pub struct MockServer {
|
||||||
addr: SocketAddr,
|
addr: SocketAddr,
|
||||||
hits: Arc<AtomicUsize>,
|
hits: Arc<AtomicUsize>,
|
||||||
bytes: Arc<std::sync::atomic::AtomicU64>,
|
bytes: Arc<std::sync::atomic::AtomicU64>,
|
||||||
|
if_ranges: Arc<std::sync::Mutex<Vec<String>>>,
|
||||||
handle: tokio::task::JoinHandle<()>,
|
handle: tokio::task::JoinHandle<()>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -141,9 +39,11 @@ impl MockServer {
|
|||||||
let addr = listener.local_addr().expect("取本地地址失败");
|
let addr = listener.local_addr().expect("取本地地址失败");
|
||||||
let hits = Arc::new(AtomicUsize::new(0));
|
let hits = Arc::new(AtomicUsize::new(0));
|
||||||
let bytes = Arc::new(std::sync::atomic::AtomicU64::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 script = Arc::new(std::sync::Mutex::new(script));
|
||||||
let counter = Arc::clone(&hits);
|
let counter = Arc::clone(&hits);
|
||||||
let body_bytes = Arc::clone(&bytes);
|
let body_bytes = Arc::clone(&bytes);
|
||||||
|
let seen_if_ranges = Arc::clone(&if_ranges);
|
||||||
|
|
||||||
let handle = tokio::spawn(async move {
|
let handle = tokio::spawn(async move {
|
||||||
loop {
|
loop {
|
||||||
@@ -153,6 +53,7 @@ impl MockServer {
|
|||||||
let script = Arc::clone(&script);
|
let script = Arc::clone(&script);
|
||||||
let counter = Arc::clone(&counter);
|
let counter = Arc::clone(&counter);
|
||||||
let bytes = Arc::clone(&body_bytes);
|
let bytes = Arc::clone(&body_bytes);
|
||||||
|
let seen = Arc::clone(&seen_if_ranges);
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let index = counter.fetch_add(1, Ordering::Relaxed);
|
let index = counter.fetch_add(1, Ordering::Relaxed);
|
||||||
let response = {
|
let response = {
|
||||||
@@ -160,7 +61,7 @@ impl MockServer {
|
|||||||
script.get(index).cloned()
|
script.get(index).cloned()
|
||||||
};
|
};
|
||||||
let response = response.unwrap_or_else(|| MockResponse::status(500));
|
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,
|
addr,
|
||||||
hits,
|
hits,
|
||||||
bytes,
|
bytes,
|
||||||
|
if_ranges,
|
||||||
handle,
|
handle,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -201,6 +103,17 @@ impl MockServer {
|
|||||||
self.bytes.load(Ordering::Relaxed)
|
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) {
|
pub fn stop(self) {
|
||||||
self.handle.abort();
|
self.handle.abort();
|
||||||
@@ -211,6 +124,7 @@ async fn serve(
|
|||||||
mut stream: TcpStream,
|
mut stream: TcpStream,
|
||||||
response: MockResponse,
|
response: MockResponse,
|
||||||
sent: &std::sync::atomic::AtomicU64,
|
sent: &std::sync::atomic::AtomicU64,
|
||||||
|
seen_if_ranges: &std::sync::Mutex<Vec<String>>,
|
||||||
) -> std::io::Result<()> {
|
) -> std::io::Result<()> {
|
||||||
let mut buffer = Vec::new();
|
let mut buffer = Vec::new();
|
||||||
let mut chunk = [0_u8; 1024];
|
let mut chunk = [0_u8; 1024];
|
||||||
@@ -230,6 +144,30 @@ async fn serve(
|
|||||||
.find_map(|line| line.strip_prefix("range: bytes="))
|
.find_map(|line| line.strip_prefix("range: bytes="))
|
||||||
.map(str::trim)
|
.map(str::trim)
|
||||||
.and_then(parse_range_header);
|
.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 {
|
if response.silent {
|
||||||
// 接了连接、读完请求头、一个字都不回:读方不设上界就永远挂在这里。
|
// 接了连接、读完请求头、一个字都不回:读方不设上界就永远挂在这里。
|
||||||
|
|||||||
Reference in new issue
Block a user