fix(official-sync): 复用失败 staging 避免重复全量下载

infrastructure/src/official_update.rs

@@ -632,6 +632,7 @@ struct OfficialPublishPlan {
     id: String,
     staging_path: PathBuf,
     version_path: PathBuf,
+    reuse_existing_staging: bool,
 }
 
 impl OfficialPublishLayout {
@@ -694,12 +695,30 @@ impl OfficialPublishLayout {
         Ok(self.current_target()?.is_some())
     }
 
-    fn plan(&self, snapshot: &OfficialUpdateSnapshot) -> OfficialPublishPlan {
+    fn plan(
+        &self,
+        snapshot: &OfficialUpdateSnapshot,
+        recovered_staging: Option<&OfficialVersionRecord>,
+    ) -> OfficialPublishPlan {
+        if let Some(record) = recovered_staging {
+            let staging_path = record
+                .staging_path
+                .clone()
+                .unwrap_or_else(|| record.resource_root.clone());
+            return OfficialPublishPlan {
+                id: record.id.clone(),
+                staging_path,
+                version_path: self.versions_dir.join(&record.id),
+                reuse_existing_staging: true,
+            };
+        }
+
         let id = publish_id(snapshot);
         OfficialPublishPlan {
             staging_path: self.staging_dir.join(&id),
             version_path: self.versions_dir.join(&id),
             id,
+            reuse_existing_staging: false,
         }
     }
 
@@ -716,12 +735,14 @@ impl OfficialPublishLayout {
 
         if path_exists_no_follow(&plan.staging_path)? {
             ensure_safe_directory_path(&plan.staging_path, "官方资源 staging 目录")?;
-            fs::remove_dir_all(&plan.staging_path).map_err(|error| {
-                format!(
-                    "清理官方资源 staging 目录失败 {}：{error}",
-                    plan.staging_path.display()
-                )
-            })?;
+            if !plan.reuse_existing_staging {
+                fs::remove_dir_all(&plan.staging_path).map_err(|error| {
+                    format!(
+                        "清理官方资源 staging 目录失败 {}：{error}",
+                        plan.staging_path.display()
+                    )
+                })?;
+            }
         }
 
         if path_exists_no_follow(&plan.version_path)? {
@@ -731,6 +752,10 @@ impl OfficialPublishLayout {
             ));
         }
 
+        if plan.reuse_existing_staging {
+            return Ok(());
+        }
+
         fs::create_dir_all(&plan.staging_path).map_err(|error| {
             format!(
                 "创建官方资源 staging 目录失败 {}：{error}",
@@ -1236,7 +1261,28 @@ impl OfficialUpdateService {
         check_shutdown_requested(&mut should_cancel)?;
 
         let download_url_count = pull_plan.all_urls().map_err(anyhow::Error::msg)?.len();
-        let publish_plan = publish_layout.plan(&current_update_snapshot);
+        let recovered_staging = recoverable_failed_staging(
+            &version_state_path,
+            &current_update_snapshot,
+            &publish_layout,
+        )
+        .map_err(anyhow::Error::msg)?;
+        if let Some(record) = recovered_staging.as_ref() {
+            progress(OfficialUpdateProgress::new(
+                "publish",
+                format!(
+                    "发现可复用的未完成 staging：id={} path={}",
+                    record.id,
+                    record
+                        .staging_path
+                        .as_ref()
+                        .unwrap_or(&record.resource_root)
+                        .display()
+                ),
+            ));
+        }
+        let publish_plan =
+            publish_layout.plan(&current_update_snapshot, recovered_staging.as_ref());
         progress(OfficialUpdateProgress::new(
             "publish",
             format!(
@@ -1248,9 +1294,16 @@ impl OfficialUpdateService {
         publish_layout
             .prepare_staging(&publish_plan)
             .map_err(anyhow::Error::msg)?;
-        publish_layout
-            .seed_staging_from_active(&active_resource_root, &publish_plan.staging_path)
-            .map_err(anyhow::Error::msg)?;
+        if publish_plan.reuse_existing_staging {
+            progress(OfficialUpdateProgress::new(
+                "publish",
+                "复用未完成 staging；跳过 active release seed，避免覆盖已下载文件",
+            ));
+        } else {
+            publish_layout
+                .seed_staging_from_active(&active_resource_root, &publish_plan.staging_path)
+                .map_err(anyhow::Error::msg)?;
+        }
         let staging_snapshot_path = snapshot_path_for(config, &publish_plan.staging_path);
         let in_progress_record = prepare_in_progress_version_state(
             &version_state_path,
@@ -1989,6 +2042,86 @@ fn recover_interrupted_version_state(path: &Path) -> anyhow::Result<()> {
     write_version_state(path, &state)
 }
 
+fn recoverable_failed_staging(
+    version_state_path: &Path,
+    snapshot: &OfficialUpdateSnapshot,
+    layout: &OfficialPublishLayout,
+) -> Result<Option<OfficialVersionRecord>, String> {
+    let Some(state) = read_version_state(version_state_path).map_err(|error| error.to_string())?
+    else {
+        return Ok(None);
+    };
+
+    for failed in state.failed_versions.iter().rev() {
+        let record = &failed.version;
+        if !version_record_matches_snapshot(record, snapshot) {
+            continue;
+        }
+
+        let Some(staging_path) = recoverable_staging_path(record, layout)? else {
+            continue;
+        };
+        let mut recovered = record.clone();
+        recovered.resource_root = staging_path.clone();
+        recovered.staging_path = Some(staging_path);
+        recovered.version_path = None;
+        recovered.completed_unix_seconds = None;
+        return Ok(Some(recovered));
+    }
+
+    Ok(None)
+}
+
+fn recoverable_staging_path(
+    record: &OfficialVersionRecord,
+    layout: &OfficialPublishLayout,
+) -> Result<Option<PathBuf>, String> {
+    let staging_path = record
+        .staging_path
+        .as_ref()
+        .unwrap_or(&record.resource_root);
+    ensure_path_within_root(&layout.staging_dir, staging_path)?;
+    if version_id_from_path(staging_path).as_deref() != Some(record.id.as_str()) {
+        return Err(format!(
+            "官方版本状态中的 staging 路径和版本 ID 不匹配：id={} path={}",
+            record.id,
+            staging_path.display()
+        ));
+    }
+
+    let metadata = match fs::symlink_metadata(staging_path) {
+        Ok(metadata) => metadata,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(error) => {
+            return Err(format!(
+                "读取可恢复 staging 元数据失败 {}：{error}",
+                staging_path.display()
+            ))
+        }
+    };
+    if metadata.file_type().is_symlink() {
+        return Err(format!(
+            "可恢复 staging 不能是 symlink：{}",
+            staging_path.display()
+        ));
+    }
+    if !metadata.is_dir() {
+        return Err(format!(
+            "可恢复 staging 已存在但不是目录：{}",
+            staging_path.display()
+        ));
+    }
+    ensure_safe_directory_path(staging_path, "可恢复 staging 目录")?;
+
+    let version_path = layout.versions_dir.join(&record.id);
+    ensure_path_within_root(&layout.versions_dir, &version_path)?;
+    if path_exists_no_follow(&version_path)? {
+        return Ok(None);
+    }
+
+    Ok(Some(staging_path.to_path_buf()))
+}
+
 fn prepare_in_progress_version_state(
     path: &Path,
     snapshot: &OfficialUpdateSnapshot,
@@ -2038,6 +2171,15 @@ fn prepare_in_progress_version_state(
     Ok(record)
 }
 
+fn version_record_matches_snapshot(
+    record: &OfficialVersionRecord,
+    snapshot: &OfficialUpdateSnapshot,
+) -> bool {
+    record.app_version == snapshot.app_version
+        && record.bundle_version == snapshot.bundle_version
+        && record.addressables_root == snapshot.addressables_root
+}
+
 fn complete_version_state(
     path: &Path,
     record: OfficialVersionRecord,
@@ -2888,6 +3030,111 @@ mod tests {
         assert!(state.failed_versions.is_empty());
     }
 
+    #[test]
+    fn recoverable_failed_staging_preserves_matching_staging() {
+        let temp = tempfile::TempDir::new().unwrap();
+        let layout = OfficialPublishLayout::new(temp.path());
+        let snapshot = OfficialUpdateSnapshot::new(
+            fixture_base_snapshot(),
+            vec![fixture_marker("current")],
+            Some(&fixture_bootstrap()),
+        );
+        let staging = temp.path().join(".staging/reuse-id");
+        fs::create_dir_all(&staging).unwrap();
+        fs::write(staging.join("kept.bin"), b"already-downloaded").unwrap();
+        let record = version_record_for_snapshot(
+            &snapshot,
+            VersionRecordInput {
+                id: "reuse-id".to_string(),
+                resource_root: staging.clone(),
+                snapshot_path: staging.join(OFFICIAL_SYNC_SNAPSHOT_FILE),
+                staging_path: Some(staging.clone()),
+                version_path: None,
+                started_unix_seconds: Some(10),
+                completed_unix_seconds: None,
+            },
+        );
+        let state = OfficialVersionState {
+            failed_versions: vec![OfficialFailedVersionRecord {
+                version: record,
+                error: "download interrupted".to_string(),
+                failed_unix_seconds: 11,
+            }],
+            updated_unix_seconds: 11,
+            ..OfficialVersionState::default()
+        };
+        let state_path = temp.path().join(OFFICIAL_VERSION_STATE_FILE);
+        write_version_state(&state_path, &state).unwrap();
+
+        let recovered = recoverable_failed_staging(&state_path, &snapshot, &layout)
+            .unwrap()
+            .unwrap();
+        let plan = layout.plan(&snapshot, Some(&recovered));
+        assert!(plan.reuse_existing_staging);
+
+        layout.prepare_staging(&plan).unwrap();
+
+        assert_eq!(
+            fs::read(staging.join("kept.bin")).unwrap(),
+            b"already-downloaded"
+        );
+        assert_eq!(plan.staging_path, staging);
+        assert_eq!(plan.version_path, temp.path().join("versions/reuse-id"));
+    }
+
+    #[test]
+    fn recoverable_failed_staging_ignores_mismatched_snapshot() {
+        let temp = tempfile::TempDir::new().unwrap();
+        let layout = OfficialPublishLayout::new(temp.path());
+        let previous_snapshot = OfficialUpdateSnapshot::new(
+            fixture_base_snapshot(),
+            vec![fixture_marker("previous")],
+            Some(&fixture_bootstrap()),
+        );
+        let mut current_base = fixture_base_snapshot();
+        current_base.addressables_root =
+            "https://prod-clientpatch.bluearchiveyostar.com/other".to_string();
+        let current_snapshot = OfficialUpdateSnapshot::new(
+            current_base,
+            vec![fixture_marker("current")],
+            Some(&fixture_bootstrap()),
+        );
+        let staging = temp.path().join(".staging/old-id");
+        fs::create_dir_all(&staging).unwrap();
+        let record = version_record_for_snapshot(
+            &previous_snapshot,
+            VersionRecordInput {
+                id: "old-id".to_string(),
+                resource_root: staging,
+                snapshot_path: temp
+                    .path()
+                    .join(".staging/old-id")
+                    .join(OFFICIAL_SYNC_SNAPSHOT_FILE),
+                staging_path: Some(temp.path().join(".staging/old-id")),
+                version_path: None,
+                started_unix_seconds: Some(10),
+                completed_unix_seconds: None,
+            },
+        );
+        let state = OfficialVersionState {
+            failed_versions: vec![OfficialFailedVersionRecord {
+                version: record,
+                error: "download interrupted".to_string(),
+                failed_unix_seconds: 11,
+            }],
+            updated_unix_seconds: 11,
+            ..OfficialVersionState::default()
+        };
+        let state_path = temp.path().join(OFFICIAL_VERSION_STATE_FILE);
+        write_version_state(&state_path, &state).unwrap();
+
+        assert!(
+            recoverable_failed_staging(&state_path, &current_snapshot, &layout)
+                .unwrap()
+                .is_none()
+        );
+    }
+
     #[test]
     fn update_rejects_dangerous_output_root_before_network_work() {
         let config = OfficialUpdateConfig {