From 46389d61fd658b41d7e526064a0336e935dc6fdb Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 8 Nov 2026 08:43:15 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(m165):=20=E4=B8=8B=E8=BD=BD=E5=BC=95?= =?UTF-8?q?=E6=93=8E=E6=8C=89=E6=95=B0=E6=8D=AE=E6=A0=B9=E8=B7=A8=E8=AF=B7?= =?UTF-8?q?=E6=B1=82=E5=A4=8D=E7=94=A8=EF=BC=88=E8=BF=90=E8=A1=8C=E6=97=B6?= =?UTF-8?q?=E5=B8=B8=E9=A9=BB=20+=20=E6=9E=84=E9=80=A0=E6=AC=A1=E6=95=B0?= =?UTF-8?q?=E5=AE=88=E5=8D=AB=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 宿主侧以前每次请求都现建一个 DownloadEngine：连接池、信号量限流、 健康度记忆、内容寻址缓存索引全部清零。现在改成按数据根共享一个实例。 - core: 新增 download::shared（进程级登记处，键 = 缓存根）—— engine_for / forget / constructions；region/health/cache/mirrors 的装配 收在这一个 build() 里（持锁构造只做本地 I/O，不联网） - core: download::blocking 改成进程内唯一常驻运行时。reqwest 的连接池绑在 驱动连接的运行时上，引擎复用必须配运行时常驻，否则复用的是个装满死连接的池 - presenter: host::engine::engine 返回 Arc（共享实例）， 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）未打，见报告。 --- crates/core/src/addon/online/mod.rs | 3 +- crates/core/src/addon/online/transport.rs | 5 + crates/core/src/download/blocking.rs | 82 ++++- crates/core/src/download/mod.rs | 5 + crates/core/src/download/shared.rs | 289 ++++++++++++++++++ crates/presenter/src/host/addon/http.rs | 5 + crates/presenter/src/host/addon/sources.rs | 16 +- crates/presenter/src/host/engine.rs | 121 ++++++-- .../src/host/install_processors_fixture.rs | 2 +- crates/presenter/src/host/settings.rs | 4 + crates/presenter/src/host/update.rs | 2 +- 11 files changed, 486 insertions(+), 48 deletions(-) create mode 100644 crates/core/src/download/shared.rs diff --git a/crates/core/src/addon/online/mod.rs b/crates/core/src/addon/online/mod.rs index d548b17..7886548 100644 --- a/crates/core/src/addon/online/mod.rs +++ b/crates/core/src/addon/online/mod.rs @@ -21,7 +21,8 @@ //! `Cargo.lock` 里已有的 `reqwest` 0.12 + `tokio` + `serde_json` 就是全部。生产实现 //! [`EngineTransport`] 走 `DownloadEngine::download_bytes`，于是**信号量限流、源健康度、 //! 指数退避、错误分类**全由 `core::download` 提供，本模块不写第二个重试循环，也不持有第二个 -//! `reqwest::Client`（与 `addon::import::fetch` 同一条纪律）。 +//! `reqwest::Client`（与 `addon::import::fetch` 同一条纪律）。那个引擎是宿主侧的进程级共享实例 +//! （M165：`download::shared::engine_for`），所以元数据请求与文件下载共用同一条连接池与限流。 //! //! 没有 POST：Modrinth 的批量接口（`POST /v2/version_files`）需要 POST 请求体，而引擎的 //! `DownloadRequest` 没有请求体字段。补全路径因此按**单摘要 GET** 逐条查（循环在 diff --git a/crates/core/src/addon/online/transport.rs b/crates/core/src/addon/online/transport.rs index 4233068..932b973 100644 --- a/crates/core/src/addon/online/transport.rs +++ b/crates/core/src/addon/online/transport.rs @@ -17,6 +17,11 @@ //! （与 `addon::import::fetch` 的纪律同源）。镜像策略同理：`mirrors.json` 里没有 //! `api.modrinth.com` 的前缀规则 ⇒ 元数据请求永远是直连，只有真正的文件下载才会被镜像 //! 规则命中（国内 gh-proxy 那条也只看 GitHub host）。 +//! +//! 引擎本身由宿主提供**且是进程级共享的**（M165：`download::shared::engine_for`，一个数据根 +//! 一个）：`EngineTransport` 只借一个 `&DownloadEngine`，不负责建也不负责销毁。于是 +//! 「检索 → 详情 → 装文件」这一串请求复用同一条连接池、同一个限流闸门、同一份镜像健康度—— +//! 这正是本模块当初拒绝自己造 client 的同一个理由，只是从「模块内不重复造」扩到了「跨请求不重复造」。 use std::future::Future; use std::pin::Pin; diff --git a/crates/core/src/download/blocking.rs b/crates/core/src/download/blocking.rs index 1a3c4f2..4f83992 100644 --- a/crates/core/src/download/blocking.rs +++ b/crates/core/src/download/blocking.rs @@ -4,35 +4,70 @@ //! 但 headless 消费者（`crates/cli` 的 `ull`）不依赖 tokio，也不该为一次安装 //! 把整个异步运行时依赖树拉进 CLI 的 manifest（D18：CLI 依赖树保持干净）。 //! -//! 因此把「建一个短命运行时 + 跑完整段 future」这两件事收在 core 里： +//! 因此把「跑完整段 future」这件事收在 core 里： //! //! ```ignore //! let outcome = unlucky_core::download::blocking::block_on(async { /* 规划 + 下载 */ }); //! ``` //! -//! 约定：**整段**安装流程包在一次 `block_on` 里（同一个运行时），不要每次下载都建一个—— -//! reqwest 的连接池绑定在创建它的运行时上，跨运行时会退化甚至报「dispatch task is gone」。 +//! # 一句话约定（M165 起） //! -//! 这是给「单线程 CLI」用的逃生口，不是通用设施：app 层请直接用原生异步。 +//! **整个进程只有一个运行时**，所有 [`block_on`] 都跑在它上面。理由两条，都是实测口径： +//! +//! 1. **连接池绑在运行时上**。`reqwest` 的连接池条目由驱动连接的运行时任务持有：运行时一销毁， +//! 池里的空闲连接就全成了尸体。M165 之前每次 `block_on` 现建一个临时运行时，于是下载引擎 +//! 的连接池**跨请求根本留不住**（每次重新 TCP + TLS）；引擎复用（[`super::shared`]）与 +//! 运行时常驻必须成对出现，只做一半会退化成「复用一个装着死连接的池」。 +//! 2. **建运行时就是起线程**。每个请求建一次 = 每次多起两个线程，再各自销毁。 +//! +//! 调用形状不变：把**一整段**流程包在一次 `block_on` 里仍然是对的（同一个 future 里少切几次 +//! 上下文，也少几次许可申请/释放）；只是这几次调用现在落在同一个运行时上，不再互相孤立。 +//! +//! # 仍然不能嵌套 +//! +//! 在已有异步上下文里调 [`block_on`] 会 panic（tokio 的硬规则）。这条与 M165 之前一致。 use std::future::Future; +use std::sync::OnceLock; -/// 在一个临时 tokio 多线程运行时上跑完 `future`。 +/// 共享运行时的 worker 数。 /// -/// 运行时随调用结束销毁，因此适合「一条命令一次阻塞」的 CLI 生命周期。 +/// 下载流程里有真实的同步 I/O（`PartFile` 落盘、`write_atomic` 写索引/健康度），它们跑在 +/// worker 线程上；M165 之前每个请求自带 2 个 worker，现在这些请求共用同一池，所以按机器并行度 +/// 给一点余量（下限 2 保证单核机器也够驱动，上限 8 免得为一个启动器起一船线程）。 +fn workers() -> usize { + std::thread::available_parallelism() + .map_or(2, std::num::NonZeroUsize::get) + .clamp(2, 8) +} + +/// 进程内唯一的阻塞桥运行时（第一次 [`block_on`] 时创建，之后一直活着）。 +/// +/// 故意不实现关闭：它的生命周期就是进程的生命周期。静态量不会被 drop，因此进程退出时 +/// 也不会去等它「优雅停机」（那正是短命运行时被销毁时的开销）。 +fn bridge() -> &'static tokio::runtime::Runtime { + static BRIDGE: OnceLock = OnceLock::new(); + BRIDGE.get_or_init(|| { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(workers()) + .thread_name("ull-blocking") + .enable_all() + .build() + .expect("创建 tokio 运行时失败（内存/线程资源不足）") + }) +} + +/// 在进程共享的多线程运行时上跑完 `future`（阻塞当前线程）。 +/// +/// 多个线程可以同时调用：它们在同一个运行时上各自阻塞自己的线程，任务在同一个 worker 池里跑， +/// 于是连接池、限流、健康度这些「跟运行时/引擎绑在一起」的东西真的能跨调用复用。 /// /// # Panics /// /// 运行时建不起来（系统资源耗尽）时 panic：此时进程已无法继续，早点暴露比静默降级好。 /// 在已有 tokio 运行时上下文里调用同样会 panic（tokio 不允许嵌套 `block_on`）。 pub fn block_on(future: F) -> F::Output { - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) - .thread_name("ull-blocking") - .enable_all() - .build() - .expect("创建 tokio 运行时失败（内存/线程资源不足）"); - runtime.block_on(future) + bridge().block_on(future) } #[cfg(test)] @@ -52,4 +87,25 @@ mod tests { }); assert!(start.elapsed() >= std::time::Duration::from_millis(5)); } + + /// 共享运行时能被多个线程同时驱动（宿主侧就是多线程并行发请求）。 + #[test] + fn shared_runtime_serves_concurrent_callers() { + let threads: Vec<_> = (0..4) + .map(|index| { + std::thread::spawn(move || { + block_on(async move { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + index + }) + }) + }) + .collect(); + let mut seen: Vec = threads + .into_iter() + .map(|handle| handle.join().expect("线程不该 panic")) + .collect(); + seen.sort_unstable(); + assert_eq!(seen, vec![0, 1, 2, 3]); + } } diff --git a/crates/core/src/download/mod.rs b/crates/core/src/download/mod.rs index a66155a..291bb46 100644 --- a/crates/core/src/download/mod.rs +++ b/crates/core/src/download/mod.rs @@ -30,8 +30,12 @@ //! cache 内容寻址缓存（记录信任） atomic 落盘守卫 / 原子写 //! resume 保守续传判定 verify 流式摘要 //! progress 事件流 error 错误分类 +//! shared 进程级共享：一个数据根一个引擎（跨请求复用连接池/限流/健康度） //! ``` //! +//! 宿主侧**不要**每次请求现建引擎：`download::shared::engine_for()` 拿到的是进程内 +//! 共享的那个，连接池、信号量限流、健康度记忆因此跨请求累积（M165）。 +//! //! 落盘守卫（[`PartFile`]）中间文件名是 `.ullpart--`：`session` 是 //! 进程号，`seq` 是进程内序号。同一数据根跑两个启动器进程时不会互相搬走对方的中间文件 //! （M35 真机：并行两次 `ull install` 各失败数百个资源，`os error 2`）；旧续传材料由 @@ -56,6 +60,7 @@ pub mod progress; pub mod region; pub mod region_system; pub mod resume; +pub mod shared; pub mod types; pub mod verify; diff --git a/crates/core/src/download/shared.rs b/crates/core/src/download/shared.rs new file mode 100644 index 0000000..5644c34 --- /dev/null +++ b/crates/core/src/download/shared.rs @@ -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>`），键是**数据根的缓存目录** +//! （`/cache`）。同一数据根的多次请求拿到同一个 `Arc`； +//! 不同数据根各有一份，缓存/健康度不会串味（有测试钉住这一条）。 +//! * **活多久**：跟随进程。登记处**不做淘汰**——淘汰会让后续请求建出第二个引擎，同一个根上 +//! 同时存在两条连接池，等于把要修的问题换个地方复现。要让地区判定等「一次性派生值」重算， +//! 用 [`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>, + builds: HashMap, +} + +fn registry() -> &'static Mutex { + static REGISTRY: OnceLock> = 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) -> Arc { + 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) -> bool { + lock() + .engines + .remove(&key_of(cache_root.as_ref())) + .is_some() +} + +/// 该数据根在本进程里**实际构造过几次引擎**。 +/// +/// 这是给守卫与诊断用的：跨请求复用成立时，同一个根无论被请求多少次，这个数都停在 `1` +/// （`forget` 之后每重建一次 +1）。只断言「结果一样」测不出复用——引擎建几次都能下同一个文件。 +#[must_use] +pub fn constructions(cache_root: impl AsRef) -> 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(); + } +} diff --git a/crates/presenter/src/host/addon/http.rs b/crates/presenter/src/host/addon/http.rs index 07d73ec..aa315b6 100644 --- a/crates/presenter/src/host/addon/http.rs +++ b/crates/presenter/src/host/addon/http.rs @@ -7,6 +7,11 @@ //! | 无自定义 header（Modrinth 的全部请求） | [`DownloadEngine::download_bytes`] | **信号量限流 + 源健康度 + 指数退避 + 分类错误**，全是 core 现成的 | //! | 带 header（只有 CurseForge 的 `X-API-KEY`） | `engine.client()` 直发 | 引擎的请求体没有 header 字段，只能自己发；代价是这条路径自己带一个 429/5xx 退避 | //! +//! 两条路径用的是**同一个引擎**，而引擎是进程级共享的（M165：`host::engine::engine` → +//! `core::download::shared::engine_for`，一个数据根一个）。所以 Modrinth 与 CurseForge 的请求、 +//! 以及它们与真正的文件下载，共用同一条连接池与同一套限流；`block_on` 也跑在同一个常驻运行时上， +//! 池里的空闲连接不会随着某次调用结束而变成死连接。 +//! //! ## 密钥纪律 //! //! 带 header 的那条路径**只把 header 值喂给请求**：不进日志、不进错误串。错误串里只有 URL diff --git a/crates/presenter/src/host/addon/sources.rs b/crates/presenter/src/host/addon/sources.rs index 1dcbc1c..fb219cb 100644 --- a/crates/presenter/src/host/addon/sources.rs +++ b/crates/presenter/src/host/addon/sources.rs @@ -96,12 +96,22 @@ fn modrinth_detail( let core_engine = engine::engine(env); let transport = EngineTransport::new(&core_engine); let client = Modrinth::new(&transport); - let project = block_on(client.project(&request.project)).map_err(failure_of)?; - let versions = block_on(client.versions(&VersionQuery { + let query = VersionQuery { project: request.project.clone(), game_version: request.game_version.clone(), loader: request.loader.and_then(loader_facet).map(str::to_string), - })) + }; + // 项目详情 + 版本列表是**两次串行 GET**，包在**一次** `block_on` 里（宿主侧少一层阻塞边界， + // 也就少一次运行时进出——M165 登记的「单请求两次 block_on」在这里收掉）。 + // + // 这不是嵌套 `block_on`：两段 `await` 依次跑在同一个 future 里，落在同一个常驻运行时 + // （`core::download::blocking`）与同一个共享引擎上，连接池/限流照常复用。 + // 失败语义不变：哪一步先失败就先返回，与拆成两次调用时的先后顺序一致，分类仍走 `failure_of`。 + let (project, versions) = block_on(async { + let project = client.project(&request.project).await?; + let versions = client.versions(&query).await?; + Ok::<_, OnlineError>((project, versions)) + }) .map_err(failure_of)?; Ok(ContentDetail { project: if project.id.is_empty() { diff --git a/crates/presenter/src/host/engine.rs b/crates/presenter/src/host/engine.rs index 03bf98d..1746d99 100644 --- a/crates/presenter/src/host/engine.rs +++ b/crates/presenter/src/host/engine.rs @@ -1,40 +1,38 @@ //! 下载引擎与 Java 落点的装配（versions / install / java 三个任务共用）。 //! +//! **引擎是进程级的、按数据根共享的**（M165）：本模块不再自己 `DownloadEngine::new`，而是向 +//! [`unlucky_core::download::shared`] 要那个数据根上已经存在的引擎。于是连接池、信号量限流、 +//! 镜像健康度、内容寻址缓存都跨请求累积，而不是每次请求清零（逐项代价与边界见那个模块的文档）。 +//! //! 与 CLI `install::build_engine` 同源：内容寻址缓存落 `.ULL/cache/`，地区判定走 -//! 24h 缓存 + 环境信号（**不做网络探测**，与 D7 的「不探测时」约定一致）。 +//! 24h 缓存 + 环境信号（**不做网络探测**，与 D7 的「不探测时」约定一致）；区别只在生命周期 +//! ——CLI 一条命令一个进程，建一次就够；GUI 进程要在成百上千次请求之间复用同一份。 use std::path::PathBuf; use std::sync::Arc; -use unlucky_core::download::cache::now_ms; -use unlucky_core::download::region::{decide_from, load_cached, store_cached}; -use unlucky_core::download::region_system::detect_signals; -use unlucky_core::download::{ - ContentCache, DownloadEngine, EngineConfig, HealthRegistry, MirrorTable, -}; +use unlucky_core::download::shared; +use unlucky_core::download::DownloadEngine; use unlucky_core::game::java::vendor::JdkVendor; use unlucky_core::game::java::{self, JavaInstallRoot}; use unlucky_core::platform::paths::ProcessEnv; use super::env::HostEnv; -/// 建下载引擎（缓存/健康度/地区判定都在数据根里）。 -pub fn engine(env: &HostEnv) -> DownloadEngine { - let cache_root = env.layout.cache(); - 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)), - ) +/// 取该数据根上**共享**的下载引擎（缓存/健康度/地区判定都在数据根里）。 +/// +/// 同一个数据根无论被请求多少次，都是同一个引擎；不同数据根各一份（缓存/健康度不串味）。 +#[must_use] +pub fn engine(env: &HostEnv) -> Arc { + shared::engine_for(env.layout.cache()) +} + +/// 让这个数据根的共享引擎失效：下一次 [`engine`] 重新构造（重读地区判定/健康度/缓存索引）。 +/// +/// 在飞请求手里的 `Arc` 不受影响。设置页「重新检测地区」清掉 `region.json` 之后必须调它， +/// 否则那句话（「下次下载重新判定」）在进程级引擎下就不成立了。 +pub fn forget(env: &HostEnv) { + shared::forget(env.layout.cache()); } /// 是否国内网络（决定默认发行版顺序与镜像优先级）。 @@ -96,12 +94,23 @@ pub fn java_scan_roots(env: &HostEnv) -> Vec { mod tests { use super::*; + /// 每个用例一个独立数据根（共享登记处是进程级的，键撞了会互相干扰）。 + fn root(tag: &str) -> std::path::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-host-engine-{tag}-{}-{seq}", + std::process::id() + )); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(dir.join(".ULL").join("cache")).expect("cache"); + dir + } + #[test] fn engine_uses_the_data_root_cache() { - let dir = std::env::temp_dir().join(format!("ull-host-engine-{}", std::process::id())); - let _ = std::fs::remove_dir_all(&dir); + let dir = root("cache"); let env = HostEnv::at_root(&dir); - std::fs::create_dir_all(env.layout.cache()).expect("cache"); let engine = engine(&env); assert_eq!(engine.config().max_concurrency, 64); // 地区判定不联网：只读缓存/信号 @@ -109,10 +118,64 @@ mod tests { std::fs::remove_dir_all(&dir).ok(); } + /// **M165 的守卫**：同一 host 的多次请求只建一个引擎。 + /// + /// 断言的是**构造次数**（`shared::constructions`），不是「能不能下到文件」——每请求现建一个 + /// 引擎照样能下到文件，只有次数能把「复用」与「没复用」分开。 + /// + /// 抓得住的变异：把 `core::download::shared::engine_for` 里 `engines.get` 的命中分支删掉 + /// （每次调用都现建），本用例红在 `constructions(...) == 1`。 + #[test] + fn host_requests_share_one_engine() { + let dir = root("share"); + let env = HostEnv::at_root(&dir); + let cache_root = env.layout.cache(); + + // 三条不同「请求」路径都从 engine() 取引擎（检索 / 清单 / 安装共用这一个漏斗）。 + let first = engine(&env); + let second = engine(&env); + let third = engine(&HostEnv::at_root(&dir)); + + assert!( + Arc::ptr_eq(&first, &second) && Arc::ptr_eq(&second, &third), + "同一数据根的多次请求必须是同一个引擎" + ); + assert_eq!(shared::constructions(&cache_root), 1, "只该构造一次引擎"); + std::fs::remove_dir_all(&dir).ok(); + } + + /// 不同数据根**不共享**引擎（缓存/健康度串味比多建一个引擎严重得多）。 + #[test] + fn different_data_roots_get_their_own_engine() { + let dir_a = root("root-a"); + let dir_b = root("root-b"); + let env_a = HostEnv::at_root(&dir_a); + let env_b = HostEnv::at_root(&dir_b); + let engine_a = engine(&env_a); + let engine_b = engine(&env_b); + assert!(!Arc::ptr_eq(&engine_a, &engine_b)); + assert_eq!(shared::constructions(env_a.layout.cache()), 1); + assert_eq!(shared::constructions(env_b.layout.cache()), 1); + std::fs::remove_dir_all(&dir_a).ok(); + std::fs::remove_dir_all(&dir_b).ok(); + } + + /// `forget`（「重新检测地区」）之后下一次请求拿到新引擎。 + #[test] + fn forget_makes_the_next_request_rebuild() { + let dir = root("forget"); + let env = HostEnv::at_root(&dir); + let before = engine(&env); + forget(&env); + let after = engine(&env); + assert!(!Arc::ptr_eq(&before, &after)); + assert_eq!(shared::constructions(env.layout.cache()), 2); + std::fs::remove_dir_all(&dir).ok(); + } + #[test] fn java_merge_priority_reads_settings() { - let dir = std::env::temp_dir().join(format!("ull-host-engine-p-{}", std::process::id())); - let _ = std::fs::remove_dir_all(&dir); + let dir = root("priority"); let env = HostEnv::at_root(&dir); std::fs::create_dir_all(env.layout.config()).expect("config"); std::fs::write( diff --git a/crates/presenter/src/host/install_processors_fixture.rs b/crates/presenter/src/host/install_processors_fixture.rs index 83ae946..8a47247 100644 --- a/crates/presenter/src/host/install_processors_fixture.rs +++ b/crates/presenter/src/host/install_processors_fixture.rs @@ -36,7 +36,7 @@ pub(super) type StepView = (usize, String, Vec, usize); pub(super) struct Harness { dir: PathBuf, pub(super) env: HostEnv, - pub(super) engine: DownloadEngine, + pub(super) engine: Arc, tx: JobSender, rx: Receiver, pub(super) meter: Arc>, diff --git a/crates/presenter/src/host/settings.rs b/crates/presenter/src/host/settings.rs index 5e45361..950ff0e 100644 --- a/crates/presenter/src/host/settings.rs +++ b/crates/presenter/src/host/settings.rs @@ -371,11 +371,15 @@ fn parse_value(key: &str, value: &str) -> Result { } /// 清掉 24h 地区缓存（下次建引擎重新判定）。 +/// +/// M165 起引擎是进程级共享的：光删 `region.json` 不够（下次请求根本不会重建引擎）， +/// 必须同时让共享登记处淘汰这一份，否则「重新检测地区」这句话在本次运行里不生效。 fn detect_region(env: &HostEnv) -> Result<(), String> { let path = env.layout.cache().join("region.json"); if path.is_file() { std::fs::remove_file(&path).map_err(|error| format!("清地区缓存失败：{error}"))?; } + super::engine::forget(env); Ok(()) } diff --git a/crates/presenter/src/host/update.rs b/crates/presenter/src/host/update.rs index 79b5801..35e6ab0 100644 --- a/crates/presenter/src/host/update.rs +++ b/crates/presenter/src/host/update.rs @@ -130,7 +130,7 @@ impl Factory { #[must_use] pub fn for_env(env: &HostEnv) -> Self { Self::Default { - engine: Arc::new(engine::engine(env)), + engine: engine::engine(env), target: Target::current(), staging: staging_dir(&env.layout.cache()), dest: unlucky_core::update::process_dest(), From afe169751f753914f12c86c332e3d5f0b90332d8 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 8 Nov 2026 09:18:33 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix(m165):=20forget=20=E4=B9=8B=E5=90=8E?= =?UTF-8?q?=E8=AE=A9=E8=87=AA=E6=9B=B4=E6=96=B0=E4=B9=9F=E6=8D=A2=E5=BC=95?= =?UTF-8?q?=E6=93=8E=EF=BC=88=E5=A4=8D=E5=AE=A1=20P2-1=EF=BC=89+=20?= =?UTF-8?q?=E5=8D=A1=E5=8F=A3=E8=AE=A1=E6=95=B0=E4=B8=8E=E7=AB=AF=E5=88=B0?= =?UTF-8?q?=E7=AB=AF=E5=AE=88=E5=8D=AB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 复审 P2-1（采「修法 1」，不是只改措辞）：`shared::forget` 只让登记处不再持有 那一份，换不掉自更新 `Factory` 里常驻的 Arc —— 于是「重新检测地区」对更新路径 不生效，直到重启。现在设置动作会把 Factory 一起重建。 - presenter/update_state: `Runtime::rebuild_factory` + `CoreHost::refresh_update_factory`； 只换 factory，不动 view/inflight（在飞的 Job 派发时已克隆走装配，重建不打断它） - presenter/dispatch: `SetSetting` 成功且键是「重新检测地区」时调 refresh - presenter/engine: `DETECT_REGION_ACTION` 常量（设置页行 id / set 分派 / dispatch 判断共用）；放这里而不是 settings.rs 是因为后者已到 400 行红线 - core/shared: 文档改准 —— forget 的范围是「登记处不再持有」，长命持有者必须自己重建； 记账从 `engine_for` 移到 `DownloadEngine::new` 这个卡口（分两把锁，避免自锁）， 于是 `constructions` 连绕过登记处另造引擎的调用点也统得住 守卫（都做了变异实证）： - update_tests::detect_region_rebuilds_the_update_factory_engine：走真实命令路径 `AppCommand::SetSetting` → 断言 Factory 换了引擎且指向登记处当前实例；变异（删掉 dispatch 里那句 refresh）→ 红在 Arc::ptr_eq - update_tests::rebuilding_the_factory_does_not_touch_a_running_job：重建不清 inflight、不改快照，在飞 Job 仍能跑完 - addon/http::tests::three_host_request_paths_share_one_engine：连走三条真实宿主路径 （Modrinth 元数据 GET / CF 带 header GET / Java 页渲染）后 constructions == 1； 变异（让 get_via_engine 自己造引擎）→ 红，left:2 right:1 顺带修：settings.rs 在上一提交被推到 403 行、engine.rs 401 行，两者都超 guards.yml 的 file-size 红线（>400 直接失败）。现在 settings.rs 400 / engine.rs 399， 全仓 `*.rs` 复跑「>400」扫描无输出。 验证：core lib 608 过、5 个 download 集成 target 42 过、presenter lib 270 过、 clippy --all-targets 干净、fmt 干净。真站点仍未打（见报告「没验的部分」）。 --- crates/core/src/download/engine.rs | 2 + crates/core/src/download/shared.rs | 74 ++++++++++----- crates/presenter/src/host/addon/http.rs | 111 ++++++++++++++++++++++ crates/presenter/src/host/dispatch.rs | 6 ++ crates/presenter/src/host/engine.rs | 8 ++ crates/presenter/src/host/settings.rs | 9 +- crates/presenter/src/host/update_state.rs | 30 ++++++ crates/presenter/src/host/update_tests.rs | 91 ++++++++++++++++++ 8 files changed, 299 insertions(+), 32 deletions(-) diff --git a/crates/core/src/download/engine.rs b/crates/core/src/download/engine.rs index 1c73895..091521e 100644 --- a/crates/core/src/download/engine.rs +++ b/crates/core/src/download/engine.rs @@ -74,6 +74,8 @@ impl DownloadEngine { let http = http::build_client(&cfg).unwrap_or_else(|_| reqwest::Client::new()); let sem = Arc::new(Semaphore::new(cfg.max_concurrency.max(1))); let seed = now_ms() ^ (u64::from(cfg.max_concurrency as u32) << 32); + // 纯记账（M165）：卡在构造这个点上，才统计得到绕过登记处的调用点（守卫靠 `constructions`）。 + super::shared::record_build(cache.root()); Self { http, cfg, diff --git a/crates/core/src/download/shared.rs b/crates/core/src/download/shared.rs index 5644c34..be5231d 100644 --- a/crates/core/src/download/shared.rs +++ b/crates/core/src/download/shared.rs @@ -26,6 +26,12 @@ //! * **活多久**：跟随进程。登记处**不做淘汰**——淘汰会让后续请求建出第二个引擎，同一个根上 //! 同时存在两条连接池，等于把要修的问题换个地方复现。要让地区判定等「一次性派生值」重算， //! 用 [`forget`] 显式失效（设置页「重新检测地区」就这么做）。 +//! * **`forget` 到什么范围**：只让登记处不再持有那一份。**已经拿到 `Arc` 的人继续用旧的**： +//! 短命请求自然结束（它们的窗口只有一次下载），但**长命持有者必须自己重建**——presenter 的 +//! 自更新子系统（`host::update::Factory`）从 `CoreHost::build` 起就把引擎存在字段里， +//! 只 forget 的话它会一直用旧实例（含旧地区档案）直到进程结束。所以「重新检测地区」那条 +//! 设置动作除了 [`forget`] 还调 `CoreHost::refresh_update_factory()` 把新引擎接过去 +//! （M165 复审 P2-1：这是本模块文档原先说漏的一半）。 //! * **并发多少**：`EngineConfig::default().max_concurrency`（64）现在是**进程级**上限， //! 不再是「每请求 64」。单请求独占时与从前一致；多请求并发时它们分摊这 64 张许可。 //! * **镜像切换 / 失效**：一律走引擎原有的候选规划 + 健康度冷却，本模块不参与判定；唯一可见的 @@ -39,7 +45,7 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex, MutexGuard, OnceLock}; +use std::sync::{Arc, Mutex, OnceLock}; use super::cache::{now_ms, ContentCache}; use super::config::EngineConfig; @@ -49,22 +55,19 @@ 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>, - builds: HashMap, +/// 登记处：缓存根 → 引擎。 +fn registry() -> &'static Mutex>> { + static ENGINES: OnceLock>>> = OnceLock::new(); + ENGINES.get_or_init(|| Mutex::new(HashMap::new())) } -fn registry() -> &'static Mutex { - static REGISTRY: OnceLock> = OnceLock::new(); - REGISTRY.get_or_init(|| Mutex::new(Registry::default())) -} - -fn lock() -> MutexGuard<'static, Registry> { - registry() - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) +/// 「这个根建过几次引擎」的账本（守卫/诊断用，**不**随 [`forget`] 清零）。 +/// +/// 与 [`registry`] **分两把锁**：记账发生在 `DownloadEngine::new` 里（见 [`record_build`]）， +/// 而 [`engine_for`] 是**持着**登记处那把锁去调构造的——合成一把锁就是自己等自己。 +fn ledger() -> &'static Mutex> { + static BUILDS: OnceLock>> = OnceLock::new(); + BUILDS.get_or_init(|| Mutex::new(HashMap::new())) } /// 缓存根的键：去掉尾部分隔符（`D:/x/cache` 与 `D:/x/cache/` 视作同一根）。 @@ -85,35 +88,54 @@ fn key_of(cache_root: &Path) -> PathBuf { #[must_use] pub fn engine_for(cache_root: impl AsRef) -> Arc { let key = key_of(cache_root.as_ref()); - let mut reg = lock(); - if let Some(engine) = reg.engines.get(&key) { + let mut engines = registry() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(engine) = 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; + engines.insert(key, Arc::clone(&engine)); engine } /// 让该数据根的共享引擎失效（下次 [`engine_for`] 重新构造）。 /// -/// 在飞请求持有的 `Arc` 不受影响（它们继续用完手里那个引擎）；之后的新请求拿到新引擎， -/// 于是「重新检测地区」这类一次性派生值能真的重算。返回是否真淘汰了一份。 +/// **只让登记处不再持有那一份**：已经拿到 `Arc` 的人继续用旧的。短命请求自然结束；长命持有者 +/// （presenter 的自更新 `Factory`）必须自己重建，否则它会一直用旧实例——本仓在「重新检测地区」 +/// 的设置动作里调 `CoreHost::refresh_update_factory()` 补上这一步（见模块文档「`forget` 到什么范围」）。 +/// 返回是否真淘汰了一份。 pub fn forget(cache_root: impl AsRef) -> bool { - lock() - .engines + registry() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) .remove(&key_of(cache_root.as_ref())) .is_some() } -/// 该数据根在本进程里**实际构造过几次引擎**。 +/// 记一笔「这个缓存根上又造了一个引擎」。 +/// +/// **唯一调用点是 [`DownloadEngine::new`]**（不是 [`engine_for`]）：记账卡在构造这个卡口上， +/// 于是 [`constructions`] 统计的是**进程里真实发生过几次构造**，与「谁造的」无关。 +/// 若只在 `engine_for` 里记账，任何绕过登记处、自己 `DownloadEngine::new` 的调用点都统计不到 +/// （M165 复审备注 2 的诉求）。 +pub(crate) fn record_build(cache_root: &Path) { + *ledger() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .entry(key_of(cache_root)) + .or_default() += 1; +} + +/// 该数据根在本进程里**实际构造过几次引擎**（不论谁构造的）。 /// /// 这是给守卫与诊断用的：跨请求复用成立时，同一个根无论被请求多少次，这个数都停在 `1` /// （`forget` 之后每重建一次 +1）。只断言「结果一样」测不出复用——引擎建几次都能下同一个文件。 #[must_use] pub fn constructions(cache_root: impl AsRef) -> usize { - lock() - .builds + ledger() + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) .get(&key_of(cache_root.as_ref())) .copied() .unwrap_or(0) diff --git a/crates/presenter/src/host/addon/http.rs b/crates/presenter/src/host/addon/http.rs index aa315b6..f73368e 100644 --- a/crates/presenter/src/host/addon/http.rs +++ b/crates/presenter/src/host/addon/http.rs @@ -126,3 +126,114 @@ pub(super) fn encode(raw: &str) -> String { } out } + +#[cfg(test)] +mod tests { + use std::io::{Read as _, Write as _}; + use std::net::TcpListener; + use std::time::Duration; + + use unlucky_core::download::shared; + + use super::get; + use crate::host::env::HostEnv; + + /// 离线回环 HTTP 服务：只答固定 JSON，数「处理了几条请求 / 接了几条连接」。 + /// + /// 用 127.0.0.1 而不是真站点：**单测不许打公网**（CI 没网，也不该有）。 + fn loopback(expected_requests: usize) -> (String, std::sync::mpsc::Receiver<(usize, usize)>) { + 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(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}/v2/search"), rx) + } + + /// 一个只用来放缓存根的临时数据根（`/.ULL/cache`）。 + fn data_root(tag: &str) -> std::path::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-http-{tag}-{}-{seq}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(dir.join(".ULL").join("cache")).expect("建缓存根"); + dir + } + + /// **端到端守卫（M165 复审备注 2）**：连走三条真实宿主请求路径，只该构造一个引擎。 + /// + /// 为什么不满足于「再断言一次 `engine()` 复用」：那种断言只钉漏斗本身——将来谁在调用点 + /// 自己 `DownloadEngine::new`，它照旧全绿。这里数的是 `shared::constructions`，而记账卡在 + /// `DownloadEngine::new` 这个卡口上（与谁造的无关），所以只要有一条路径另造引擎就会红。 + /// + /// 三条路径： + /// 1. [`get`] 无 header = 内容库的 Modrinth 元数据 GET（走 `engine.download_bytes`）； + /// 2. [`get`] 带 header = CurseForge 那条（走 `engine.client()`，另一条机制）； + /// 3. `host::java::vendors` = Java 页渲染（走 `domestic()`，非 HTTP 的漏斗消费者）。 + /// + /// 没走 `versions` 清单：它的地址是写死的 launchermeta，单测里要么打真网、要么吃满连接超时 + /// —— 那正是「测试里不该发生的事」，所以选了同样真实但完全离线的三条。 + #[test] + fn three_host_request_paths_share_one_engine() { + let dir = data_root("paths"); + let env = HostEnv::at_root(&dir); + let cache_root = env.layout.cache(); + let (url, counts) = loopback(2); + + let without_headers = get(&env, &url, &[]).expect("Modrinth 元数据 GET"); + let with_headers = get(&env, &url, &[("x-api-key", "test-key")]).expect("CF 带 header GET"); + let vendors = crate::host::java::vendors(&env, &[]); + + assert_eq!( + without_headers, with_headers, + "两条路径读的是同一份回环响应" + ); + assert!(!vendors.is_empty(), "Java 页应有发行版行"); + let (_, requests) = counts + .recv_timeout(Duration::from_secs(10)) + .expect("回环服务应报告计数"); + assert_eq!(requests, 2, "两条 HTTP 路径各发一条请求"); + assert_eq!( + shared::constructions(&cache_root), + 1, + "三条宿主请求路径必须共用同一个引擎（谁绕过漏斗另造一个，这条就红）" + ); + std::fs::remove_dir_all(&dir).ok(); + } +} diff --git a/crates/presenter/src/host/dispatch.rs b/crates/presenter/src/host/dispatch.rs index a8a679d..081d9fd 100644 --- a/crates/presenter/src/host/dispatch.rs +++ b/crates/presenter/src/host/dispatch.rs @@ -219,6 +219,12 @@ impl CoreHost { } AppCommand::SetSetting { key, value } => match settings::set(&self.env, &key, &value) { Ok(()) => { + // 「重新检测地区」：`settings::set` 已经删了 `region.json` 并让共享引擎失效， + // 但自更新的 `Factory` 常驻持有旧引擎 —— 不在这条路径上重建它，自更新会继续 + // 按旧地区档案跑，直到重启（M165 复审 P2-1）。 + if key == super::engine::DETECT_REGION_ACTION { + self.refresh_update_factory(); + } self.send(CoreEvent::SettingsChanged(settings::groups(&self.env))); self.java_chain.clear(); self.refresh_java(); diff --git a/crates/presenter/src/host/engine.rs b/crates/presenter/src/host/engine.rs index 1746d99..3dc1e04 100644 --- a/crates/presenter/src/host/engine.rs +++ b/crates/presenter/src/host/engine.rs @@ -40,6 +40,14 @@ pub fn domestic(env: &HostEnv) -> bool { engine(env).region().is_domestic() } +/// 「重新检测地区」这个动作键（设置页的行 id、`settings::set` 的分派、`dispatch` 里「要不要重建 +/// 自更新 Factory」的判断共用一个真相源）。 +/// +/// 动作不是普通设置：它要清 `region.json`、让 [`forget`] 失效共享引擎，并让自更新的 `Factory` +/// 换引擎（M165 复审 P2-1）——所以三处都得认出它。常量放这里（引擎/地区这一侧），不放已经顶到 +/// 400 行红线的 `settings.rs`。 +pub(super) const DETECT_REGION_ACTION: &str = "download.action.detect_region"; + /// 设置里的 `java.vendorPriority`（空 = 按地区默认序）。 pub fn java_priority(env: &HostEnv) -> Vec { env.store() diff --git a/crates/presenter/src/host/settings.rs b/crates/presenter/src/host/settings.rs index 950ff0e..acb633e 100644 --- a/crates/presenter/src/host/settings.rs +++ b/crates/presenter/src/host/settings.rs @@ -136,7 +136,7 @@ fn download_group(env: &HostEnv) -> SettingGroupView { &text(env, "download.proxy_url"), )) .with(SettingRowView::action( - "download.action.detect_region", + super::engine::DETECT_REGION_ACTION, "重新检测地区", "清掉 24h 地区缓存，下次下载重新判定", "detect_region", @@ -322,7 +322,7 @@ fn mcp_group(env: &HostEnv) -> SettingGroupView { /// /// 未知键、值类型不符、越界、落盘失败时返回用户可读文案。 pub fn set(env: &HostEnv, key: &str, value: &str) -> Result<(), String> { - if key == "download.action.detect_region" { + if key == super::engine::DETECT_REGION_ACTION { return detect_region(env); } let parsed = parse_value(key, value)?; @@ -370,10 +370,7 @@ fn parse_value(key: &str, value: &str) -> Result { } } -/// 清掉 24h 地区缓存（下次建引擎重新判定）。 -/// -/// M165 起引擎是进程级共享的：光删 `region.json` 不够（下次请求根本不会重建引擎）， -/// 必须同时让共享登记处淘汰这一份，否则「重新检测地区」这句话在本次运行里不生效。 +/// 清掉 24h 地区缓存并让共享引擎失效（下次请求重新判定；自更新那份由 `dispatch` 重建 Factory）。 fn detect_region(env: &HostEnv) -> Result<(), String> { let path = env.layout.cache().join("region.json"); if path.is_file() { diff --git a/crates/presenter/src/host/update_state.rs b/crates/presenter/src/host/update_state.rs index 931ee5b..31105df 100644 --- a/crates/presenter/src/host/update_state.rs +++ b/crates/presenter/src/host/update_state.rs @@ -66,6 +66,27 @@ impl Runtime { &self.view } + /// 重建「怎么造 Job」的装配（M165 复审 P2-1）。 + /// + /// 自更新的 `Factory` 常驻持有下载引擎，`download::shared::forget` 换不掉它；「重新检测地区」 + /// 要生效就得把新引擎接过来。只换 `factory`，**不动** `view` / `inflight` / 迁移收据： + /// 后台在飞的那条 Job 在派发时（`begin_check` / `begin_apply` → `Job`）就已经把 factory + /// 克隆进线程了，所以重建**不会**打断正在跑的检查/应用，只影响下一次派发。 + pub fn rebuild_factory(&mut self, env: &HostEnv) { + self.factory = Factory::for_env(env); + } + + /// 装配里那个下载引擎（单测断言「换了引擎」用；`Dir` 形态没有引擎）。 + #[cfg(test)] + pub(super) fn factory_engine( + &self, + ) -> Option<&std::sync::Arc> { + match &self.factory { + Factory::Default { engine, .. } => Some(engine), + Factory::Dir { .. } => None, + } + } + /// 当前快照（克隆给事件/线程）。 #[must_use] pub fn snapshot(&self) -> UpdateView { @@ -222,6 +243,15 @@ impl CoreHost { self.send(CoreEvent::Update(Box::new(view))); } + /// 让自更新子系统手里的引擎跟上共享登记处（M165 复审 P2-1）。 + /// + /// 自更新的 `Factory` 从 `CoreHost::build` 起就把引擎存进字段，`shared::forget` 只让登记处 + /// 不再持有那一份，换不掉它。所以「重新检测地区」除了失效登记处，还必须调本方法，否则自更新 + /// 会继续按**旧地区档案**展开候选，直到进程结束。 + pub(super) fn refresh_update_factory(&mut self) { + self.update.rebuild_factory(&self.env); + } + /// 派一次检查（`manual` = 设置页点的「立即检查」；`false` = 启动时的静默检查）。 pub(super) fn start_update_check(&mut self, manual: bool) { let Some(current) = self.update.current() else { diff --git a/crates/presenter/src/host/update_tests.rs b/crates/presenter/src/host/update_tests.rs index 568127e..a0d3a69 100644 --- a/crates/presenter/src/host/update_tests.rs +++ b/crates/presenter/src/host/update_tests.rs @@ -17,6 +17,7 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; +use unlucky_core::download::region::{RegionMode, RegionProfile}; use unlucky_core::download::verify::digest_bytes; use unlucky_core::download::HashAlgo; use unlucky_core::update::{Applied, Channel, Target, Version}; @@ -242,3 +243,93 @@ fn a_higher_version_without_our_platform_is_not_up_to_date() { mod apply; #[path = "update_session_tests.rs"] mod session; + +// ---------------------------------------------------------------- M165 复审 P2-1 +// +// 这两条走的是**真实的命令路径**（`AppCommand::SetSetting` → `dispatch` → `settings::set`）， +// 不是直接调内部函数：要守的正是「动作路径上有没有把该做的事做全」。 + +/// 「重新检测地区」之后，**自更新手里的引擎也要换**。 +/// +/// 自更新的 `Factory` 从 `CoreHost::build` 起就把引擎存在字段里（常驻到进程结束）； +/// `download::shared::forget` 只让登记处不再持有那一份，**换不掉**它。于是只 forget 的话， +/// 自更新会继续按**旧地区档案**展开候选，直到重启——M165 复审 P2-1 的那个洞。 +/// +/// 抓得住的变异：删掉 `host/dispatch.rs` 的 `SetSetting` 成功分支里那句 +/// `self.refresh_update_factory()` ⇒ 本用例红在 `Arc::ptr_eq`（Factory 还是淘汰前那一份）。 +#[test] +fn detect_region_rebuilds_the_update_factory_engine() { + use std::sync::Arc; + use unlucky_core::download::shared; + + let tmp = TempDir::new("region-refresh"); + let env = HostEnv::at_root(tmp.path()); + std::fs::create_dir_all(env.layout.cache()).expect("建缓存目录"); + // 钉一份确定的地区档案：第一次构造不依赖本机时区/语言，断言才有确定的起点。 + let region_path = env.layout.cache().join("region.json"); + unlucky_core::download::region::store_cached( + ®ion_path, + &RegionProfile::pinned(RegionMode::Direct, unlucky_core::download::cache::now_ms()), + ) + .expect("写 region.json"); + + let mut host = crate::host::CoreHost::headless(env.clone()); + let cache_root = env.layout.cache(); + assert_eq!( + shared::constructions(&cache_root), + 1, + "装配时只该造一个引擎" + ); + let before = host + .update + .factory_engine() + .expect("生产装配（Default）应有引擎") + .clone(); + assert_eq!(before.region().mode, RegionMode::Direct); + + host.dispatch_command(crate::dto::AppCommand::SetSetting { + key: "download.action.detect_region".to_string(), + value: String::new(), + }); + + let after = host + .update + .factory_engine() + .expect("重建后仍该有引擎") + .clone(); + assert!( + !Arc::ptr_eq(&before, &after), + "「重新检测地区」必须换掉自更新用的引擎（否则它按旧地区档案跑到进程结束）" + ); + assert!( + Arc::ptr_eq(&after, &crate::host::engine::engine(&env)), + "自更新必须指向登记处的当前引擎，而不是被淘汰的那一份" + ); + assert_eq!( + shared::constructions(&cache_root), + 2, + "淘汰一次 = 多构造一次（既不多也不少）" + ); +} + +/// 重建 `Factory` **不打断**在飞的检查/应用：在跑的那条 `Job` 派发时就已经把装配克隆走了。 +#[test] +fn rebuilding_the_factory_does_not_touch_a_running_job() { + let tmp = TempDir::new("factory-rebuild"); + dir_source(&tmp, &manifest_for("0.9.0", b"new", "notes"), b"new"); + let env = HostEnv::at_root(tmp.path()); + let mut runtime = Runtime::with_factory(&env, factory(&tmp)); + + // 派发一条（模拟「正在查」），再重建装配： + let job = runtime.begin_check(true).expect("应能派发"); + assert!(runtime.inflight()); + runtime.rebuild_factory(&env); + assert!(runtime.inflight(), "重建装配不该清掉在飞标记"); + assert!( + matches!(runtime.view().outcome, UpdateOutcome::Checking), + "重建装配不该改状态快照" + ); + // 在飞那条 Job 仍然可用（它拿着自己那份装配），能正常跑完并回报。 + let outcome = outcome_of(job.check(&version("0.2.2"), Channel::Stable)); + assert_eq!(outcome.as_str(), "available"); +}