feat: support cancellable official sync

Cargo.lock

@@ -205,6 +205,7 @@ dependencies = [
  "bat-core",
  "blake3",
  "hex",
+ "libc",
  "md-5",
  "serde",
  "serde_json",

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]

infrastructure/src/official_download.rs

@@ -332,6 +332,17 @@ impl OfficialResourcePullService {
         &self,
         plan: &OfficialResourcePullPlan,
         mut progress: impl FnMut(OfficialResourcePullProgress),
+    ) -> Result<OfficialResourcePullReport, String> {
+        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<OfficialResourcePullReport, String> {
         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 {

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<OfficialUpdateReport> {
+        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<OfficialUpdateReport> {
         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(&current_snapshot)
             ),
         ));
-        let endpoint_markers =
-            collect_endpoint_markers(&fetcher, &current_snapshot, &mut progress)?;
+        let endpoint_markers = collect_endpoint_markers(
+            &fetcher,
+            &current_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<OfficialResourcePullPlan> {
+    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<Vec<OfficialEndpointMarkerSnapshot>> {
     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<OfficialEndpointMarkerRole> {
     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<ResolvedBootstrap> {
+    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<SeedCatalogs> {
     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<Self> {
         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<bool> {
+    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<u32> {
+    contents.trim().parse::<u32>().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);
+    }
 }