Merge branch 'feat/m165-dlengine-reuse-download-engine-acro'
Guards / 依赖许可与来源(cargo-deny) (push) Canceled after 0s
Guards / 文件行数 ≤400 守门 (push) Canceled after 0s
Guards / unsafe 落点守门(集合必须恰好一个文件) (push) Canceled after 0s
Guards / design 零第三方依赖守门 (push) Canceled after 0s
Guards / presenter 零 UI 守门 (push) Canceled after 0s
Guards / core 零 UI 守门 (push) Canceled after 0s
Dev Guards / 架构与合规守门(同 main 口径) (push) Canceled after 0s
Dev Guards / fmt (rustfmt --check) (push) Canceled after 0s
Dev Guards / clippy (ubuntu) (push) Canceled after 0s
Guards / 依赖许可与来源(cargo-deny) (push) Canceled after 0s
Guards / 文件行数 ≤400 守门 (push) Canceled after 0s
Guards / unsafe 落点守门(集合必须恰好一个文件) (push) Canceled after 0s
Guards / design 零第三方依赖守门 (push) Canceled after 0s
Guards / presenter 零 UI 守门 (push) Canceled after 0s
Guards / core 零 UI 守门 (push) Canceled after 0s
Dev Guards / 架构与合规守门(同 main 口径) (push) Canceled after 0s
Dev Guards / fmt (rustfmt --check) (push) Canceled after 0s
Dev Guards / clippy (ubuntu) (push) Canceled after 0s
This commit is contained in:
commit
5e1f0868f1
15 files changed
+756
-51
No files matched your search
@@ -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** 逐条查(循环在
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<tokio::runtime::Runtime> = 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<F: Future>(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<u32> = threads
|
||||
.into_iter()
|
||||
.map(|handle| handle.join().expect("线程不该 panic"))
|
||||
.collect();
|
||||
seen.sort_unstable();
|
||||
assert_eq!(seen, vec![0, 1, 2, 3]);
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -30,8 +30,12 @@
|
||||
//! cache 内容寻址缓存(记录信任) atomic 落盘守卫 / 原子写
|
||||
//! resume 保守续传判定 verify 流式摘要
|
||||
//! progress 事件流 error 错误分类
|
||||
//! shared 进程级共享:一个数据根一个引擎(跨请求复用连接池/限流/健康度)
|
||||
//! ```
|
||||
//!
|
||||
//! 宿主侧**不要**每次请求现建引擎:`download::shared::engine_for(<cache 根>)` 拿到的是进程内
|
||||
//! 共享的那个,连接池、信号量限流、健康度记忆因此跨请求累积(M165)。
|
||||
//!
|
||||
//! 落盘守卫([`PartFile`])中间文件名是 `<name>.ullpart-<session>-<seq>`:`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;
|
||||
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
//! 进程级共享引擎:**一个数据根一个 `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`] 显式失效(设置页「重新检测地区」就这么做)。
|
||||
//! * **`forget` 到什么范围**:只让登记处不再持有那一份。**已经拿到 `Arc` 的人继续用旧的**:
|
||||
//! 短命请求自然结束(它们的窗口只有一次下载),但**长命持有者必须自己重建**——presenter 的
|
||||
//! 自更新子系统(`host::update::Factory`)从 `CoreHost::build` 起就把引擎存在字段里,
|
||||
//! 只 forget 的话它会一直用旧实例(含旧地区档案)直到进程结束。所以「重新检测地区」那条
|
||||
//! 设置动作除了 [`forget`] 还调 `CoreHost::refresh_update_factory()` 把新引擎接过去
|
||||
//! (M165 复审 P2-1:这是本模块文档原先说漏的一半)。
|
||||
//! * **并发多少**:`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, 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;
|
||||
|
||||
/// 登记处:缓存根 → 引擎。
|
||||
fn registry() -> &'static Mutex<HashMap<PathBuf, Arc<DownloadEngine>>> {
|
||||
static ENGINES: OnceLock<Mutex<HashMap<PathBuf, Arc<DownloadEngine>>>> = OnceLock::new();
|
||||
ENGINES.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
/// 「这个根建过几次引擎」的账本(守卫/诊断用,**不**随 [`forget`] 清零)。
|
||||
///
|
||||
/// 与 [`registry`] **分两把锁**:记账发生在 `DownloadEngine::new` 里(见 [`record_build`]),
|
||||
/// 而 [`engine_for`] 是**持着**登记处那把锁去调构造的——合成一把锁就是自己等自己。
|
||||
fn ledger() -> &'static Mutex<HashMap<PathBuf, usize>> {
|
||||
static BUILDS: OnceLock<Mutex<HashMap<PathBuf, usize>>> = OnceLock::new();
|
||||
BUILDS.get_or_init(|| Mutex::new(HashMap::new()))
|
||||
}
|
||||
|
||||
/// 缓存根的键:去掉尾部分隔符(`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 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));
|
||||
engines.insert(key, Arc::clone(&engine));
|
||||
engine
|
||||
}
|
||||
|
||||
/// 让该数据根的共享引擎失效(下次 [`engine_for`] 重新构造)。
|
||||
///
|
||||
/// **只让登记处不再持有那一份**:已经拿到 `Arc` 的人继续用旧的。短命请求自然结束;长命持有者
|
||||
/// (presenter 的自更新 `Factory`)必须自己重建,否则它会一直用旧实例——本仓在「重新检测地区」
|
||||
/// 的设置动作里调 `CoreHost::refresh_update_factory()` 补上这一步(见模块文档「`forget` 到什么范围」)。
|
||||
/// 返回是否真淘汰了一份。
|
||||
pub fn forget(cache_root: impl AsRef<Path>) -> bool {
|
||||
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<Path>) -> usize {
|
||||
ledger()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.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();
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
@@ -121,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)
|
||||
}
|
||||
|
||||
/// 一个只用来放缓存根的临时数据根(`<root>/.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();
|
||||
}
|
||||
}
|
||||
@@ -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() {
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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<DownloadEngine> {
|
||||
shared::engine_for(env.layout.cache())
|
||||
}
|
||||
|
||||
/// 让这个数据根的共享引擎失效:下一次 [`engine`] 重新构造(重读地区判定/健康度/缓存索引)。
|
||||
///
|
||||
/// 在飞请求手里的 `Arc` 不受影响。设置页「重新检测地区」清掉 `region.json` 之后必须调它,
|
||||
/// 否则那句话(「下次下载重新判定」)在进程级引擎下就不成立了。
|
||||
pub fn forget(env: &HostEnv) {
|
||||
shared::forget(env.layout.cache());
|
||||
}
|
||||
|
||||
/// 是否国内网络(决定默认发行版顺序与镜像优先级)。
|
||||
@@ -42,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<JdkVendor> {
|
||||
env.store()
|
||||
@@ -96,12 +102,23 @@ pub fn java_scan_roots(env: &HostEnv) -> Vec<PathBuf> {
|
||||
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 +126,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(
|
||||
|
||||
@@ -36,7 +36,7 @@ pub(super) type StepView = (usize, String, Vec<String>, usize);
|
||||
pub(super) struct Harness {
|
||||
dir: PathBuf,
|
||||
pub(super) env: HostEnv,
|
||||
pub(super) engine: DownloadEngine,
|
||||
pub(super) engine: Arc<DownloadEngine>,
|
||||
tx: JobSender,
|
||||
rx: Receiver<HostTask>,
|
||||
pub(super) meter: Arc<Mutex<Meter>>,
|
||||
|
||||
@@ -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,12 +370,13 @@ fn parse_value(key: &str, value: &str) -> Result<Value, String> {
|
||||
}
|
||||
}
|
||||
|
||||
/// 清掉 24h 地区缓存(下次建引擎重新判定)。
|
||||
/// 清掉 24h 地区缓存并让共享引擎失效(下次请求重新判定;自更新那份由 `dispatch` 重建 Factory)。
|
||||
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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -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(),
|
||||
|
||||
@@ -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<unlucky_core::download::DownloadEngine>> {
|
||||
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 {
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
Reference in new issue
Block a user