feat(m165): 下载引擎按数据根跨请求复用（运行时常驻 + 构造次数守卫）
宿主侧以前每次请求都现建一个 DownloadEngine：连接池、信号量限流、 健康度记忆、内容寻址缓存索引全部清零。现在改成按数据根共享一个实例。 - core: 新增 download::shared（进程级登记处，键 = 缓存根）—— engine_for / forget / constructions；region/health/cache/mirrors 的装配 收在这一个 build() 里（持锁构造只做本地 I/O，不联网） - core: download::blocking 改成进程内唯一常驻运行时。reqwest 的连接池绑在 驱动连接的运行时上，引擎复用必须配运行时常驻，否则复用的是个装满死连接的池 - presenter: host::engine::engine 返回 Arc<DownloadEngine>（共享实例）， addon/update/versions/java/authlib 等全部调用点随之改；新增 forget 供 「重新检测地区」失效（否则进程级引擎下那句话不成立） - presenter: addon 详情页把「项目 + 版本列表」两次 block_on 收成一次 失败分类 / 镜像回退 / Retry-After 语义未改动（error.rs、http.rs、engine.rs 的 候选循环一字未动）。 守卫（都带变异实证）： - shared::tests::one_engine_per_cache_root + host::engine::tests:: host_requests_share_one_engine：断言 constructions(root) == 1（不是只看结果） - shared::tests::pooled_connection_survives_across_block_on_calls：127.0.0.1 回环服务统计 accept 次数，两次 block_on 必须只拨号一次 验证：cargo test -p unlucky-core --lib（608 过）、5 个 download 集成 target（42 过）、 cargo test -p unlucky-presenter --lib（267 过）、cargo clippy --all-targets、 cargo fmt --check 全绿。真站点（Modrinth/CurseForge/BMCLAPI）未打，见报告。
This commit is contained in:
AskaEth committed
1 parent
b4e6fbf8c1
commit
46389d61fd
11 files changed
+486
-48
No files matched your search
@@ -0,0 +1,289 @@
|
||||
//! 进程级共享引擎：**一个数据根一个 `DownloadEngine`**，跨请求复用。
|
||||
//!
|
||||
//! # 为什么要有这一层（M165）
|
||||
//!
|
||||
//! 在此之前，宿主侧每处理一次「请求」（一次检索 / 一次详情 / 一次装文件 / 一次拉清单）都会
|
||||
//! 现建一个引擎（`DownloadEngine::new`）。建的时候什么都会重新来一遍，于是下面这些**跨请求
|
||||
//! 本该积累的状态每次都清零**：
|
||||
//!
|
||||
//! | 引擎里的东西 | 每建一次就丢什么 |
|
||||
//! |--------------|------------------|
|
||||
//! | `reqwest::Client`（`http::build_client`） | 整条连接池：每个请求重新 TCP + TLS 握手 |
|
||||
//! | `Semaphore`（`max_concurrency`） | 限流变成「每请求 64」，N 个并发请求 = 64×N 条在飞 |
|
||||
//! | `HealthRegistry`（`open_in`） | 从盘上重读 `health.json`；`last_save_ms` 被重置成「现在」⇒ 收尾那次 `flush()` 落在 5s 节流窗口里直接返回，**这个短命引擎记下的失败数一并丢掉**（只有「进/出冷却」会立刻落盘） |
|
||||
//! | `ContentCache`（`open`） | 从盘上重读并解析整份 `index.json`；`persist()` 用**本请求这一份内存索引**覆写全文件 ⇒ 两个并发请求各自覆写，后写的把先写的登记条目冲掉 |
|
||||
//! | `MirrorTable::builtin()` | 重新解析一遍内置 `mirrors.json` |
|
||||
//! | 地区档案 + `Jitter` | 重读 `region.json`；抖动序列重新播种 |
|
||||
//!
|
||||
//! 失败分类、镜像回退、`Retry-After` 语义一概不动——复用只改「谁来建、建几次」，不改
|
||||
//! [`DownloadEngine`] 内部的任何判定。
|
||||
//!
|
||||
//! # 边界（谁持有 / 活多久 / 并发多少）
|
||||
//!
|
||||
//! * **谁持有**：本模块的进程级登记处（`OnceLock<Mutex<…>>`），键是**数据根的缓存目录**
|
||||
//! （`<launcher_home>/cache`）。同一数据根的多次请求拿到同一个 `Arc<DownloadEngine>`；
|
||||
//! 不同数据根各有一份，缓存/健康度不会串味（有测试钉住这一条）。
|
||||
//! * **活多久**：跟随进程。登记处**不做淘汰**——淘汰会让后续请求建出第二个引擎，同一个根上
|
||||
//! 同时存在两条连接池，等于把要修的问题换个地方复现。要让地区判定等「一次性派生值」重算，
|
||||
//! 用 [`forget`] 显式失效（设置页「重新检测地区」就这么做）。
|
||||
//! * **并发多少**：`EngineConfig::default().max_concurrency`（64）现在是**进程级**上限，
|
||||
//! 不再是「每请求 64」。单请求独占时与从前一致；多请求并发时它们分摊这 64 张许可。
|
||||
//! * **镜像切换 / 失效**：一律走引擎原有的候选规划 + 健康度冷却，本模块不参与判定；唯一可见的
|
||||
//! 差别是健康度记忆现在真的跨请求累积（这本就是它存在的理由）。
|
||||
//!
|
||||
//! # 与 [`super::blocking::block_on`] 的绑定关系
|
||||
//!
|
||||
//! `reqwest` 的连接池绑在**驱动连接的运行时**上：引擎复用了而每次 `block_on` 都新起一个运行时，
|
||||
//! 池里的空闲连接就属于已经销毁的运行时，复用会退化成「拿到的都是死连接」。所以引擎复用必须
|
||||
//! 配运行时常驻（`blocking` 模块那头一并改成进程内唯一运行时）。两者是同一件事的两半。
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
|
||||
|
||||
use super::cache::{now_ms, ContentCache};
|
||||
use super::config::EngineConfig;
|
||||
use super::engine::DownloadEngine;
|
||||
use super::health::HealthRegistry;
|
||||
use super::mirrors::MirrorTable;
|
||||
use super::region::{decide_from, load_cached, store_cached};
|
||||
use super::region_system::detect_signals;
|
||||
|
||||
/// 登记处：缓存根 → 引擎，外加「这个根建过几次引擎」（守卫/诊断用，**不**随淘汰清零）。
|
||||
#[derive(Debug, Default)]
|
||||
struct Registry {
|
||||
engines: HashMap<PathBuf, Arc<DownloadEngine>>,
|
||||
builds: HashMap<PathBuf, usize>,
|
||||
}
|
||||
|
||||
fn registry() -> &'static Mutex<Registry> {
|
||||
static REGISTRY: OnceLock<Mutex<Registry>> = OnceLock::new();
|
||||
REGISTRY.get_or_init(|| Mutex::new(Registry::default()))
|
||||
}
|
||||
|
||||
fn lock() -> MutexGuard<'static, Registry> {
|
||||
registry()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
/// 缓存根的键：去掉尾部分隔符（`D:/x/cache` 与 `D:/x/cache/` 视作同一根）。
|
||||
fn key_of(cache_root: &Path) -> PathBuf {
|
||||
let text = cache_root.to_string_lossy();
|
||||
let trimmed = text.trim_end_matches(['/', '\\']);
|
||||
if trimmed.is_empty() {
|
||||
cache_root.to_path_buf()
|
||||
} else {
|
||||
PathBuf::from(trimmed)
|
||||
}
|
||||
}
|
||||
|
||||
/// 取该数据根上的共享引擎（第一次调用时构造，之后一律复用）。
|
||||
///
|
||||
/// 构造过程只做本地 I/O（读 `region.json` / `health.json` / `index.json`，解析内置
|
||||
/// `mirrors.json`），**不联网**；因此持锁构造不会把网络等待带进临界区。
|
||||
#[must_use]
|
||||
pub fn engine_for(cache_root: impl AsRef<Path>) -> Arc<DownloadEngine> {
|
||||
let key = key_of(cache_root.as_ref());
|
||||
let mut reg = lock();
|
||||
if let Some(engine) = reg.engines.get(&key) {
|
||||
return Arc::clone(engine);
|
||||
}
|
||||
let engine = Arc::new(build(&key));
|
||||
reg.engines.insert(key.clone(), Arc::clone(&engine));
|
||||
*reg.builds.entry(key).or_default() += 1;
|
||||
engine
|
||||
}
|
||||
|
||||
/// 让该数据根的共享引擎失效（下次 [`engine_for`] 重新构造）。
|
||||
///
|
||||
/// 在飞请求持有的 `Arc` 不受影响（它们继续用完手里那个引擎）；之后的新请求拿到新引擎，
|
||||
/// 于是「重新检测地区」这类一次性派生值能真的重算。返回是否真淘汰了一份。
|
||||
pub fn forget(cache_root: impl AsRef<Path>) -> bool {
|
||||
lock()
|
||||
.engines
|
||||
.remove(&key_of(cache_root.as_ref()))
|
||||
.is_some()
|
||||
}
|
||||
|
||||
/// 该数据根在本进程里**实际构造过几次引擎**。
|
||||
///
|
||||
/// 这是给守卫与诊断用的：跨请求复用成立时，同一个根无论被请求多少次，这个数都停在 `1`
|
||||
/// （`forget` 之后每重建一次 +1）。只断言「结果一样」测不出复用——引擎建几次都能下同一个文件。
|
||||
#[must_use]
|
||||
pub fn constructions(cache_root: impl AsRef<Path>) -> usize {
|
||||
lock()
|
||||
.builds
|
||||
.get(&key_of(cache_root.as_ref()))
|
||||
.copied()
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// 真实构造（**唯一**一处 `DownloadEngine::new` 的宿主侧装配）。
|
||||
fn build(cache_root: &Path) -> DownloadEngine {
|
||||
let now = now_ms();
|
||||
let region_path = cache_root.join("region.json");
|
||||
let region = load_cached(®ion_path, now).unwrap_or_else(|| {
|
||||
let profile = decide_from(&detect_signals(), now);
|
||||
let _ = store_cached(®ion_path, &profile);
|
||||
profile
|
||||
});
|
||||
DownloadEngine::new(
|
||||
EngineConfig::default(),
|
||||
Arc::new(MirrorTable::builtin()),
|
||||
Arc::new(region),
|
||||
Arc::new(HealthRegistry::open_in(cache_root, now)),
|
||||
Arc::new(ContentCache::open(cache_root)),
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// 每个用例一个独立数据根（登记处是进程级的，键撞了会互相干扰）。
|
||||
fn root(tag: &str) -> PathBuf {
|
||||
static SEQ: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
|
||||
let seq = SEQ.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
||||
let dir =
|
||||
std::env::temp_dir().join(format!("ull-shared-{tag}-{}-{seq}", std::process::id()));
|
||||
let _ = std::fs::remove_dir_all(&dir);
|
||||
std::fs::create_dir_all(&dir).expect("建缓存根");
|
||||
dir
|
||||
}
|
||||
|
||||
/// 守卫：同一个根上「多次请求」只构造一个引擎（**断言构造次数，不是只看结果**）。
|
||||
///
|
||||
/// 抓得住的变异：把 [`engine_for`] 里的 `engines.get` 命中分支删掉（每次现建），
|
||||
/// 本用例立刻红在 `constructions == 1` 与 `Arc::ptr_eq` 两处。
|
||||
#[test]
|
||||
fn one_engine_per_cache_root() {
|
||||
let dir = root("one");
|
||||
let first = engine_for(&dir);
|
||||
let second = engine_for(&dir);
|
||||
assert_eq!(constructions(&dir), 1, "同一根只该构造一次引擎");
|
||||
assert!(Arc::ptr_eq(&first, &second), "两次请求必须是同一个引擎");
|
||||
// 尾部分隔符不该被当成另一个根。
|
||||
let with_sep = PathBuf::from(format!("{}/", dir.display()));
|
||||
assert!(
|
||||
Arc::ptr_eq(&first, &engine_for(&with_sep)),
|
||||
"尾部分隔符不该造成第二个引擎"
|
||||
);
|
||||
assert_eq!(constructions(&dir), 1);
|
||||
std::fs::remove_dir_all(&dir).ok();
|
||||
}
|
||||
|
||||
/// 两个数据根**不共享**引擎：缓存 / 健康度串味比多建一个引擎严重得多。
|
||||
#[test]
|
||||
fn different_cache_roots_do_not_share() {
|
||||
let a = root("a");
|
||||
let b = root("b");
|
||||
let engine_a = engine_for(&a);
|
||||
let engine_b = engine_for(&b);
|
||||
assert!(!Arc::ptr_eq(&engine_a, &engine_b));
|
||||
assert_eq!(constructions(&a), 1);
|
||||
assert_eq!(constructions(&b), 1);
|
||||
assert_eq!(engine_a.cache().root(), a.as_path());
|
||||
assert_eq!(engine_b.cache().root(), b.as_path());
|
||||
std::fs::remove_dir_all(&a).ok();
|
||||
std::fs::remove_dir_all(&b).ok();
|
||||
}
|
||||
|
||||
/// 失效后重建（设置页「重新检测地区」依赖这条语义）。
|
||||
#[test]
|
||||
fn forget_rebuilds_once() {
|
||||
let dir = root("forget");
|
||||
let before = engine_for(&dir);
|
||||
assert!(forget(&dir), "已存在的引擎应被淘汰");
|
||||
assert!(!forget(&dir), "淘汰后不该再有可淘汰的");
|
||||
let after = engine_for(&dir);
|
||||
assert!(!Arc::ptr_eq(&before, &after));
|
||||
assert_eq!(constructions(&dir), 2, "失效一次 = 多构造一次");
|
||||
std::fs::remove_dir_all(&dir).ok();
|
||||
}
|
||||
|
||||
/// 离线回环 HTTP 服务：记下「接受了几条连接 / 处理了几条请求」。
|
||||
///
|
||||
/// 只回固定 JSON、只讲 HTTP/1.1 keep-alive，够用来判「两次下载有没有复用同一条连接」。
|
||||
fn loopback_server(
|
||||
expected_requests: usize,
|
||||
) -> (String, std::sync::mpsc::Receiver<(usize, usize)>) {
|
||||
use std::io::{Read as _, Write as _};
|
||||
use std::net::TcpListener;
|
||||
|
||||
let listener = TcpListener::bind("127.0.0.1:0").expect("绑定回环端口");
|
||||
let addr = listener.local_addr().expect("取端口");
|
||||
let (tx, rx) = std::sync::mpsc::channel();
|
||||
std::thread::spawn(move || {
|
||||
let mut conns = 0usize;
|
||||
let mut requests = 0usize;
|
||||
while requests < expected_requests {
|
||||
let Ok((mut socket, _)) = listener.accept() else {
|
||||
break;
|
||||
};
|
||||
conns += 1;
|
||||
let _ = socket.set_read_timeout(Some(std::time::Duration::from_secs(5)));
|
||||
let mut seen = Vec::new();
|
||||
loop {
|
||||
let mut chunk = [0u8; 1024];
|
||||
match socket.read(&mut chunk) {
|
||||
Ok(0) | Err(_) => break,
|
||||
Ok(read) => seen.extend_from_slice(&chunk[..read]),
|
||||
}
|
||||
if !seen.windows(4).any(|window| window == b"\r\n\r\n") {
|
||||
continue;
|
||||
}
|
||||
requests += 1;
|
||||
let body = br#"{"ok":true}"#;
|
||||
let head = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n",
|
||||
body.len()
|
||||
);
|
||||
if socket.write_all(head.as_bytes()).is_err() || socket.write_all(body).is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
seen.clear();
|
||||
if requests >= expected_requests {
|
||||
let _ = tx.send((conns, requests));
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
let _ = tx.send((conns, requests));
|
||||
});
|
||||
(format!("http://{addr}/meta.json"), rx)
|
||||
}
|
||||
|
||||
/// **复用真的落到连接上**：同一个引擎跨两次 `block_on` 取同一个 host，只应该拨号一次。
|
||||
///
|
||||
/// 这条钉住的是「引擎复用」与「运行时常驻」必须成对出现的那半个承诺：只复用引擎、
|
||||
/// 每次 `block_on` 新起运行时的话，连接池里的空闲连接属于已销毁的运行时，
|
||||
/// 第二次请求会重新 TCP 握手——`conns` 就是 2 而不是 1。
|
||||
///
|
||||
/// 离线可跑：服务端是 127.0.0.1 的回环监听器，不碰任何外网站点。
|
||||
#[test]
|
||||
fn pooled_connection_survives_across_block_on_calls() {
|
||||
use super::super::blocking::block_on;
|
||||
use super::super::types::{DownloadRequest, ResourceKind};
|
||||
|
||||
let dir = root("pool");
|
||||
let (url, counts) = loopback_server(2);
|
||||
let engine = engine_for(&dir);
|
||||
for seq in 0..2 {
|
||||
let request = DownloadRequest::new(
|
||||
ResourceKind::Other,
|
||||
url.clone(),
|
||||
dir.join(format!("meta-{seq}.json")),
|
||||
);
|
||||
let bytes = block_on(engine.download_bytes(request)).expect("回环请求应成功");
|
||||
assert_eq!(bytes, br#"{"ok":true}"#.to_vec());
|
||||
}
|
||||
let (conns, requests) = counts
|
||||
.recv_timeout(std::time::Duration::from_secs(10))
|
||||
.expect("服务端应报告计数");
|
||||
assert_eq!(requests, 2, "两次下载 = 两条请求");
|
||||
assert_eq!(conns, 1, "两次下载应复用同一条连接");
|
||||
std::fs::remove_dir_all(&dir).ok();
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user