feat(download): 多线程下载 + 重试指数退避(issue #17)

下载主循环由串行改为「并发下载 + 串行簿记」三段式:
- Phase A:无网络前置校验(官方性/目标路径/建目录)+ 判定跳过/需下载,
  非官方 URL 在任何下载前 fail-fast
- Phase B:need-download 项经 scoped 线程池并发下载(默认并发 4,
  可配 1..=256)。worker 只做只读 &self 的 pull_one(各 URL 独立
  .part/目标文件),经 mpsc 把结果送回主线程;manifest/quarantine
  簿记与进度回调全在主线程串行执行,无需加锁。首个失败或 should_cancel
  置 cancel 标志,其余 worker 在任务边界停止
- Phase C:按 plan 顺序串行收尾——seed .hash 校验(顺序相关、可
  fail-fast)+ 构建有序结果

fail-fast 与「不发布不完整资源」不变量保留;进度事件按 URL 配对但
顺序不再单调(并发下天然如此)。

curl 重试加指数退避(网络类 200ms→400ms→800ms…上限 5s;ETXTBSY 仍走
极短退避),并发下对官方 CDN 更礼貌;退避基值 cfg(test) 下为 0 不拖慢
单测。

并发度经 OfficialUpdateConfig.download_concurrency 贯通,CLI
--download-concurrency 与 BAT_DOWNLOAD_CONCURRENCY 可配,.env 模板
与 USERGUIDE/CURRENT_STATUS/CHANGELOG 同步。

验证:新增并发正确性测试(并发 8:每 URL 恰一次 started+finished、
全部落盘)、并发度钳制、退避时长计算、CLI/env 解析单测;progress
排序测试改为顺序无关不变量;workspace 全测试(20 套件) + fmt +
clippy --all-targets -D warnings 全绿。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-18 03:25:04 -07:00
co-authored by Claude Fable 5
parent 8efd8f36b4
commit 0ab3f3b953
7 changed files with 449 additions and 78 deletions
+10 -4
View File
@@ -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
+1 -1
View File
@@ -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 <URL>` 显式指定或 `--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 误判为可复用。
+2 -1
View File
@@ -84,6 +84,7 @@ HTTPS_PROXY=http://user:pass@127.0.0.1:7890 bat --auto-discover --daemon
| `--proxy <URL\|auto\|none>` | curl 代理覆盖(默认 `auto`,从环境变量检测)。scheme 支持 http/https/socks4/socks4a/socks5/socks5h |
| `--no-proxy` | 强制直连 |
| `--unzip <PATH>` | unzip 可执行文件(默认 `unzip` |
| `--download-concurrency <N>` | 并行下载数,`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 任务后端尚未接入,当前任务历史持久化在 `<state-dir>/bat-tasks.json`
-`BAT_SKIP_ENV_FILE=1` 可让 `bat` 完全跳过 `.env` 的生成与加载。
@@ -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 <URL|auto|none> curl proxy override (default: auto from env)");
eprintln!(" --no-proxy Force direct curl connections");
eprintln!(" --unzip <PATH> unzip executable (default: unzip)");
eprintln!(" --download-concurrency <N> 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<Vec<PatchPlatform>, String> {
.collect()
}
/// 解析下载并发度:正整数,钳制到 `1..=256`。
fn parse_download_concurrency(value: &str) -> anyhow::Result<usize> {
let parsed = value
.parse::<usize>()
.map_err(|error| anyhow::anyhow!("下载并发度无效:{error}"))?;
if parsed == 0 {
return Err(anyhow::anyhow!("下载并发度必须大于 0"));
}
Ok(parsed.min(256))
}
fn parse_platform(value: &str) -> Result<PatchPlatform, String> {
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(&[
+44 -4
View File
@@ -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);
}
}
+318 -66
View File
@@ -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::<String>::new();
let mut processed_urls = HashSet::<String>::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<PlannedDownload> = 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<usize> = planned
.iter()
.enumerate()
.filter(|(_, item)| item.existing.is_none())
.map(|(index, _)| index)
.collect();
let mut download_results: Vec<Option<Result<PullOneResult, PullOneError>>> =
(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::<WorkerMessage>();
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::<String>::new();
let mut processed_urls = HashSet::<String>::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<PullOneResult>,
}
/// worker 线程经 mpsc 送回主线程的消息。
enum WorkerMessage {
/// worker 已领取某计划项、即将下载(主线程据此发 started 进度)。
Started { plan_index: usize },
/// 某计划项下载结束(成功或失败)。
Done {
plan_index: usize,
result: Result<PullOneResult, PullOneError>,
},
}
#[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<usize> = started.iter().map(|event| event.index).collect();
indices.sort_unstable();
assert_eq!(indices, (1..=expected_urls).collect::<Vec<_>>());
assert!(started.iter().all(|event| event.total == expected_urls));
// started 与 finished 覆盖同一组 URLfinished 均为已下载。
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();
+7 -2
View File
@@ -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();