From 16e73327b4c03c4b06c88f7d06b8b5f9345acf92 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Fri, 31 Jul 2026 13:00:53 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E8=A1=A5=E9=BD=90=E5=8F=91?= =?UTF-8?q?=E5=B8=83=E6=B8=85=E7=90=86=E4=B8=8E=E6=A0=A1=E9=AA=8C=E8=BF=9B?= =?UTF-8?q?=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- infrastructure/src/bin/bat_official_sync.rs | 219 +++++++++-- infrastructure/src/lib.rs | 5 +- infrastructure/src/official_download.rs | 388 ++++++++++++++++++-- infrastructure/src/official_update.rs | 42 ++- 4 files changed, 592 insertions(+), 62 deletions(-) diff --git a/infrastructure/src/bin/bat_official_sync.rs b/infrastructure/src/bin/bat_official_sync.rs index 1a66faf..3eda3f3 100644 --- a/infrastructure/src/bin/bat_official_sync.rs +++ b/infrastructure/src/bin/bat_official_sync.rs @@ -11,14 +11,15 @@ use bat_infrastructure::{ read_parse_cache_at, read_snapshot, read_textunit_index_at, read_version_state, redact_proxy_url, resolve_curl_proxy, validate_output_root, validate_runtime_state_dir, write_file_atomic, CurlProxyConfig, CurlProxyMode, OfficialEndpointMarkerRole, - OfficialFailedVersionRecord, OfficialServerInfoSource, OfficialTextUnitQuery, - OfficialUpdateConfig, OfficialUpdateProgress, OfficialUpdateReport, OfficialUpdateService, - OfficialUpdateSnapshot, OfficialUpdateStatus, OfficialVerificationSummary, - OfficialVersionRecord, OfficialVersionState, PatchApplyKind, PatchApplyParams, - PatchApplyReport, SqliteResourceRepository, UnityFsFieldPatchParams, UnityFsPatchReport, - UnityFsStringFieldPatchParams, UnityFsTextAssetPatchParams, LOCALIZED_CURRENT_LINK, - LOCALIZED_PATCH_MANIFEST_FILE, LOCALIZED_VERSIONS_DIR, LOCALIZED_VERSION_STATE_FILE, - OFFICIAL_PARSE_CACHE_FILE, OFFICIAL_TEXTUNIT_INDEX_FILE, PRIVATE_FILE_MODE, + OfficialFailedVersionRecord, OfficialResourceHashVerification, OfficialResourceVerification, + OfficialServerInfoSource, OfficialTextUnitQuery, OfficialUpdateConfig, OfficialUpdateProgress, + OfficialUpdateReport, OfficialUpdateService, OfficialUpdateSnapshot, OfficialUpdateStatus, + OfficialVerificationSummary, OfficialVersionRecord, OfficialVersionState, PatchApplyKind, + PatchApplyParams, PatchApplyReport, SqliteResourceRepository, UnityFsFieldPatchParams, + UnityFsPatchReport, UnityFsStringFieldPatchParams, UnityFsTextAssetPatchParams, + LOCALIZED_CURRENT_LINK, LOCALIZED_PATCH_MANIFEST_FILE, LOCALIZED_VERSIONS_DIR, + LOCALIZED_VERSION_STATE_FILE, OFFICIAL_PARSE_CACHE_FILE, OFFICIAL_TEXTUNIT_INDEX_FILE, + PRIVATE_FILE_MODE, }; use serde::de::DeserializeOwned; use serde::{Deserialize, Serialize}; @@ -48,6 +49,7 @@ const DAEMON_LOG_FILE: &str = "bat-daemon.log"; const DAEMON_STRUCTURED_LOG_FILE: &str = "bat-events.jsonl"; const DAEMON_SOCKET_FILE: &str = "bat.sock"; const DAEMON_CONTROL_LOCK_FILE: &str = "bat-control.lock"; +const SHUTDOWN_SIGNAL_POLL_MILLISECONDS: u64 = 100; /// 后台进程保存代理凭据的专用文件，仅供 restart/reload 复用，永不序列化到状态输出。 const DAEMON_PROXY_SECRET_FILE: &str = "bat-proxy.secret"; /// 代理凭据下传子进程使用的环境变量；避免凭据出现在子进程 argv（/proc//cmdline）。 @@ -354,21 +356,22 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { if options.daemon_child { validate_runtime_state_dir(&options.state_dir).map_err(anyhow::Error::msg)?; } + install_shutdown_signal_handlers()?; + clear_shutdown_signal_request(); let service = OfficialUpdateService::new(); let mut logger = ProgressLogger::new(options.progress); let daemon_state_dir = options.state_dir.clone(); if options.daemon_child { logger.attach_structured_log(daemon_structured_log_path(&daemon_state_dir)); } - let daemon_control = if options.daemon_child { - Some(new_daemon_control()) - } else { - None - }; + // 前台 watch 也使用控制状态，使信号停止和 daemon/RPC 停止共享同一条退出路径。 + let daemon_control = Some(new_daemon_control()); // 进程内同步锁：watch 循环与任务 worker 在跑同步前都获取它，互相等待而非撞文件锁失败。 let sync_lock = Arc::new(Mutex::new(())); - let (_task_worker, _task_context, _rpc_server) = if let Some(control) = daemon_control.as_ref() - { + let (_task_worker, _task_context, _rpc_server) = if options.daemon_child { + let control = daemon_control + .as_ref() + .expect("daemon control must exist for daemon child"); // 任务历史持久化在 state dir（此前已通过 validate_runtime_state_dir 校验）。 let (registry, restore_summary) = TaskRegistry::with_persistence(&daemon_state_dir); logger.log_text("daemon", format!("任务历史：{restore_summary}")); @@ -403,7 +406,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { ), ); loop { - if daemon_control_stop_requested(daemon_control.as_ref()) { + if watch_stop_requested(daemon_control.as_ref()) { logger.log_text("daemon", "收到停止请求，watch 循环准备退出"); break; } @@ -472,7 +475,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { ); logger.log(event); }, - || daemon_control_stop_requested(daemon_control.as_ref()), + || watch_stop_requested(daemon_control.as_ref()), ) }; match run_result { @@ -566,7 +569,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { next_retry_seconds: Some(sleep_for.as_secs()), }; eprintln!("{}", serde_json::to_string_pretty(&payload)?); - if daemon_control_stop_requested(daemon_control.as_ref()) { + if watch_stop_requested(daemon_control.as_ref()) { logger.log_text("daemon", "停止请求已中断本轮同步，watch 循环准备退出"); break; } @@ -576,7 +579,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { match wait_for_daemon_wake(daemon_control.as_ref(), sleep_for) { DaemonWake::Timeout => {} DaemonWake::Stop => { - logger.log_text("daemon", "收到停止请求，watch 循环准备退出"); + logger.log_text("daemon", "收到停止请求或进程信号，watch 循环准备退出"); break; } DaemonWake::Refresh { force } => { @@ -672,6 +675,10 @@ struct DaemonDownloadProgress { failure_attempts: Option, #[serde(default)] quarantined: Option, + #[serde(default)] + verification: Option, + #[serde(default)] + official_hash: Option, } #[derive(Debug)] @@ -1737,10 +1744,68 @@ fn lexically_normalize_path(path: &Path) -> PathBuf { } } +#[cfg(unix)] +static SHUTDOWN_SIGNAL_REQUESTED: AtomicBool = AtomicBool::new(false); + +#[cfg(unix)] +extern "C" fn handle_shutdown_signal(_signal: libc::c_int) { + // Signal handlers may only touch lock-free process state; the watch loop + // performs logging, cancellation, and cleanup after observing this flag. + SHUTDOWN_SIGNAL_REQUESTED.store(true, Ordering::SeqCst); +} + +#[cfg(unix)] +fn install_shutdown_signal_handlers() -> anyhow::Result<()> { + unsafe { + if libc::signal( + libc::SIGINT, + handle_shutdown_signal as *const () as libc::sighandler_t, + ) == libc::SIG_ERR + { + return Err(anyhow::anyhow!("注册 SIGINT 优雅退出处理器失败")); + } + if libc::signal( + libc::SIGTERM, + handle_shutdown_signal as *const () as libc::sighandler_t, + ) == libc::SIG_ERR + { + return Err(anyhow::anyhow!("注册 SIGTERM 优雅退出处理器失败")); + } + } + Ok(()) +} + +#[cfg(not(unix))] +fn install_shutdown_signal_handlers() -> anyhow::Result<()> { + Ok(()) +} + +#[cfg(unix)] +fn clear_shutdown_signal_request() { + SHUTDOWN_SIGNAL_REQUESTED.store(false, Ordering::SeqCst); +} + +#[cfg(not(unix))] +fn clear_shutdown_signal_request() {} + +#[cfg(unix)] +fn shutdown_signal_requested() -> bool { + SHUTDOWN_SIGNAL_REQUESTED.load(Ordering::SeqCst) +} + +#[cfg(not(unix))] +fn shutdown_signal_requested() -> bool { + false +} + fn new_daemon_control() -> DaemonControl { Arc::new((Mutex::new(DaemonControlState::default()), Condvar::new())) } +fn watch_stop_requested(control: Option<&DaemonControl>) -> bool { + shutdown_signal_requested() || daemon_control_stop_requested(control) +} + fn daemon_control_stop_requested(control: Option<&DaemonControl>) -> bool { let Some(control) = control else { return false; @@ -1781,21 +1846,42 @@ fn daemon_control_request_reload(control: &DaemonControl) { } fn wait_for_daemon_wake(control: Option<&DaemonControl>, timeout: Duration) -> DaemonWake { + let started_at = Instant::now(); let Some(control) = control else { - thread::sleep(timeout); - return DaemonWake::Timeout; + loop { + if shutdown_signal_requested() { + return DaemonWake::Stop; + } + let elapsed = started_at.elapsed(); + if elapsed >= timeout { + return DaemonWake::Timeout; + } + let remaining = timeout.saturating_sub(elapsed); + thread::sleep(remaining.min(Duration::from_millis(SHUTDOWN_SIGNAL_POLL_MILLISECONDS))); + } }; let (lock, cvar) = &**control; let Ok(mut state) = lock.lock() else { return DaemonWake::Stop; }; - if let Some(wake) = take_daemon_wake(&mut state) { - return wake; + loop { + if shutdown_signal_requested() { + return DaemonWake::Stop; + } + if let Some(wake) = take_daemon_wake(&mut state) { + return wake; + } + let elapsed = started_at.elapsed(); + if elapsed >= timeout { + return DaemonWake::Timeout; + } + let remaining = timeout.saturating_sub(elapsed); + let wait_for = remaining.min(Duration::from_millis(SHUTDOWN_SIGNAL_POLL_MILLISECONDS)); + let Ok((next_state, _timeout)) = cvar.wait_timeout(state, wait_for) else { + return DaemonWake::Stop; + }; + state = next_state; } - let Ok((mut state, _timeout)) = cvar.wait_timeout(state, timeout) else { - return DaemonWake::Stop; - }; - take_daemon_wake(&mut state).unwrap_or(DaemonWake::Timeout) } fn take_daemon_wake(state: &mut DaemonControlState) -> Option { @@ -4499,6 +4585,18 @@ fn print_list(label: &str, values: &[String], limit: usize) { fn format_daemon_download_progress(progress: &DaemonDownloadProgress) -> String { let status = progress.status.as_deref().unwrap_or("running"); + if let Some(hash) = progress.official_hash.as_ref() { + return format!( + "{}/{} official_hash algorithm={} expected={} actual={} data={} hash={}", + progress.index, + progress.total, + hash.algorithm.as_str(), + hash.expected, + hash.actual, + hash.data_url, + hash.hash_url + ); + } if status == "failed" { return format!( "{}/{} failed kind={} http={} retryable={} attempts={} quarantined={} {}", @@ -4524,6 +4622,19 @@ fn format_daemon_download_progress(progress: &DaemonDownloadProgress) -> String progress.url ); } + if let Some(verification) = progress.verification.as_ref() { + return format!( + "{}/{} {} bytes={} blake3={} zip_checked={} zip_verified={} {}", + progress.index, + progress.total, + status, + verification.actual_bytes, + verification.actual_blake3, + verification.zip_checked, + verification.zip_structure_verified, + progress.url + ); + } format!( "{}/{} {} {}", progress.index, progress.total, status, progress.url @@ -5631,6 +5742,8 @@ fn update_daemon_progress(state_dir: &Path, event: &OfficialUpdateProgress) -> a failure_retryable: event.download_failure_retryable, failure_attempts: event.download_failure_attempts, quarantined: event.download_quarantined, + verification: event.download_verification.clone(), + official_hash: event.official_hash_verification.clone(), }), _ if event.stage != "download" => None, _ => status.download_progress, @@ -6169,6 +6282,8 @@ impl RotatingStructuredLogger { "failure_retryable": event.download_failure_retryable, "failure_attempts": event.download_failure_attempts, "quarantined": event.download_quarantined, + "verification": event.download_verification, + "official_hash": event.official_hash_verification, })), }); let mut line = serde_json::to_vec(&payload)?; @@ -9020,6 +9135,31 @@ mod tests { ); } + #[cfg(unix)] + #[test] + fn shutdown_signals_wake_watch_without_mutating_daemon_control() { + clear_shutdown_signal_request(); + let control = new_daemon_control(); + + handle_shutdown_signal(libc::SIGINT); + assert!(watch_stop_requested(Some(&control))); + assert_eq!( + wait_for_daemon_wake(Some(&control), Duration::from_secs(60)), + DaemonWake::Stop + ); + + clear_shutdown_signal_request(); + handle_shutdown_signal(libc::SIGTERM); + assert!(watch_stop_requested(Some(&control))); + assert_eq!( + wait_for_daemon_wake(Some(&control), Duration::from_secs(60)), + DaemonWake::Stop + ); + + clear_shutdown_signal_request(); + assert!(!watch_stop_requested(Some(&control))); + } + #[test] fn daemon_control_lock_rejects_active_owner() { let temp = tempfile::TempDir::new().unwrap(); @@ -9151,6 +9291,21 @@ mod tests { event.download_index = Some(1); event.download_total = Some(2); event.download_url = Some("https://prod-clientpatch.bluearchiveyostar.com/a.zip".into()); + event.download_verification = Some(OfficialResourceVerification { + expected_bytes: Some(123), + actual_bytes: 123, + expected_blake3: Some("expected-blake3".to_string()), + actual_blake3: "expected-blake3".to_string(), + zip_checked: true, + zip_structure_verified: true, + }); + event.official_hash_verification = Some(OfficialResourceHashVerification { + data_url: "https://prod-clientpatch.bluearchiveyostar.com/a.bytes".to_string(), + hash_url: "https://prod-clientpatch.bluearchiveyostar.com/a.hash".to_string(), + algorithm: bat_infrastructure::OfficialResourceHashAlgorithm::XxHash32Decimal, + expected: "2044170421".to_string(), + actual: "2044170421".to_string(), + }); for _ in 0..8 { logger @@ -9170,6 +9325,16 @@ mod tests { assert_eq!(value["stage"], "download"); assert_eq!(value["download"]["index"], 1); assert_eq!(value["download"]["total"], 2); + assert_eq!(value["download"]["verification"]["actual_bytes"], 123); + assert_eq!( + value["download"]["verification"]["zip_structure_verified"], + true + ); + assert_eq!( + value["download"]["official_hash"]["algorithm"], + "xxhash32_decimal" + ); + assert_eq!(value["download"]["official_hash"]["expected"], "2044170421"); } #[cfg(unix)] diff --git a/infrastructure/src/lib.rs b/infrastructure/src/lib.rs index f1f70b3..5005699 100644 --- a/infrastructure/src/lib.rs +++ b/infrastructure/src/lib.rs @@ -59,9 +59,10 @@ pub use official_download::{ read_download_manifest_at, DownloadError, OfficialDownloadManifest, OfficialDownloadManifestEntry, OfficialLocalManifestAuditItem, OfficialLocalManifestAuditReport, OfficialLocalManifestAuditStatus, - OfficialLocalVerificationReport, OfficialResourcePullItem, OfficialResourcePullProgress, + OfficialLocalVerificationReport, OfficialResourceHashAlgorithm, + OfficialResourceHashVerification, OfficialResourcePullItem, OfficialResourcePullProgress, OfficialResourcePullProgressKind, OfficialResourcePullReport, OfficialResourcePullService, - OfficialResourcePullStatus, + OfficialResourcePullStatus, OfficialResourceVerification, }; pub use official_game_main_config::OfficialGameMainConfigBootstrapService; pub use official_launcher::{ diff --git a/infrastructure/src/official_download.rs b/infrastructure/src/official_download.rs index abd7094..f794a65 100644 --- a/infrastructure/src/official_download.rs +++ b/infrastructure/src/official_download.rs @@ -91,6 +91,8 @@ pub enum OfficialResourcePullProgressKind { Started, /// A URL finished as skipped, resumed, or downloaded. Finished, + /// An official sidecar hash pair passed verification. + Verification, /// A URL failed after retry classification and was quarantined for this run. Failed, } @@ -101,11 +103,29 @@ impl OfficialResourcePullProgressKind { match self { Self::Started => "started", Self::Finished => "finished", + Self::Verification => "verification", Self::Failed => "failed", } } } +/// Local validation results for one completed resource. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct OfficialResourceVerification { + /// Expected byte size from the local manifest, when reusing a file. + pub expected_bytes: Option, + /// Actual byte size observed after the transfer. + pub actual_bytes: u64, + /// Expected BLAKE3 from the local manifest, when reusing a file. + pub expected_blake3: Option, + /// Actual BLAKE3 computed from the completed file. + pub actual_blake3: String, + /// Whether ZIP structure validation was required for this resource. + pub zip_checked: bool, + /// Whether the required ZIP structure validation passed. + pub zip_structure_verified: bool, +} + /// Progress for one URL in an official pull plan. #[derive(Debug, Clone, PartialEq, Eq)] pub struct OfficialResourcePullProgress { @@ -128,6 +148,10 @@ pub struct OfficialResourcePullProgress { pub bytes: Option, /// Bytes transferred during this run, when `kind` is `Finished`. pub transferred_bytes: Option, + /// Local size/BLAKE3/ZIP validation for a completed URL. + pub verification: Option, + /// Official sidecar hash verification completed after this URL. + pub official_hash: Option, /// Stable failure kind label, when `kind` is `Failed`. pub failure_kind: Option, /// HTTP status parsed from curl stderr, when available. @@ -150,6 +174,8 @@ impl OfficialResourcePullProgress { status: None, bytes: None, transferred_bytes: None, + verification: None, + official_hash: None, failure_kind: None, failure_http_status: None, failure_retryable: None, @@ -165,6 +191,7 @@ impl OfficialResourcePullProgress { status: OfficialResourcePullStatus, bytes: u64, transferred_bytes: u64, + verification: OfficialResourceVerification, ) -> Self { Self { kind: OfficialResourcePullProgressKind::Finished, @@ -174,6 +201,32 @@ impl OfficialResourcePullProgress { status: Some(status), bytes: Some(bytes), transferred_bytes: Some(transferred_bytes), + verification: Some(verification), + official_hash: None, + failure_kind: None, + failure_http_status: None, + failure_retryable: None, + failure_attempts: None, + quarantined: false, + } + } + + fn verification( + index: usize, + total: usize, + url: String, + official_hash: OfficialResourceHashVerification, + ) -> Self { + Self { + kind: OfficialResourcePullProgressKind::Verification, + index, + total, + url, + status: None, + bytes: None, + transferred_bytes: None, + verification: None, + official_hash: Some(official_hash), failure_kind: None, failure_http_status: None, failure_retryable: None, @@ -191,6 +244,8 @@ impl OfficialResourcePullProgress { status: None, bytes: None, transferred_bytes: None, + verification: None, + official_hash: None, failure_kind: error.failure_kind().map(ToOwned::to_owned), failure_http_status: error.http_status(), failure_retryable: error.retryable(), @@ -201,9 +256,10 @@ impl OfficialResourcePullProgress { } /// Official hash algorithm used by a verified resource sidecar. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub enum OfficialResourceHashAlgorithm { /// Decimal text form of `xxHash32` with seed `0`. + #[serde(rename = "xxhash32_decimal")] XxHash32Decimal, } @@ -217,7 +273,7 @@ impl OfficialResourceHashAlgorithm { } /// Successful official hash verification for one downloaded resource. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct OfficialResourceHashVerification { /// Downloaded data URL that was verified. pub data_url: String, @@ -534,6 +590,40 @@ impl OfficialResourcePullService { self.output_root.join(DOWNLOAD_QUARANTINE_FILE) } + /// Removes resource files recorded by an older manifest but absent from + /// the current pull plan. + /// + /// This is intentionally limited to destinations owned by the previous + /// download manifest. Management files such as snapshots, version state, + /// and quarantine records are left untouched. + pub fn prune_stale_manifest_entries( + &self, + plan: &OfficialResourcePullPlan, + ) -> Result { + self.ensure_output_root_ready()?; + let expected_urls = plan.all_urls()?.into_iter().collect::>(); + let mut manifest = self.read_download_manifest()?; + let stale_entries = manifest + .entries + .iter() + .filter(|(url, _)| !expected_urls.contains(*url)) + .map(|(url, entry)| (url.clone(), entry.destination.clone())) + .collect::>(); + + for (url, relative_destination) in &stale_entries { + let destination = self.output_root.join(relative_destination); + self.remove_managed_destination(&destination, url)?; + self.remove_managed_destination(&partial_path_for(&destination), url)?; + manifest.entries.remove(url); + } + + if !stale_entries.is_empty() { + self.write_download_manifest(&manifest)?; + } + + Ok(stale_entries.len()) + } + /// Executes the plan and downloads every official URL to disk. pub fn pull( &self, @@ -630,8 +720,8 @@ impl OfficialResourcePullService { let result = self.pull_one(&item.url, &item.destination); match result { - Ok(pull_result) => { - if let Err(error) = self + Ok(mut pull_result) => { + let verification_result = self .clear_quarantine_entry(&item.url) .and_then(|_| { self.record_download_manifest_entry( @@ -640,17 +730,25 @@ impl OfficialResourcePullService { &item.destination, ) }) - .and_then(|_| self.write_download_manifest(&manifest)) - { - let error = PullOneError::plain(format!("记录下载 manifest 失败：{error}")); - self.record_quarantine_entry(&item.url, &item.destination, &error)?; - progress(OfficialResourcePullProgress::failed( - completed_count, - total, - item.url.clone(), - &error, - )); - return Err(DownloadError::new( + .and_then(|verification| { + self.write_download_manifest(&manifest) + .map(|_| verification) + }); + match verification_result { + Ok(verification) => { + pull_result.verification = verification; + } + Err(error) => { + let error = + PullOneError::plain(format!("记录下载 manifest 失败：{error}")); + self.record_quarantine_entry(&item.url, &item.destination, &error)?; + progress(OfficialResourcePullProgress::failed( + completed_count, + total, + item.url.clone(), + &error, + )); + return Err(DownloadError::new( error.error_code(), format!( "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}", @@ -659,6 +757,7 @@ impl OfficialResourcePullService { error.message ), )); + } } completed_count += 1; @@ -669,6 +768,7 @@ impl OfficialResourcePullService { pull_result.status, pull_result.bytes, pull_result.transferred_bytes, + pull_result.verification.clone(), )); download_results[plan_index] = Some(Ok(pull_result)); } @@ -719,8 +819,9 @@ impl OfficialResourcePullService { existing.status, existing.bytes, existing.transferred_bytes, + existing.verification.clone(), )); - *existing + existing.clone() } else { // Phase B 已保证需下载项此时均为 Ok（失败会在上面 fail-fast 返回）。 match download_results[plan_index].take() { @@ -735,13 +836,21 @@ impl OfficialResourcePullService { }; processed_urls.insert(item.url.clone()); - self.verify_ready_official_hashes( + let newly_verified_hashes = self.verify_ready_official_hashes( &official_hash_pairs, &processed_urls, &mut verified_hash_urls, &mut verified_hashes, &mut manifest, )?; + for verification in newly_verified_hashes { + progress(OfficialResourcePullProgress::verification( + completed_count, + total, + verification.data_url.clone(), + verification, + )); + } items.push(OfficialResourcePullItem { url: item.url.clone(), destination: item.destination.clone(), @@ -960,6 +1069,9 @@ impl OfficialResourcePullService { .map_err(PullOneError::plain)? .ok_or_else(|| format!("下载完成后目标文件缺失 {}", destination.display())) .map_err(PullOneError::plain)?; + let verification = self + .local_verification(url, destination, None, None) + .map_err(PullOneError::plain)?; Ok(PullOneResult { bytes, @@ -969,6 +1081,7 @@ impl OfficialResourcePullService { bytes }, status, + verification, }) } @@ -1010,7 +1123,8 @@ impl OfficialResourcePullService { verified_hash_urls: &mut HashSet, verified_hashes: &mut Vec, manifest: &mut OfficialDownloadManifest, - ) -> Result<(), String> { + ) -> Result, String> { + let mut newly_verified = Vec::new(); for pair in pairs { if verified_hash_urls.contains(&pair.hash_url) { continue; @@ -1040,11 +1154,12 @@ impl OfficialResourcePullService { } }; + newly_verified.push(verification.clone()); verified_hashes.push(verification); verified_hash_urls.insert(pair.hash_url.clone()); } - Ok(()) + Ok(newly_verified) } fn hash_pair_files_exist(&self, pair: &OfficialSeedHashPair) -> Result { @@ -1124,6 +1239,41 @@ impl OfficialResourcePullService { Ok(manifest) } + fn remove_managed_destination(&self, path: &Path, url: &str) -> Result<(), String> { + ensure_path_within_root(&self.output_root, path)?; + ensure_safe_file_target(&self.output_root, path, "旧官方资源清理目标")?; + + match fs::symlink_metadata(path) { + Ok(metadata) if metadata.file_type().is_symlink() => Err(format!( + "旧官方资源清理目标不能是 symlink：url={url} path={}", + path.display() + )), + Ok(metadata) if metadata.is_file() => { + fs::remove_file(path).map_err(|error| { + format!( + "清理旧官方资源失败：url={url} path={} error={error}", + path.display() + ) + })?; + remove_empty_parent_directories(&self.output_root, path.parent())?; + Ok(()) + } + Ok(metadata) if metadata.is_dir() => Err(format!( + "旧官方资源清理目标不能是目录：url={url} path={}", + path.display() + )), + Ok(_) => Err(format!( + "旧官方资源清理目标不是普通文件：url={url} path={}", + path.display() + )), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(format!( + "检查旧官方资源清理目标失败：url={url} path={} error={error}", + path.display() + )), + } + } + fn write_download_manifest(&self, manifest: &OfficialDownloadManifest) -> Result<(), String> { self.ensure_output_root_ready()?; let path = self.download_manifest_path(); @@ -1253,23 +1403,52 @@ impl OfficialResourcePullService { return Ok(None); } + let verification = self.local_verification( + url, + destination, + Some(entry.bytes), + Some(entry.blake3.clone()), + )?; Ok(Some(PullOneResult { bytes, transferred_bytes: 0, status: OfficialResourcePullStatus::SkippedExisting, + verification, })) } + fn local_verification( + &self, + url: &str, + destination: &Path, + expected_bytes: Option, + expected_blake3: Option, + ) -> Result { + let actual_bytes = file_len_if_exists(destination)? + .ok_or_else(|| format!("读取资源文件信息失败 {}", destination.display()))?; + let actual_blake3 = blake3_file_hex(destination)?; + let zip_checked = url_or_path_has_zip_extension(url) || path_has_zip_extension(destination); + if zip_checked { + self.validate_zip_if_needed(url, destination)?; + } + + Ok(OfficialResourceVerification { + expected_bytes, + actual_bytes, + expected_blake3, + actual_blake3, + zip_checked, + zip_structure_verified: zip_checked, + }) + } + fn record_download_manifest_entry( &self, manifest: &mut OfficialDownloadManifest, url: &str, destination: &Path, - ) -> Result<(), String> { - self.validate_zip_if_needed(url, destination)?; - let bytes = file_len_if_exists(destination)? - .ok_or_else(|| format!("读取已下载文件信息失败 {}", destination.display()))?; - let digest = blake3_file_hex(destination)?; + ) -> Result { + let mut verification = self.local_verification(url, destination, None, None)?; let relative_destination = self.relative_destination(destination)?; manifest.entries.insert( @@ -1277,12 +1456,14 @@ impl OfficialResourcePullService { OfficialDownloadManifestEntry { url: url.to_string(), destination: relative_destination, - bytes, - blake3: digest, + bytes: verification.actual_bytes, + blake3: verification.actual_blake3.clone(), }, ); - Ok(()) + verification.expected_bytes = Some(verification.actual_bytes); + verification.expected_blake3 = Some(verification.actual_blake3.clone()); + Ok(verification) } fn validate_zip_if_needed(&self, url: &str, destination: &Path) -> Result<(), String> { @@ -1703,11 +1884,12 @@ struct PlannedDownload { existing: Option, } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq)] struct PullOneResult { bytes: u64, transferred_bytes: u64, status: OfficialResourcePullStatus, + verification: OfficialResourceVerification, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -1788,6 +1970,48 @@ fn partial_path_for(destination: &Path) -> PathBuf { PathBuf::from(partial) } +fn remove_empty_parent_directories(root: &Path, parent: Option<&Path>) -> Result<(), String> { + let mut current = parent; + while let Some(path) = current { + if path == root { + break; + } + + let metadata = match fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => { + current = path.parent(); + continue; + } + Err(error) => { + return Err(format!( + "检查旧官方资源父目录失败 {}：{error}", + path.display() + )); + } + }; + if metadata.file_type().is_symlink() { + return Err(format!( + "旧官方资源父目录不能是 symlink：{}", + path.display() + )); + } + if !metadata.is_dir() { + break; + } + + let mut entries = fs::read_dir(path) + .map_err(|error| format!("读取旧官方资源父目录失败 {}：{error}", path.display()))?; + if entries.next().is_some() { + break; + } + fs::remove_dir(path) + .map_err(|error| format!("清理旧官方资源空目录失败 {}：{error}", path.display()))?; + current = path.parent(); + } + Ok(()) +} + fn unix_seconds_now() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -2711,6 +2935,95 @@ exit 22 assert!(manifest_state.has_any_resources()); } + #[test] + fn prune_stale_manifest_entries_removes_old_resource_and_partial_only() { + let out_dir = TempDir::new().unwrap(); + let service = OfficialResourcePullService::new(out_dir.path()); + let plan = build_official_pull_plan(discovery_plan(), inventory()); + let expected_url = plan.all_urls().unwrap().into_iter().next().unwrap(); + let expected_destination = service.destination_for_url(&expected_url).unwrap(); + fs::create_dir_all(expected_destination.parent().unwrap()).unwrap(); + fs::write(&expected_destination, b"keep").unwrap(); + + let stale_url = + "https://prod-clientpatch.bluearchiveyostar.com/r93_token/obsolete/old.bundle"; + let stale_destination = service.destination_for_url(stale_url).unwrap(); + fs::create_dir_all(stale_destination.parent().unwrap()).unwrap(); + fs::write(&stale_destination, b"remove").unwrap(); + fs::write(partial_path_for(&stale_destination), b"remove-partial").unwrap(); + + let mut manifest = OfficialDownloadManifest::default(); + service + .record_download_manifest_entry(&mut manifest, &expected_url, &expected_destination) + .unwrap(); + manifest.entries.insert( + stale_url.to_string(), + OfficialDownloadManifestEntry { + url: stale_url.to_string(), + destination: service.relative_destination(&stale_destination).unwrap(), + bytes: 6, + blake3: "stale".to_string(), + }, + ); + service.write_download_manifest(&manifest).unwrap(); + + let snapshot_path = out_dir.path().join("official-sync-snapshot.json"); + let quarantine_path = out_dir.path().join(DOWNLOAD_QUARANTINE_FILE); + fs::write(&snapshot_path, b"snapshot").unwrap(); + fs::write(&quarantine_path, b"quarantine").unwrap(); + + assert_eq!(service.prune_stale_manifest_entries(&plan).unwrap(), 1); + assert!(!stale_destination.exists()); + assert!(!partial_path_for(&stale_destination).exists()); + assert!(!stale_destination.parent().unwrap().exists()); + assert!(expected_destination.exists()); + assert_eq!(fs::read(&snapshot_path).unwrap(), b"snapshot"); + assert_eq!(fs::read(&quarantine_path).unwrap(), b"quarantine"); + + let pruned_manifest = service.read_download_manifest().unwrap(); + assert!(pruned_manifest.entries.contains_key(&expected_url)); + assert!(!pruned_manifest.entries.contains_key(stale_url)); + } + + #[cfg(unix)] + #[test] + fn prune_stale_manifest_entries_rejects_symlink_destination() { + use std::os::unix::fs::symlink; + + let out_dir = TempDir::new().unwrap(); + let service = OfficialResourcePullService::new(out_dir.path()); + let plan = build_official_pull_plan(discovery_plan(), inventory()); + let stale_url = + "https://prod-clientpatch.bluearchiveyostar.com/r93_token/obsolete/old.bundle"; + let stale_destination = service.destination_for_url(stale_url).unwrap(); + fs::create_dir_all(stale_destination.parent().unwrap()).unwrap(); + let outside = out_dir.path().join("outside-resource"); + fs::write(&outside, b"must-keep").unwrap(); + symlink(&outside, &stale_destination).unwrap(); + + let mut manifest = OfficialDownloadManifest::default(); + manifest.entries.insert( + stale_url.to_string(), + OfficialDownloadManifestEntry { + url: stale_url.to_string(), + destination: service.relative_destination(&stale_destination).unwrap(), + bytes: 9, + blake3: "stale".to_string(), + }, + ); + service.write_download_manifest(&manifest).unwrap(); + + let error = service.prune_stale_manifest_entries(&plan).unwrap_err(); + assert!(error.contains("symlink")); + assert!(stale_destination.exists()); + assert_eq!(fs::read(&outside).unwrap(), b"must-keep"); + assert!(service + .read_download_manifest() + .unwrap() + .entries + .contains_key(stale_url)); + } + #[test] fn redownloads_existing_file_when_manifest_hash_mismatches() { let out_dir = TempDir::new().unwrap(); @@ -2974,8 +3287,6 @@ exit 22 .unwrap(); assert_eq!(report.items.len(), expected_urls); - assert_eq!(events.len(), expected_urls * 2); - let started: Vec<_> = events .iter() .filter(|event| event.kind == OfficialResourcePullProgressKind::Started) @@ -2984,8 +3295,13 @@ exit 22 .iter() .filter(|event| event.kind == OfficialResourcePullProgressKind::Finished) .collect(); + let verifications: Vec<_> = events + .iter() + .filter(|event| event.kind == OfficialResourcePullProgressKind::Verification) + .collect(); assert_eq!(started.len(), expected_urls); assert_eq!(finished.len(), expected_urls); + assert_eq!(verifications.len(), report.verified_hashes.len()); // 每个 started 的 total 一致；index 表示已完成数量，不能超过总数。 assert!(started.iter().all(|event| event.index <= expected_urls)); @@ -3001,6 +3317,18 @@ exit 22 assert!(finished .iter() .all(|event| event.status == Some(OfficialResourcePullStatus::Downloaded))); + assert!(finished.iter().all(|event| { + event.verification.as_ref().is_some_and(|verification| { + verification.actual_bytes > 0 && !verification.actual_blake3.is_empty() + }) + })); + assert!(verifications.iter().all(|event| { + event.official_hash.as_ref().is_some_and(|hash| { + hash.algorithm == OfficialResourceHashAlgorithm::XxHash32Decimal + && hash.expected == hash.actual + && !hash.hash_url.is_empty() + }) + })); assert!(events .iter() .any(|event| event.url.ends_with("/TableBundles/ExcelDB.db"))); diff --git a/infrastructure/src/official_update.rs b/infrastructure/src/official_update.rs index de42461..b5435d2 100644 --- a/infrastructure/src/official_update.rs +++ b/infrastructure/src/official_update.rs @@ -32,9 +32,10 @@ use crate::{ build_official_pull_plan_for_platform_inventory, build_official_sync_plan, changed_endpoint_urls, default_official_platforms, DownloadError, OfficialGameMainConfigBootstrapService, OfficialLauncherBootstrapService, - OfficialResourcePullPlan, OfficialResourcePullProgress, OfficialResourcePullProgressKind, - OfficialResourcePullService, YostarJpLauncherCdnConfig, YostarJpLauncherGameConfig, - YostarJpLauncherManifestUrl, YostarJpLauncherRemoteManifest, + OfficialResourceHashVerification, OfficialResourcePullPlan, OfficialResourcePullProgress, + OfficialResourcePullProgressKind, OfficialResourcePullService, OfficialResourceVerification, + YostarJpLauncherCdnConfig, YostarJpLauncherGameConfig, YostarJpLauncherManifestUrl, + YostarJpLauncherRemoteManifest, }; use crate::{ read_parse_cache_at, OfficialParseCacheService, OfficialParseConfig, OfficialParseSummary, @@ -865,6 +866,10 @@ pub struct OfficialUpdateProgress { pub download_failure_attempts: Option, /// Whether the URL was recorded in the quarantine manifest. pub download_quarantined: Option, + /// Local size/BLAKE3/ZIP verification for a completed URL. + pub download_verification: Option, + /// Official `.hash` sidecar verification when a seed catalog pair passes. + pub official_hash_verification: Option, } impl OfficialUpdateProgress { @@ -884,6 +889,8 @@ impl OfficialUpdateProgress { download_failure_retryable: None, download_failure_attempts: None, download_quarantined: None, + download_verification: None, + official_hash_verification: None, } } @@ -906,6 +913,8 @@ impl OfficialUpdateProgress { self.download_failure_attempts = event.failure_attempts; self.download_quarantined = (event.kind == OfficialResourcePullProgressKind::Failed).then_some(event.quarantined); + self.download_verification = event.verification.clone(); + self.official_hash_verification = event.official_hash.clone(); self } } @@ -1808,6 +1817,18 @@ impl OfficialUpdateService { &config.curl_command, ) .with_proxy_config(config.curl_proxy.clone()); + let pruned_stale_resource_count = staging_fetcher + .prune_stale_manifest_entries(&pull_plan) + .map_err(anyhow::Error::msg)?; + if pruned_stale_resource_count > 0 { + progress(OfficialUpdateProgress::new( + "publish", + format!( + "已清理新 manifest 删除的旧官方资源：{} 项", + pruned_stale_resource_count + ), + )); + } report.staging_path = Some(publish_plan.staging_path.clone()); report.snapshot_path = staging_snapshot_path.clone(); report.download_manifest = staging_fetcher.download_manifest_path(); @@ -2701,6 +2722,21 @@ fn progress_from_pull_event(event: OfficialResourcePullProgress) -> OfficialUpda ), ) .with_download_progress(&event), + OfficialResourcePullProgressKind::Verification => { + let hash = event.official_hash.as_ref(); + OfficialUpdateProgress::new( + "verify", + format!( + "官方 hash 校验通过：算法={} 期望={} 实际={} 数据 URL={} hash URL={}", + hash.map(|value| value.algorithm.as_str()).unwrap_or("unknown"), + hash.map(|value| value.expected.as_str()).unwrap_or("unknown"), + hash.map(|value| value.actual.as_str()).unwrap_or("unknown"), + hash.map(|value| value.data_url.as_str()).unwrap_or(event.url.as_str()), + hash.map(|value| value.hash_url.as_str()).unwrap_or("unknown") + ), + ) + .with_download_progress(&event) + } } }