From 99e6b3a23afa449690175bb5164dc77e3ef37877 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Fri, 31 Jul 2026 13:12:21 +0800 Subject: [PATCH] =?UTF-8?q?fix(sync):=20=E7=AB=8B=E5=8D=B3=E6=A0=A1?= =?UTF-8?q?=E9=AA=8C=E5=AE=98=E6=96=B9=20hash=20sidecar?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- infrastructure/src/official_download.rs | 210 +++++++++++------------- 1 file changed, 94 insertions(+), 116 deletions(-) diff --git a/infrastructure/src/official_download.rs b/infrastructure/src/official_download.rs index f794a65..d3a3c5f 100644 --- a/infrastructure/src/official_download.rs +++ b/infrastructure/src/official_download.rs @@ -695,20 +695,15 @@ impl OfficialResourcePullService { }); } - // Phase B：顺序下载 need-download 项。每个 URL 的目标和 `.part` 都是独立 - // 的，但这里保留单线程执行，便于维持稳定进度、稳定日志和简单的失败恢复。 - let download_indices: Vec = planned - .iter() - .enumerate() - .filter(|(_, item)| item.existing.is_none()) - .map(|(index, _)| index) - .collect(); - let mut download_results: Vec>> = - (0..planned.len()).map(|_| None).collect(); + // Phase B：按 plan 顺序处理每个 URL。下载或复用完成并写入 manifest 后， + // 立即尝试校验已经到齐的官方 `.bytes/.hash` pair，避免把文件级问题延后到整轮末尾。 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::::new(); + let mut processed_urls = HashSet::::new(); - for &plan_index in &download_indices { - let item = &planned[plan_index]; + for item in &planned { 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); - match result { - Ok(mut pull_result) => { - let verification_result = self - .clear_quarantine_entry(&item.url) - .and_then(|_| { - self.record_download_manifest_entry( - &mut manifest, - &item.url, - &item.destination, - ) - }) - .and_then(|verification| { - self.write_download_manifest(&manifest) - .map(|_| verification) - }); - match verification_result { - Ok(verification) => { - pull_result.verification = verification; + let result = if let Some(existing) = &item.existing { + self.clear_quarantine_entry(&item.url)?; + existing.clone() + } else { + match self.pull_one(&item.url, &item.destination) { + Ok(mut pull_result) => { + let verification_result = self + .clear_quarantine_entry(&item.url) + .and_then(|_| { + self.record_download_manifest_entry( + &mut manifest, + &item.url, + &item.destination, + ) + }) + .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={}；{}", + item.url, + self.download_quarantine_path().display(), + error.message + ), + )); + } } - 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( + + 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={}；{}", @@ -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::::new(); - let mut processed_urls = HashSet::::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));