diff --git a/CHANGELOG.md b/CHANGELOG.md index de3ee33..f649e27 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,14 +6,20 @@ ## [未发布] +### 新增 +- Addressables catalog 提取 `m_Crc`（bundle IEEE CRC-32）：`ResourceEntry` 新增 `crc` 字段（compact/expanded 两种形态均解析），SQLite 持久化并对旧库幂等迁移补列；core 新增 `crc32_ieee` 与 `ResourceEntry::verify_downloaded_bytes`（按声明的 size/CRC 校验字节）（issue #2） +- UnityFS 解析新增目录条目越界校验：directory 的 `offset+size` 必须落在解压数据区内，截断/损坏 bundle 的越界目录条目不再被静默接受（issue #3） +- 官方资源下载支持多线程：并行仅作用于实际网络下载（默认并发 4，可经 `--download-concurrency` / `BAT_DOWNLOAD_CONCURRENCY` 配置为 `1..=256`），manifest/quarantine 簿记与 seed `.hash` 校验保持串行，`fail-fast` 与「不发布不完整资源」不变量不变（issue #17） + +### 修复 +- 官方下载失败重试之间加入指数退避（网络类失败 200ms→400ms→800ms…，上限 5s），并发下载时对官方 CDN 更礼貌 + ### 计划 -- [ ] 实现 Go CLI 最小可用入口（默认经 daemon RPC 或 `bat --json` 进程边界） +- [ ] 实现 `bat-api`：仿 BlueArchive 官方 API 的 Go HTTP 服务（含鉴权/签名验签，issue #19） - [ ] 官方同步结果接入 CAS + ResourceRepository 的用户级工作流 -- [ ] 实现 AssetBundle 解析器（UnityFS header/block/directory 起步） -- [ ] 继续逆向 Addressables catalog 可校验字段 +- [ ] 官方下载/导入路径接入 CRC/size 校验（复用 `verify_downloaded_bytes`） - [ ] 实现翻译系统 - [ ] 实现 Patch 引擎 -- [ ] 实现 API Server - [ ] 实现 Web 管理后台 ## [0.2.0] - 2026-07-17 diff --git a/CURRENT_STATUS.md b/CURRENT_STATUS.md index 84c41a6..c56b351 100644 --- a/CURRENT_STATUS.md +++ b/CURRENT_STATUS.md @@ -19,7 +19,7 @@ Rust 侧官方日服资源链路已经从实验验证推进到正式入口： 3. 默认平台为 `Windows + Android`。 4. 能生成官方全量 pull plan，执行真实下载，维护 release 内的 `official-download-manifest.json`。 5. 下载后使用本地 manifest 的 size + BLAKE3 校验复用文件；所有 `.zip` 在下载验收、复用、本地 audit/verify 时做 ZIP 结构校验；官方 seed `.hash` 使用标准 `xxHash32(seed=0)` 强校验（早期实现的非标准 avalanche 常量已修正）。 -6. 支持 `.part` 断点续传、失败后 clean retry、本地 manifest audit/repair、失败 staging 恢复复用、403/404/5xx 分类重试、下载 quarantine 诊断，以及旧 launcher 包官方 primary/backup CDN 切换。 +6. 支持 `.part` 断点续传、失败后 clean retry、本地 manifest audit/repair、失败 staging 恢复复用、403/404/5xx 分类重试（重试带指数退避）、下载 quarantine 诊断，以及旧 launcher 包官方 primary/backup CDN 切换。下载支持多线程：并行仅作用于实际网络下载（默认并发 4，可经 `--download-concurrency`/`BAT_DOWNLOAD_CONCURRENCY` 配置为 `1..=256`），manifest/quarantine 簿记与 seed `.hash` 校验保持串行，`fail-fast` 与「不发布不完整资源」不变量不变。 7. 支持 curl 传输层本地代理：默认自动检测 `HTTPS_PROXY` / `ALL_PROXY` / `HTTP_PROXY` 及小写环境变量（带凭据的代理推荐用环境变量配置），也可用 `--proxy ` 显式指定或 `--no-proxy` 强制直连；代理决策会写入 progress log、daemon log 和 `bat doctor` 诊断输出。代理凭据不落世界可读位置：日志/`status` 脱敏，传给 curl 经 `ALL_PROXY` 环境变量而非 argv，`--daemon` 下经环境变量下传后台子进程、不进子进程 argv 或 `bat-status.json`，复用凭据存于 `bat-proxy.secret`（`0600`）且 `clean-stable` 会清除。 8. `bat --watch` 可常驻运行，`bat --daemon` 可后台运行并用 `bat status` / `bat stop` / `bat restart` / `bat reload` / `bat logs` 管理；daemon 使用 `bat.sock` Unix socket JSON-RPC 作为 live 控制通道，PID/状态/日志文件作为快照和 fallback，`bat-events.jsonl` 记录带轮转的结构化事件日志，`bat-control.lock` 串行化控制命令；正常检查默认每 1 小时一次；远端和本地一致时默认静默，失败后默认 60 秒快速重试；CLI 默认向 stdout 输出人类可读摘要，向 stderr 输出 ASCII banner、progress log、失败分类和 quarantine 状态，需要机器输出时使用 `--json --no-progress`。 9. 远端 snapshot 未变化但输出目录为空时，会按首次运行执行全量拉取；官方 seed `.hash` 校验失败时会清理对应 manifest 条目，避免失败产物被后续本地 audit 误判为可复用。 diff --git a/USERGUIDE.md b/USERGUIDE.md index 0700c7a..d1ff78c 100644 --- a/USERGUIDE.md +++ b/USERGUIDE.md @@ -84,6 +84,7 @@ HTTPS_PROXY=http://user:pass@127.0.0.1:7890 bat --auto-discover --daemon | `--proxy ` | curl 代理覆盖（默认 `auto`，从环境变量检测）。scheme 支持 http/https/socks4/socks4a/socks5/socks5h | | `--no-proxy` | 强制直连 | | `--unzip ` | unzip 可执行文件（默认 `unzip`） | +| `--download-concurrency ` | 并行下载数，`1..=256`（默认 `4`）。并行仅作用于实际网络下载；manifest/quarantine 簿记与 seed `.hash` 校验仍串行，`fail-fast` 与「不发布不完整资源」不变量保留。可下载失败重试带指数退避以对官方 CDN 礼貌 | | `--dry-run` | 不写同步状态 | | `--plan` | dry-run 时输出计划中的 URL | | `--force` | 强制下载/刷新 | @@ -126,7 +127,7 @@ HTTPS_PROXY=http://user:pass@127.0.0.1:7890 bat --auto-discover --daemon - 优先级：**命令行参数 > 进程环境变量 > `.env` > 内置默认值**。 - 语法：每行 `KEY=VALUE`；`#` 开头为注释；值两侧成对引号会剥除；空值视为未设置。 -- 支持的键：`BAT_OUTPUT`、`BAT_STATE_DIR`、`BAT_AUTO_DISCOVER`、`BAT_WATCH`、`BAT_DAEMON`、`BAT_PROXY`、`BAT_NO_PROXY`、`BAT_INTERVAL_SECONDS`、`BAT_ERROR_RETRY_SECONDS`、`BAT_APP_VERSION`、`BAT_CONNECTION_GROUP`、`BAT_LAUNCHER_VERSION`、`BAT_PLATFORMS`、`BAT_CURL`、`BAT_UNZIP`、`BAT_JSON`、`BAT_QUIET_UP_TO_DATE`；也可以直接写 `HTTPS_PROXY` 等通用环境变量（走现有代理自动检测）。布尔值支持 `1/0/true/false/yes/no/on/off`。 +- 支持的键：`BAT_OUTPUT`、`BAT_STATE_DIR`、`BAT_AUTO_DISCOVER`、`BAT_WATCH`、`BAT_DAEMON`、`BAT_PROXY`、`BAT_NO_PROXY`、`BAT_INTERVAL_SECONDS`、`BAT_ERROR_RETRY_SECONDS`、`BAT_DOWNLOAD_CONCURRENCY`、`BAT_APP_VERSION`、`BAT_CONNECTION_GROUP`、`BAT_LAUNCHER_VERSION`、`BAT_PLATFORMS`、`BAT_CURL`、`BAT_UNZIP`、`BAT_JSON`、`BAT_QUIET_UP_TO_DATE`；也可以直接写 `HTTPS_PROXY` 等通用环境变量（走现有代理自动检测）。布尔值支持 `1/0/true/false/yes/no/on/off`。 - `BAT_WATCH` / `BAT_DAEMON` 只对无子命令的 `bat` 生效（两者同时为 `1` 时 daemon 优先）；命令行显式传入 `--watch` / `--daemon` / `--dry-run` 时 `.env` 的模式开关让位。`status` / `verify` 等子命令不受它们影响。 - `BAT_REDIS_URL` / `BAT_REDIS_PASSWORD` 为**预留键**：Redis 任务后端尚未接入，当前任务历史持久化在 `/bat-tasks.json`。 - 设 `BAT_SKIP_ENV_FILE=1` 可让 `bat` 完全跳过 `.env` 的生成与加载。 diff --git a/infrastructure/src/bin/bat_official_sync.rs b/infrastructure/src/bin/bat_official_sync.rs index 425582f..7df6be6 100644 --- a/infrastructure/src/bin/bat_official_sync.rs +++ b/infrastructure/src/bin/bat_official_sync.rs @@ -4957,6 +4957,8 @@ BAT_AUTO_DISCOVER=1 # 正常检查间隔与失败重试间隔（秒） #BAT_INTERVAL_SECONDS=3600 #BAT_ERROR_RETRY_SECONDS=60 +# 下载并发度（并行网络下载数，1~256；默认 4，对官方 CDN 礼貌） +#BAT_DOWNLOAD_CONCURRENCY=4 # ---- 网络 ---- # 显式代理 URL（支持 http/https/socks4/socks4a/socks5/socks5h）。 @@ -5141,6 +5143,9 @@ fn apply_bat_env_overrides( if let Some(v) = value("BAT_UNZIP") { options.config.unzip_command = PathBuf::from(v); } + if let Some(v) = value("BAT_DOWNLOAD_CONCURRENCY") { + options.config.download_concurrency = parse_download_concurrency(&v)?; + } if let Some(v) = value("BAT_PROXY") { options.config.curl_proxy = parse_proxy_config(&v)?; } @@ -5318,6 +5323,11 @@ fn parse_args_with_env( "--unzip" => { options.config.unzip_command = PathBuf::from(next_option_value(&mut args, &flag)?); } + "--download-concurrency" => { + let value = next_option_value(&mut args, &flag)?; + options.config.download_concurrency = parse_download_concurrency(&value)?; + options.sync_option_explicit = true; + } "--dry-run" => { options.config.dry_run = true; options.sync_option_explicit = true; @@ -5656,6 +5666,7 @@ fn print_usage(binary: &str) { eprintln!(" --proxy curl proxy override (default: auto from env)"); eprintln!(" --no-proxy Force direct curl connections"); eprintln!(" --unzip unzip executable (default: unzip)"); + eprintln!(" --download-concurrency Parallel downloads, 1..=256 (default: 4)"); eprintln!(" --dry-run Do not write sync state"); eprintln!(" --plan Include planned URLs in dry-run"); eprintln!(" --force Force download/refresh"); @@ -5730,6 +5741,17 @@ fn parse_platforms(value: &str) -> Result, String> { .collect() } +/// 解析下载并发度：正整数，钳制到 `1..=256`。 +fn parse_download_concurrency(value: &str) -> anyhow::Result { + let parsed = value + .parse::() + .map_err(|error| anyhow::anyhow!("下载并发度无效：{error}"))?; + if parsed == 0 { + return Err(anyhow::anyhow!("下载并发度必须大于 0")); + } + Ok(parsed.min(256)) +} + fn parse_platform(value: &str) -> Result { match value.to_ascii_lowercase().as_str() { "windows" | "win" => Ok(PatchPlatform::Windows), @@ -6012,6 +6034,51 @@ mod tests { assert_eq!(registry.get("task-1-2").unwrap().status, "succeeded"); } + #[test] + fn parses_download_concurrency_flag_and_env() { + // 默认值（与 OfficialUpdateConfig 默认一致）。 + assert_eq!( + parse(&["bat"]).unwrap().config.download_concurrency, + OfficialUpdateConfig::default().download_concurrency + ); + // CLI 显式设置 + 上限钳制。 + assert_eq!( + parse(&["bat", "--download-concurrency", "16"]) + .unwrap() + .config + .download_concurrency, + 16 + ); + assert_eq!( + parse(&["bat", "--download-concurrency", "9999"]) + .unwrap() + .config + .download_concurrency, + 256 + ); + // 0 与非数字报错。 + assert!(parse(&["bat", "--download-concurrency", "0"]).is_err()); + assert!(parse(&["bat", "--download-concurrency", "abc"]).is_err()); + // 环境变量（含 .env）设置默认值；CLI 覆盖之。 + assert_eq!( + parse_with_env(&["bat"], &[("BAT_DOWNLOAD_CONCURRENCY", "12")]) + .unwrap() + .config + .download_concurrency, + 12 + ); + assert_eq!( + parse_with_env( + &["bat", "--download-concurrency", "3"], + &[("BAT_DOWNLOAD_CONCURRENCY", "12")] + ) + .unwrap() + .config + .download_concurrency, + 3 + ); + } + #[test] fn parses_auto_discover_sync_args() { let options = parse(&[ diff --git a/infrastructure/src/curl_transfer.rs b/infrastructure/src/curl_transfer.rs index 927e6aa..de0224c 100644 --- a/infrastructure/src/curl_transfer.rs +++ b/infrastructure/src/curl_transfer.rs @@ -427,6 +427,34 @@ impl std::fmt::Display for CurlRetryError { impl std::error::Error for CurlRetryError {} +/// 网络类可重试失败的退避基值（毫秒）。生产 200ms、指数增长；测试下为 0 +/// 以免拖慢单测（单测仍验证退避时长的计算，只是不真正 sleep）。 +#[cfg(not(test))] +const RETRY_BACKOFF_BASE_MS: u64 = 200; +#[cfg(test)] +const RETRY_BACKOFF_BASE_MS: u64 = 0; + +/// 退避上限（毫秒）。 +const RETRY_BACKOFF_MAX_MS: u64 = 5_000; + +/// 计算第 `attempt` 次失败后、下次重试前的退避时长。 +fn retry_backoff(busy: bool, attempt: usize) -> std::time::Duration { + std::time::Duration::from_millis(backoff_delay_ms(RETRY_BACKOFF_BASE_MS, busy, attempt)) +} + +/// 退避时长（毫秒）的纯计算，便于独立于 cfg 门控的基值做单测。 +/// +/// - `ETXTBSY`（fork/exec 竞态）：极短固定退避，只为让兄弟进程完成 execve。 +/// - 其余网络类可重试失败：指数退避（`base·2^(attempt-1)`，上限 5s），并发下载 +/// 时对官方 CDN 更礼貌，避免 N 个连接失败后同时立即重发。 +fn backoff_delay_ms(base: u64, busy: bool, attempt: usize) -> u64 { + if busy { + return 5 * attempt as u64; + } + let shift = attempt.saturating_sub(1).min(5) as u32; + base.saturating_mul(1u64 << shift).min(RETRY_BACKOFF_MAX_MS) +} + pub(crate) fn run_curl_with_retry_with_proxy( curl_command: &Path, url: &str, @@ -448,8 +476,6 @@ pub(crate) fn run_curl_with_retry_with_proxy( Err(error) => CurlFailure::process_failed(url, destination, curl_command, error), }; let retryable = failure.retryable(); - // ETXTBSY 是 fork/exec 竞态窗口造成的瞬时忙，立刻重试往往仍落在同一窗口内。 - // 让出 CPU 并做一次极短退避，使持有可写句柄的兄弟进程完成其 execve。 let busy = failure.kind == CurlFailureKind::ProcessBusy; failures.push(CurlAttemptFailure { attempt, @@ -459,8 +485,8 @@ pub(crate) fn run_curl_with_retry_with_proxy( if !retryable { break; } - if busy && attempt < attempts { - std::thread::sleep(std::time::Duration::from_millis(5 * attempt as u64)); + if attempt < attempts { + std::thread::sleep(retry_backoff(busy, attempt)); } } @@ -838,4 +864,18 @@ mod tests { assert_eq!(command_env(&command, "ALL_PROXY"), Some(None)); assert_eq!(command_env(&command, "HTTPS_PROXY"), Some(None)); } + + #[test] + fn backoff_is_exponential_capped_and_busy_is_short() { + // 网络类失败：base·2^(attempt-1)，上限 5s。 + assert_eq!(backoff_delay_ms(200, false, 1), 200); + assert_eq!(backoff_delay_ms(200, false, 2), 400); + assert_eq!(backoff_delay_ms(200, false, 3), 800); + assert_eq!(backoff_delay_ms(200, false, 4), 1600); + // 高次方触顶 5s 上限。 + assert_eq!(backoff_delay_ms(200, false, 10), 5000); + // ETXTBSY 走极短固定退避，与指数无关。 + assert_eq!(backoff_delay_ms(200, true, 1), 5); + assert_eq!(backoff_delay_ms(200, true, 3), 15); + } } diff --git a/infrastructure/src/official_download.rs b/infrastructure/src/official_download.rs index fc1feaa..b291a98 100644 --- a/infrastructure/src/official_download.rs +++ b/infrastructure/src/official_download.rs @@ -16,6 +16,9 @@ use std::fs::{self, File}; use std::io::Read; use std::path::{Path, PathBuf}; use std::process::Command; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::sync::mpsc; +use std::thread; use std::time::{SystemTime, UNIX_EPOCH}; /// 官方资源下载错误：携带统一错误码，便于 CLI/RPC 归类。 @@ -61,6 +64,10 @@ const DOWNLOAD_QUARANTINE_FILE: &str = "official-download-quarantine.json"; const DOWNLOAD_MANIFEST_VERSION: u32 = 1; const DOWNLOAD_QUARANTINE_VERSION: u32 = 1; const DEFAULT_RETRY_ATTEMPTS: usize = 3; +/// 默认下载并发度。保守取值，对官方 CDN 礼貌；可经配置调到 1..=256。 +pub(crate) const DEFAULT_DOWNLOAD_CONCURRENCY: usize = 4; +/// 下载并发度上限。 +const MAX_DOWNLOAD_CONCURRENCY: usize = 256; /// Outcome for one official resource pull item. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -481,6 +488,7 @@ pub struct OfficialResourcePullService { curl_command: PathBuf, curl_proxy: CurlProxyConfig, retry_attempts: usize, + download_concurrency: usize, } impl OfficialResourcePullService { @@ -499,6 +507,7 @@ impl OfficialResourcePullService { curl_command: curl_command.into(), curl_proxy: CurlProxyConfig::default(), retry_attempts: DEFAULT_RETRY_ATTEMPTS, + download_concurrency: DEFAULT_DOWNLOAD_CONCURRENCY, } } @@ -514,6 +523,17 @@ impl OfficialResourcePullService { self } + /// 设置下载并发度（并行执行的网络下载数），钳制到 `1..=256`。 + pub fn with_download_concurrency(mut self, concurrency: usize) -> Self { + self.download_concurrency = concurrency.clamp(1, MAX_DOWNLOAD_CONCURRENCY); + self + } + + /// 返回当前下载并发度。 + pub fn download_concurrency(&self) -> usize { + self.download_concurrency + } + /// Returns the output root used for downloaded files. pub fn output_root(&self) -> &Path { &self.output_root @@ -562,29 +582,21 @@ impl OfficialResourcePullService { let mut manifest = self.read_download_manifest()?; let urls = plan.all_urls()?; let total = urls.len(); - let mut items = Vec::new(); - let mut verified_hashes = Vec::new(); - let mut verified_hash_urls = HashSet::::new(); - let mut processed_urls = HashSet::::new(); - for (offset, url) in urls.into_iter().enumerate() { - if should_cancel() { - return Err("官方资源拉取已被停止请求中断".to_string().into()); - } - - let index = offset + 1; - progress(OfficialResourcePullProgress::started( - index, - total, - url.clone(), - )); + // Phase A：无网络的前置校验（fail-fast）。逐个校验官方性、算目标路径、 + // 建目录，并判定该 URL 是「已验证可跳过」还是「需下载」。任何非官方 + // URL 在发起任何下载前即拒绝。 + if should_cancel() { + return Err("官方资源拉取已被停止请求中断".to_string().into()); + } + let mut planned: Vec = Vec::with_capacity(total); + for url in urls { if !is_official_yostar_jp_url(&url) { return Err(DownloadError::new( bat_core::ErrorCode::NON_OFFICIAL_URL, format!("拒绝下载非官方 URL：{url}"), )); } - let destination = self.destination_for_url(&url)?; if let Some(parent) = destination.parent() { ensure_safe_directory_path(parent, "下载目标目录")?; @@ -596,50 +608,193 @@ impl OfficialResourcePullService { ensure_safe_file_target(&self.output_root, &destination, "下载目标文件")?; let force_refresh = force_refresh_urls.contains(&url); - let result = if force_refresh { + let existing = if force_refresh { None } else { self.validated_existing_file(&url, &destination, &manifest)? }; - let result = if let Some(result) = result { - self.clear_quarantine_entry(&url)?; - result - } else { - let result = match self.pull_one(&url, &destination) { - Ok(result) => result, - Err(error) => { - self.record_quarantine_entry(&url, &destination, &error)?; - progress(OfficialResourcePullProgress::failed( - index, - total, - url.clone(), - &error, - )); - return Err(DownloadError::new( - error.error_code(), - format!( - "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={url} quarantine={}；{}", - self.download_quarantine_path().display(), - error.message - ), - )); + planned.push(PlannedDownload { + url, + destination, + existing, + }); + } + + // Phase B：并发下载 need-download 项（各 URL 目标/`.part` 相互独立， + // 天然可并行）。worker 只做只读 `&self` 的网络下载，经 mpsc 把结果送回 + // 主线程；manifest/quarantine 簿记与进度回调全部在主线程串行完成，无需 + // 加锁。首个失败或 `should_cancel` 会置 cancel 标志，其余 worker 在下一 + // 个任务边界停止，保持 fail-fast 与「不发布不完整资源」不变量。 + let download_indices: Vec = planned + .iter() + .enumerate() + .filter(|(_, item)| item.existing.is_none()) + .map(|(index, _)| index) + .collect(); + let mut download_results: Vec>> = + (0..planned.len()).map(|_| None).collect(); + + if !download_indices.is_empty() { + let concurrency = self.download_concurrency.clamp(1, MAX_DOWNLOAD_CONCURRENCY); + let cursor = AtomicUsize::new(0); + let cancel = AtomicBool::new(false); + let (sender, receiver) = mpsc::channel::(); + + thread::scope(|scope| { + for _ in 0..concurrency.min(download_indices.len()) { + let sender = sender.clone(); + let cursor = &cursor; + let cancel = &cancel; + let download_indices = &download_indices; + let planned = &planned; + let service = &*self; + scope.spawn(move || loop { + if cancel.load(Ordering::Relaxed) { + break; + } + let slot = cursor.fetch_add(1, Ordering::Relaxed); + let Some(&plan_index) = download_indices.get(slot) else { + break; + }; + let item = &planned[plan_index]; + if sender.send(WorkerMessage::Started { plan_index }).is_err() { + break; + } + let result = service.pull_one(&item.url, &item.destination); + if result.is_err() { + cancel.store(true, Ordering::Relaxed); + } + if sender + .send(WorkerMessage::Done { plan_index, result }) + .is_err() + { + break; + } + }); + } + // 主线程持有的 sender 副本必须丢弃，否则 receiver 永不结束。 + drop(sender); + + for message in receiver { + match message { + WorkerMessage::Started { plan_index } => { + let item = &planned[plan_index]; + progress(OfficialResourcePullProgress::started( + plan_index + 1, + total, + item.url.clone(), + )); + // 停止请求：置 cancel，让 worker 在下个任务边界退出。 + if should_cancel() { + cancel.store(true, Ordering::Relaxed); + } + } + WorkerMessage::Done { plan_index, result } => { + if let Ok(pull_result) = &result { + let item = &planned[plan_index]; + // manifest/quarantine 簿记在主线程串行执行。 + if let Err(error) = self + .clear_quarantine_entry(&item.url) + .and_then(|_| { + self.record_download_manifest_entry( + &mut manifest, + &item.url, + &item.destination, + ) + }) + .and_then(|_| self.write_download_manifest(&manifest)) + { + // 簿记失败：记为该项错误并触发 fail-fast。 + cancel.store(true, Ordering::Relaxed); + download_results[plan_index] = Some(Err(PullOneError::plain( + format!("记录下载 manifest 失败：{error}"), + ))); + continue; + } + progress(OfficialResourcePullProgress::finished( + plan_index + 1, + total, + item.url.clone(), + pull_result.status, + pull_result.bytes, + pull_result.transferred_bytes, + )); + } + download_results[plan_index] = Some(result); + } } - }; - self.clear_quarantine_entry(&url)?; - self.record_download_manifest_entry(&mut manifest, &url, &destination)?; - self.write_download_manifest(&manifest)?; - result + } + }); + + if should_cancel() && download_results.iter().any(Option::is_none) { + return Err("官方资源拉取已被停止请求中断".to_string().into()); + } + + // fail-fast：按 plan 顺序取首个失败项，记 quarantine、发 failed 进度并中止。 + for &plan_index in &download_indices { + if let Some(Some(Err(error))) = download_results.get(plan_index) { + let item = &planned[plan_index]; + self.record_quarantine_entry(&item.url, &item.destination, error)?; + progress(OfficialResourcePullProgress::failed( + plan_index + 1, + total, + item.url.clone(), + error, + )); + return Err(DownloadError::new( + error.error_code(), + format!( + "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}", + item.url, + self.download_quarantine_path().display(), + error.message + ), + )); + } + } + } + + if should_cancel() { + return Err("官方资源拉取已被停止请求中断".to_string().into()); + } + + // Phase C：按 plan 顺序串行收尾——跳过项清 quarantine 并补发进度， + // 逐项做官方 seed `.hash` 校验（顺序相关、可 fail-fast），构建有序结果。 + let mut items = Vec::with_capacity(planned.len()); + let mut verified_hashes = Vec::new(); + let mut verified_hash_urls = HashSet::::new(); + let mut processed_urls = HashSet::::new(); + for (plan_index, item) in planned.iter().enumerate() { + let result = if let Some(existing) = &item.existing { + self.clear_quarantine_entry(&item.url)?; + progress(OfficialResourcePullProgress::started( + plan_index + 1, + total, + item.url.clone(), + )); + progress(OfficialResourcePullProgress::finished( + plan_index + 1, + total, + item.url.clone(), + existing.status, + existing.bytes, + existing.transferred_bytes, + )); + *existing + } else { + // Phase B 已保证需下载项此时均为 Ok（失败会在上面 fail-fast 返回）。 + match download_results[plan_index].take() { + Some(Ok(result)) => result, + _ => { + return Err(DownloadError::new( + bat_core::ErrorCode::INTERNAL, + format!("内部错误：下载结果缺失 url={}", item.url), + )) + } + } }; - progress(OfficialResourcePullProgress::finished( - index, - total, - url.clone(), - result.status, - result.bytes, - result.transferred_bytes, - )); - processed_urls.insert(url.clone()); + processed_urls.insert(item.url.clone()); self.verify_ready_official_hashes( &official_hash_pairs, &processed_urls, @@ -648,17 +803,14 @@ impl OfficialResourcePullService { &mut manifest, )?; items.push(OfficialResourcePullItem { - url, - destination, + url: item.url.clone(), + destination: item.destination.clone(), bytes: result.bytes, transferred_bytes: result.transferred_bytes, status: result.status, }); } - if should_cancel() { - return Err("官方资源拉取已被停止请求中断".to_string().into()); - } self.verify_all_official_hashes_are_complete(&official_hash_pairs, &verified_hash_urls)?; Ok(OfficialResourcePullReport { @@ -1603,6 +1755,25 @@ fn default_download_quarantine_version() -> u32 { DOWNLOAD_QUARANTINE_VERSION } +/// Phase A 产出的单个下载计划项：URL、目标路径，以及若命中本地 manifest +/// 校验则带上「已验证可跳过」的结果（`existing`）。 +struct PlannedDownload { + url: String, + destination: PathBuf, + existing: Option, +} + +/// worker 线程经 mpsc 送回主线程的消息。 +enum WorkerMessage { + /// worker 已领取某计划项、即将下载（主线程据此发 started 进度）。 + Started { plan_index: usize }, + /// 某计划项下载结束（成功或失败）。 + Done { + plan_index: usize, + result: Result, + }, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] struct PullOneResult { bytes: u64, @@ -2853,6 +3024,8 @@ exit 22 let curl_path = bin_dir.path().join("curl"); write_fake_curl(&curl_path); + // 并发下事件顺序不固定，此处只断言与顺序无关的不变量；并发正确性另有 + // downloads_run_concurrently_and_each_url_reports_once 专测。 let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path); let plan = build_official_pull_plan_for_platforms( discovery_plan(), @@ -2868,19 +3041,98 @@ exit 22 assert_eq!(report.items.len(), expected_urls); assert_eq!(events.len(), expected_urls * 2); - assert_eq!(events[0].kind, OfficialResourcePullProgressKind::Started); - assert_eq!(events[0].index, 1); - assert_eq!(events[0].total, expected_urls); - assert_eq!(events[1].kind, OfficialResourcePullProgressKind::Finished); - assert_eq!( - events[1].status, - Some(OfficialResourcePullStatus::Downloaded) - ); + + let started: Vec<_> = events + .iter() + .filter(|event| event.kind == OfficialResourcePullProgressKind::Started) + .collect(); + let finished: Vec<_> = events + .iter() + .filter(|event| event.kind == OfficialResourcePullProgressKind::Finished) + .collect(); + assert_eq!(started.len(), expected_urls); + assert_eq!(finished.len(), expected_urls); + + // 每个 started 的 total 一致；index 覆盖 1..=N 且不重复。 + let mut indices: Vec = started.iter().map(|event| event.index).collect(); + indices.sort_unstable(); + assert_eq!(indices, (1..=expected_urls).collect::>()); + assert!(started.iter().all(|event| event.total == expected_urls)); + + // started 与 finished 覆盖同一组 URL；finished 均为已下载。 + let started_urls: HashSet<_> = started.iter().map(|event| event.url.clone()).collect(); + let finished_urls: HashSet<_> = finished.iter().map(|event| event.url.clone()).collect(); + assert_eq!(started_urls, finished_urls); + assert!(finished + .iter() + .all(|event| event.status == Some(OfficialResourcePullStatus::Downloaded))); assert!(events .iter() .any(|event| event.url.ends_with("/TableBundles/ExcelDB.db"))); } + #[test] + fn downloads_run_concurrently_and_each_url_reports_once() { + let out_dir = TempDir::new().unwrap(); + let bin_dir = TempDir::new().unwrap(); + let curl_path = bin_dir.path().join("curl"); + write_fake_curl(&curl_path); + + // 并发度 8：每个 URL 恰好一次 started + 一次 finished，全部文件落盘。 + let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path) + .with_download_concurrency(8); + assert_eq!(service.download_concurrency(), 8); + let plan = build_official_pull_plan_for_platforms( + discovery_plan(), + inventory(), + &[PatchPlatform::Windows], + ); + let all_urls = plan.all_urls().unwrap(); + let mut events = Vec::new(); + let report = service + .pull_with_progress(&plan, |event| events.push(event)) + .unwrap(); + + assert_eq!(report.items.len(), all_urls.len()); + for url in &all_urls { + let started = events + .iter() + .filter(|event| { + &event.url == url && event.kind == OfficialResourcePullProgressKind::Started + }) + .count(); + let finished = events + .iter() + .filter(|event| { + &event.url == url && event.kind == OfficialResourcePullProgressKind::Finished + }) + .count(); + assert_eq!(started, 1, "url {url} 的 started 次数"); + assert_eq!(finished, 1, "url {url} 的 finished 次数"); + } + // 所有下载都记入 manifest。 + let manifest = service.read_download_manifest().unwrap(); + assert_eq!(manifest.entries.len(), all_urls.len()); + } + + #[test] + fn download_concurrency_is_clamped() { + let service = OfficialResourcePullService::with_curl_command("/tmp/unused", "curl"); + assert_eq!(service.download_concurrency(), DEFAULT_DOWNLOAD_CONCURRENCY); + assert_eq!( + OfficialResourcePullService::with_curl_command("/tmp/unused", "curl") + .with_download_concurrency(0) + .download_concurrency(), + 1 + ); + assert_eq!( + OfficialResourcePullService::with_curl_command("/tmp/unused", "curl") + .with_download_concurrency(9999) + .download_concurrency(), + MAX_DOWNLOAD_CONCURRENCY + ); + } + #[test] fn retries_transient_download_failures() { let out_dir = TempDir::new().unwrap(); diff --git a/infrastructure/src/official_update.rs b/infrastructure/src/official_update.rs index 8d5ec34..e49f684 100644 --- a/infrastructure/src/official_update.rs +++ b/infrastructure/src/official_update.rs @@ -96,6 +96,8 @@ pub struct OfficialUpdateConfig { pub audit_local: bool, /// Repair local files when the local manifest audit fails. pub repair: bool, + /// 下载并发度（并行执行的网络下载数）。默认 4，钳制到 `1..=256`。 + pub download_concurrency: usize, } impl Default for OfficialUpdateConfig { @@ -117,6 +119,7 @@ impl Default for OfficialUpdateConfig { force: false, audit_local: true, repair: true, + download_concurrency: crate::official_download::DEFAULT_DOWNLOAD_CONCURRENCY, } } } @@ -898,7 +901,8 @@ impl OfficialUpdateService { &active_resource_root, &config.curl_command, ) - .with_proxy_config(config.curl_proxy.clone()); + .with_proxy_config(config.curl_proxy.clone()) + .with_download_concurrency(config.download_concurrency); let snapshot_path = snapshot_path_for(config, &active_resource_root); let bootstrap_cache_path = config.bootstrap_cache_path(); @@ -1337,7 +1341,8 @@ impl OfficialUpdateService { &publish_plan.staging_path, &config.curl_command, ) - .with_proxy_config(config.curl_proxy.clone()); + .with_proxy_config(config.curl_proxy.clone()) + .with_download_concurrency(config.download_concurrency); report.staging_path = Some(publish_plan.staging_path.clone()); report.snapshot_path = staging_snapshot_path.clone(); report.download_manifest = staging_fetcher.download_manifest_path();