From 2bdc55a1bec5edf7132f16d43334d20cee1246d0 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Fri, 10 Jul 2026 00:36:40 +0800 Subject: [PATCH] feat: support cancellable official sync --- Cargo.lock | 1 + infrastructure/Cargo.toml | 1 + infrastructure/src/official_download.rs | 19 +++ infrastructure/src/official_update.rs | 187 +++++++++++++++++++++--- 4 files changed, 188 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ee192a9..25996f7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -205,6 +205,7 @@ dependencies = [ "bat-core", "blake3", "hex", + "libc", "md-5", "serde", "serde_json", diff --git a/infrastructure/Cargo.toml b/infrastructure/Cargo.toml index cf41be3..31f55de 100644 --- a/infrastructure/Cargo.toml +++ b/infrastructure/Cargo.toml @@ -23,6 +23,7 @@ async-trait.workspace = true md-5 = "0.10" hex = "0.4" tempfile = "3.14" +libc = "0.2" sqlx = { version = "0.8", features = ["runtime-tokio", "sqlite"] } [dev-dependencies] diff --git a/infrastructure/src/official_download.rs b/infrastructure/src/official_download.rs index 0bd5994..ed1a188 100644 --- a/infrastructure/src/official_download.rs +++ b/infrastructure/src/official_download.rs @@ -332,6 +332,17 @@ impl OfficialResourcePullService { &self, plan: &OfficialResourcePullPlan, mut progress: impl FnMut(OfficialResourcePullProgress), + ) -> Result { + self.pull_with_progress_and_cancellation(plan, &mut progress, || false) + } + + /// Executes the plan, emits URL progress, and aborts between URLs when + /// `should_cancel` returns true. + pub fn pull_with_progress_and_cancellation( + &self, + plan: &OfficialResourcePullPlan, + mut progress: impl FnMut(OfficialResourcePullProgress), + mut should_cancel: impl FnMut() -> bool, ) -> Result { fs::create_dir_all(&self.output_root).map_err(|error| { format!( @@ -347,6 +358,10 @@ impl OfficialResourcePullService { let total = urls.len(); let mut items = Vec::new(); for (offset, url) in urls.into_iter().enumerate() { + if should_cancel() { + return Err("official resource pull interrupted by shutdown request".to_string()); + } + let index = offset + 1; progress(OfficialResourcePullProgress::started( index, @@ -400,6 +415,10 @@ impl OfficialResourcePullService { }); } + if should_cancel() { + return Err("official resource pull interrupted by shutdown request".to_string()); + } + let verified_hashes = self.verify_downloaded_official_hashes(&official_hash_pairs)?; Ok(OfficialResourcePullReport { diff --git a/infrastructure/src/official_update.rs b/infrastructure/src/official_update.rs index 3e4223a..afdfa11 100644 --- a/infrastructure/src/official_update.rs +++ b/infrastructure/src/official_update.rs @@ -23,6 +23,7 @@ use bat_adapters::official::yostar_jp::{ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::fs::{self, OpenOptions}; +use std::io::Write; use std::path::{Path, PathBuf}; /// Current official update snapshot schema version. @@ -426,6 +427,17 @@ impl OfficialUpdateService { &self, config: &OfficialUpdateConfig, mut progress: impl FnMut(OfficialUpdateProgress), + ) -> anyhow::Result { + self.run_with_progress_and_cancellation(config, &mut progress, || false) + } + + /// Executes one official update run, emits progress events, and aborts + /// between safe steps when `should_cancel` returns true. + pub fn run_with_progress_and_cancellation( + &self, + config: &OfficialUpdateConfig, + mut progress: impl FnMut(OfficialUpdateProgress), + mut should_cancel: impl FnMut() -> bool, ) -> anyhow::Result { progress(OfficialUpdateProgress::new( "start", @@ -436,6 +448,7 @@ impl OfficialUpdateService { config.auto_discover ), )); + check_shutdown_requested(&mut should_cancel)?; let _lock = if config.dry_run { progress(OfficialUpdateProgress::new( @@ -452,6 +465,7 @@ impl OfficialUpdateService { progress(OfficialUpdateProgress::new("lock", "state lock acquired")); Some(lock) }; + check_shutdown_requested(&mut should_cancel)?; let default_platforms = default_official_platforms(); let platforms = config @@ -481,6 +495,7 @@ impl OfficialUpdateService { &bootstrap_cache_path, !config.dry_run, &mut progress, + &mut should_cancel, )?) } else { progress(OfficialUpdateProgress::new( @@ -489,6 +504,7 @@ impl OfficialUpdateService { )); None }; + check_shutdown_requested(&mut should_cancel)?; let app_version = config .app_version @@ -557,6 +573,7 @@ impl OfficialUpdateService { "server-info", format!("parsing server-info bytes={}", server_info_bytes.len()), )); + check_shutdown_requested(&mut should_cancel)?; let server_info = YostarJpServerInfo::from_slice(&server_info_bytes)?; let current_snapshot = server_info .sync_snapshot(&connection_group, &app_version, platforms) @@ -576,13 +593,18 @@ impl OfficialUpdateService { marker_endpoint_count(¤t_snapshot) ), )); - let endpoint_markers = - collect_endpoint_markers(&fetcher, ¤t_snapshot, &mut progress)?; + let endpoint_markers = collect_endpoint_markers( + &fetcher, + ¤t_snapshot, + &mut progress, + &mut should_cancel, + )?; let current_update_snapshot = OfficialUpdateSnapshot::new( current_snapshot.clone(), endpoint_markers, bootstrap.as_ref(), ); + check_shutdown_requested(&mut should_cancel)?; progress(OfficialUpdateProgress::new( "snapshot", format!("reading previous snapshot {}", snapshot_path.display()), @@ -617,6 +639,7 @@ impl OfficialUpdateService { platforms, &fetcher, &mut progress, + &mut should_cancel, )?) } else { progress(OfficialUpdateProgress::new( @@ -684,6 +707,7 @@ impl OfficialUpdateService { should_download, repair_needed ), )); + check_shutdown_requested(&mut should_cancel)?; let mut report = OfficialUpdateReport { update_status: if should_download { OfficialUpdateStatus::WouldDownload @@ -765,6 +789,7 @@ impl OfficialUpdateService { )); return Ok(report); } + check_shutdown_requested(&mut should_cancel)?; let pull_plan = pull_plan .as_ref() @@ -775,9 +800,13 @@ impl OfficialUpdateService { format!("downloading or reusing {download_url_count} official URLs"), )); let pull_report = fetcher - .pull_with_progress(pull_plan, |event| { - progress(progress_from_pull_event(event)); - }) + .pull_with_progress_and_cancellation( + pull_plan, + |event| { + progress(progress_from_pull_event(event)); + }, + &mut should_cancel, + ) .map_err(anyhow::Error::msg)?; progress(OfficialUpdateProgress::new( "download", @@ -793,6 +822,7 @@ impl OfficialUpdateService { "audit", "running final local manifest audit", )); + check_shutdown_requested(&mut should_cancel)?; let final_audit = fetcher .audit_local_manifest(pull_plan) .map_err(anyhow::Error::msg)?; @@ -833,7 +863,9 @@ fn build_pull_plan( platforms: &[PatchPlatform], fetcher: &OfficialResourcePullService, progress: &mut dyn FnMut(OfficialUpdateProgress), + should_cancel: &mut dyn FnMut() -> bool, ) -> anyhow::Result { + check_shutdown_requested(should_cancel)?; let discovery = server_info .discovery_plan(connection_group, app_version, platforms) .map_err(anyhow::Error::msg)?; @@ -846,7 +878,9 @@ fn build_pull_plan( discovery.endpoints.len() ), )); - let seed_catalogs = fetch_seed_catalogs(fetcher, &discovery, progress)?; + check_shutdown_requested(should_cancel)?; + let seed_catalogs = fetch_seed_catalogs(fetcher, &discovery, progress, should_cancel)?; + check_shutdown_requested(should_cancel)?; progress(OfficialUpdateProgress::new( "inventory", "parsing seed catalogs into download inventory", @@ -877,6 +911,7 @@ fn collect_endpoint_markers( fetcher: &OfficialResourcePullService, snapshot: &YostarJpSyncSnapshot, progress: &mut dyn FnMut(OfficialUpdateProgress), + should_cancel: &mut dyn FnMut() -> bool, ) -> anyhow::Result> { let mut markers = Vec::new(); @@ -884,6 +919,7 @@ fn collect_endpoint_markers( let Some(role) = marker_role(endpoint.kind) else { continue; }; + check_shutdown_requested(should_cancel)?; progress(OfficialUpdateProgress::new( "marker", format!( @@ -906,9 +942,20 @@ fn collect_endpoint_markers( }); } + check_shutdown_requested(should_cancel)?; Ok(markers) } +fn check_shutdown_requested(should_cancel: &mut dyn FnMut() -> bool) -> anyhow::Result<()> { + if should_cancel() { + return Err(anyhow::anyhow!( + "official update interrupted by shutdown request" + )); + } + + Ok(()) +} + fn marker_role(kind: YostarJpResourceEndpointKind) -> Option { match kind { YostarJpResourceEndpointKind::TableCatalogHash @@ -1124,7 +1171,9 @@ fn resolve_bootstrap( cache_path: &Path, write_cache: bool, progress: &mut dyn FnMut(OfficialUpdateProgress), + should_cancel: &mut dyn FnMut() -> bool, ) -> anyhow::Result { + check_shutdown_requested(should_cancel)?; let launcher = OfficialLauncherBootstrapService::with_curl_command( launcher_version, curl_command.to_string_lossy().to_string(), @@ -1136,6 +1185,7 @@ fn resolve_bootstrap( let (game_config, manifest_url, manifest) = launcher .fetch_latest_remote_manifest() .map_err(anyhow::Error::msg)?; + check_shutdown_requested(should_cancel)?; let launcher_metadata = launcher_metadata_from_parts(launcher_version, &game_config, &manifest_url, &manifest); progress(OfficialUpdateProgress::new( @@ -1170,12 +1220,14 @@ fn resolve_bootstrap( "game-main-config", "cache miss; fetching resources.assets and parsing GameMainConfig", )); + check_shutdown_requested(should_cancel)?; let bootstrapper = OfficialGameMainConfigBootstrapService::with_commands( launcher_version, curl_command.to_path_buf(), unzip_command.to_path_buf(), ); let bootstrap = bootstrapper.fetch_bootstrap().map_err(anyhow::Error::msg)?; + check_shutdown_requested(should_cancel)?; let launcher_metadata = LauncherMetadataSnapshot { launcher_version: launcher_version.to_string(), game_latest_version: bootstrap.game_config.game_latest_version.clone(), @@ -1219,6 +1271,7 @@ fn fetch_seed_catalogs( fetcher: &OfficialResourcePullService, discovery: &YostarJpResourceDiscoveryPlan, progress: &mut dyn FnMut(OfficialUpdateProgress), + should_cancel: &mut dyn FnMut() -> bool, ) -> anyhow::Result { let mut catalogs = SeedCatalogs::default(); @@ -1232,6 +1285,7 @@ fn fetch_seed_catalogs( continue; } + check_shutdown_requested(should_cancel)?; progress(OfficialUpdateProgress::new( "catalog", format!( @@ -1247,6 +1301,7 @@ fn fetch_seed_catalogs( catalogs.insert(endpoint, bytes)?; } + check_shutdown_requested(should_cancel)?; Ok(catalogs) } @@ -1338,22 +1393,35 @@ impl OfficialUpdateLock { fn acquire(config: &OfficialUpdateConfig) -> anyhow::Result { fs::create_dir_all(&config.output_root)?; let path = config.lock_path(); - match OpenOptions::new().write(true).create_new(true).open(&path) { - Ok(file) => { - let pid = std::process::id().to_string(); - use std::io::Write; - let mut file = file; - file.write_all(pid.as_bytes())?; - Ok(Self { path }) + for attempt in 0..=1 { + match OpenOptions::new().write(true).create_new(true).open(&path) { + Ok(mut file) => { + let pid = std::process::id().to_string(); + file.write_all(pid.as_bytes())?; + return Ok(Self { path }); + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + if attempt == 0 && remove_stale_lock(&path)? { + continue; + } + return Err(anyhow::anyhow!( + "official update output is locked: {}", + path.display() + )); + } + Err(error) => { + return Err(anyhow::anyhow!( + "failed to acquire official update lock {}: {error}", + path.display() + )); + } } - Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => Err( - anyhow::anyhow!("official update output is locked: {}", path.display()), - ), - Err(error) => Err(anyhow::anyhow!( - "failed to acquire official update lock {}: {error}", - path.display() - )), } + + Err(anyhow::anyhow!( + "failed to acquire official update lock {}", + path.display() + )) } } @@ -1363,6 +1431,44 @@ impl Drop for OfficialUpdateLock { } } +fn remove_stale_lock(path: &Path) -> anyhow::Result { + let contents = fs::read_to_string(path).unwrap_or_default(); + let Some(pid) = parse_lock_pid(&contents) else { + return Ok(false); + }; + if process_exists(pid) { + return Ok(false); + } + + fs::remove_file(path)?; + Ok(true) +} + +fn parse_lock_pid(contents: &str) -> Option { + contents.trim().parse::().ok().filter(|pid| *pid > 0) +} + +#[cfg(unix)] +fn process_exists(pid: u32) -> bool { + let Ok(pid) = libc::pid_t::try_from(pid) else { + return false; + }; + + if pid <= 0 { + return false; + } + + unsafe { + libc::kill(pid, 0) == 0 + || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + } +} + +#[cfg(not(unix))] +fn process_exists(_pid: u32) -> bool { + true +} + #[cfg(test)] mod tests { use super::*; @@ -1539,4 +1645,45 @@ mod tests { drop(first); assert!(OfficialUpdateLock::acquire(&config).is_ok()); } + + #[test] + fn non_dry_run_lock_removes_stale_pid_file() { + let temp = tempfile::TempDir::new().unwrap(); + let config = OfficialUpdateConfig { + output_root: temp.path().to_path_buf(), + ..OfficialUpdateConfig::default() + }; + fs::create_dir_all(&config.output_root).unwrap(); + fs::write(config.lock_path(), "999999999").unwrap(); + + let lock = OfficialUpdateLock::acquire(&config).unwrap(); + assert!(config.lock_path().exists()); + drop(lock); + assert!(!config.lock_path().exists()); + } + + #[test] + fn update_run_can_be_cancelled_before_network_work() { + let temp = tempfile::TempDir::new().unwrap(); + let config = OfficialUpdateConfig { + output_root: temp.path().to_path_buf(), + dry_run: true, + ..OfficialUpdateConfig::default() + }; + + let mut checks = 0; + let error = OfficialUpdateService::new() + .run_with_progress_and_cancellation( + &config, + |_| {}, + || { + checks += 1; + true + }, + ) + .unwrap_err(); + + assert!(error.to_string().contains("shutdown request")); + assert_eq!(checks, 1); + } }