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(),