fix(sync): 立即校验官方 hash sidecar

infrastructure/src/official_download.rs

@@ -695,20 +695,15 @@ impl OfficialResourcePullService {
             });
         }
 
-        // Phase B：顺序下载 need-download 项。每个 URL 的目标和 `.part` 都是独立
+        // Phase B：按 plan 顺序处理每个 URL。下载或复用完成并写入 manifest 后，
-        // 的，但这里保留单线程执行，便于维持稳定进度、稳定日志和简单的失败恢复。
+        // 立即尝试校验已经到齐的官方 `.bytes/.hash` pair，避免把文件级问题延后到整轮末尾。
-        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();
         let mut completed_count = 0usize;
+        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 in &download_indices {
+        for item in &planned {
-            let item = &planned[plan_index];
             progress(OfficialResourcePullProgress::started(
                 completed_count,
                 total,
@@ -718,37 +713,62 @@ impl OfficialResourcePullService {
                 return Err("官方资源拉取已被停止请求中断".to_string().into());
             }
 
-            let result = self.pull_one(&item.url, &item.destination);
+            let result = if let Some(existing) = &item.existing {
-            match result {
+                self.clear_quarantine_entry(&item.url)?;
-                Ok(mut pull_result) => {
+                existing.clone()
-                    let verification_result = self
+            } else {
-                        .clear_quarantine_entry(&item.url)
+                match self.pull_one(&item.url, &item.destination) {
-                        .and_then(|_| {
+                    Ok(mut pull_result) => {
-                            self.record_download_manifest_entry(
+                        let verification_result = self
-                                &mut manifest,
+                            .clear_quarantine_entry(&item.url)
-                                &item.url,
+                            .and_then(|_| {
-                                &item.destination,
+                                self.record_download_manifest_entry(
-                            )
+                                    &mut manifest,
-                        })
+                                    &item.url,
-                        .and_then(|verification| {
+                                    &item.destination,
-                            self.write_download_manifest(&manifest)
+                                )
-                                .map(|_| verification)
+                            })
-                        });
+                            .and_then(|verification| {
-                    match verification_result {
+                                self.write_download_manifest(&manifest)
-                        Ok(verification) => {
+                                    .map(|_| verification)
-                            pull_result.verification = 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={}；{}",
+                                        item.url,
+                                        self.download_quarantine_path().display(),
+                                        error.message
+                                    ),
+                                ));
+                            }
                         }
-                        Err(error) => {
+
-                            let error =
+                        pull_result
-                                PullOneError::plain(format!("记录下载 manifest 失败：{error}"));
+                    }
-                            self.record_quarantine_entry(&item.url, &item.destination, &error)?;
+                    Err(error) => {
-                            progress(OfficialResourcePullProgress::failed(
+                        self.record_quarantine_entry(&item.url, &item.destination, &error)?;
-                                completed_count,
+                        progress(OfficialResourcePullProgress::failed(
-                                total,
+                            completed_count,
-                                item.url.clone(),
+                            total,
-                                &error,
+                            item.url.clone(),
-                            ));
+                            &error,
-                            return Err(DownloadError::new(
+                        ));
+                        return Err(DownloadError::new(
                             error.error_code(),
                             format!(
                                 "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
@@ -757,84 +777,20 @@ impl OfficialResourcePullService {
                                 error.message
                             ),
                         ));
-                        }
-                    }
-
-                    completed_count += 1;
-                    progress(OfficialResourcePullProgress::finished(
-                        completed_count,
-                        total,
-                        item.url.clone(),
-                        pull_result.status,
-                        pull_result.bytes,
-                        pull_result.transferred_bytes,
-                        pull_result.verification.clone(),
-                    ));
-                    download_results[plan_index] = Some(Ok(pull_result));
-                }
-                Err(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={}；{}",
-                            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(
-                    completed_count,
-                    total,
-                    item.url.clone(),
-                ));
-                completed_count += 1;
-                progress(OfficialResourcePullProgress::finished(
-                    completed_count,
-                    total,
-                    item.url.clone(),
-                    existing.status,
-                    existing.bytes,
-                    existing.transferred_bytes,
-                    existing.verification.clone(),
-                ));
-                existing.clone()
-            } 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),
-                        ))
                     }
                 }
             };
 
+            completed_count += 1;
+            progress(OfficialResourcePullProgress::finished(
+                completed_count,
+                total,
+                item.url.clone(),
+                result.status,
+                result.bytes,
+                result.transferred_bytes,
+                result.verification.clone(),
+            ));
             processed_urls.insert(item.url.clone());
             let newly_verified_hashes = self.verify_ready_official_hashes(
                 &official_hash_pairs,
@@ -3302,6 +3258,28 @@ exit 22
         assert_eq!(started.len(), expected_urls);
         assert_eq!(finished.len(), expected_urls);
         assert_eq!(verifications.len(), report.verified_hashes.len());
+        for verification_event in &verifications {
+            let hash = verification_event
+                .official_hash
+                .as_ref()
+                .expect("verification event must carry official hash detail");
+            let hash_finished_index = events
+                .iter()
+                .position(|event| {
+                    event.kind == OfficialResourcePullProgressKind::Finished
+                        && event.url == hash.hash_url
+                })
+                .expect("hash sidecar must finish before verification");
+            let verification_index = events
+                .iter()
+                .position(|event| std::ptr::eq(event, *verification_event))
+                .expect("verification event must be present in event stream");
+            assert_eq!(
+                verification_index,
+                hash_finished_index + 1,
+                "official hash verification must run immediately after sidecar is complete"
+            );
+        }
 
         // 每个 started 的 total 一致；index 表示已完成数量，不能超过总数。
         assert!(started.iter().all(|event| event.index <= expected_urls));