fix(sync): 移除多线程下载并补齐staging复用回归

infrastructure/src/official_download.rs

@@ -16,9 +16,6 @@ use std::fs::{self, File};
 use std::io::Read;
 use std::path::{Path, PathBuf};
 use std::process::Command;
-use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::sync::mpsc;
-use std::thread;
 use std::time::{SystemTime, UNIX_EPOCH};
 
 /// 官方资源下载错误：携带统一错误码，便于 CLI/RPC 归类。
@@ -64,10 +61,6 @@ const DOWNLOAD_QUARANTINE_FILE: &str = "official-download-quarantine.json";
 const DOWNLOAD_MANIFEST_VERSION: u32 = 1;
 const DOWNLOAD_QUARANTINE_VERSION: u32 = 1;
 const DEFAULT_RETRY_ATTEMPTS: usize = 3;
-/// 默认下载并发度。保守取值，对官方 CDN 礼貌；可经配置调到 1..=256。
-pub(crate) const DEFAULT_DOWNLOAD_CONCURRENCY: usize = 4;
-/// 下载并发度上限。
-const MAX_DOWNLOAD_CONCURRENCY: usize = 256;
 
 /// Outcome for one official resource pull item.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -493,7 +486,6 @@ pub struct OfficialResourcePullService {
     curl_command: PathBuf,
     curl_proxy: CurlProxyConfig,
     retry_attempts: usize,
-    download_concurrency: usize,
 }
 
 impl OfficialResourcePullService {
@@ -512,7 +504,6 @@ impl OfficialResourcePullService {
             curl_command: curl_command.into(),
             curl_proxy: CurlProxyConfig::default(),
             retry_attempts: DEFAULT_RETRY_ATTEMPTS,
-            download_concurrency: DEFAULT_DOWNLOAD_CONCURRENCY,
         }
     }
 
@@ -528,17 +519,6 @@ impl OfficialResourcePullService {
         self
     }
 
-    /// 设置下载并发度（并行执行的网络下载数），钳制到 `1..=256`。
-    pub fn with_download_concurrency(mut self, concurrency: usize) -> Self {
-        self.download_concurrency = concurrency.clamp(1, MAX_DOWNLOAD_CONCURRENCY);
-        self
-    }
-
-    /// 返回当前下载并发度。
-    pub fn download_concurrency(&self) -> usize {
-        self.download_concurrency
-    }
-
     /// Returns the output root used for downloaded files.
     pub fn output_root(&self) -> &Path {
         &self.output_root
@@ -625,11 +605,8 @@ impl OfficialResourcePullService {
             });
         }
 
-        // Phase B：并发下载 need-download 项（各 URL 目标/`.part` 相互独立，
-        // 天然可并行）。worker 只做只读 `&self` 的网络下载，经 mpsc 把结果送回
-        // 主线程；manifest/quarantine 簿记与进度回调全部在主线程串行完成，无需
-        // 加锁。首个失败或 `should_cancel` 会置 cancel 标志，其余 worker 在下一
-        // 个任务边界停止，保持 fail-fast 与「不发布不完整资源」不变量。
+        // Phase B：顺序下载 need-download 项。每个 URL 的目标和 `.part` 都是独立
+        // 的，但这里保留单线程执行，便于维持稳定进度、稳定日志和简单的失败恢复。
         let download_indices: Vec<usize> = planned
             .iter()
             .enumerate()
@@ -640,113 +617,68 @@ impl OfficialResourcePullService {
             (0..planned.len()).map(|_| None).collect();
         let mut completed_count = 0usize;
 
-        if !download_indices.is_empty() {
-            let concurrency = self.download_concurrency.clamp(1, MAX_DOWNLOAD_CONCURRENCY);
-            let cursor = AtomicUsize::new(0);
-            let cancel = AtomicBool::new(false);
-            let (sender, receiver) = mpsc::channel::<WorkerMessage>();
-
-            thread::scope(|scope| {
-                for _ in 0..concurrency.min(download_indices.len()) {
-                    let sender = sender.clone();
-                    let cursor = &cursor;
-                    let cancel = &cancel;
-                    let download_indices = &download_indices;
-                    let planned = &planned;
-                    let service = &*self;
-                    scope.spawn(move || loop {
-                        if cancel.load(Ordering::Relaxed) {
-                            break;
-                        }
-                        let slot = cursor.fetch_add(1, Ordering::Relaxed);
-                        let Some(&plan_index) = download_indices.get(slot) else {
-                            break;
-                        };
-                        let item = &planned[plan_index];
-                        if sender.send(WorkerMessage::Started { plan_index }).is_err() {
-                            break;
-                        }
-                        let result = service.pull_one(&item.url, &item.destination);
-                        if result.is_err() {
-                            cancel.store(true, Ordering::Relaxed);
-                        }
-                        if sender
-                            .send(WorkerMessage::Done { plan_index, result })
-                            .is_err()
-                        {
-                            break;
-                        }
-                    });
-                }
-                // 主线程持有的 sender 副本必须丢弃，否则 receiver 永不结束。
-                drop(sender);
-
-                for message in receiver {
-                    match message {
-                        WorkerMessage::Started { plan_index } => {
-                            let item = &planned[plan_index];
-                            progress(OfficialResourcePullProgress::started(
-                                completed_count,
-                                total,
-                                item.url.clone(),
-                            ));
-                            // 停止请求：置 cancel，让 worker 在下个任务边界退出。
-                            if should_cancel() {
-                                cancel.store(true, Ordering::Relaxed);
-                            }
-                        }
-                        WorkerMessage::Done { plan_index, result } => {
-                            if let Ok(pull_result) = &result {
-                                let item = &planned[plan_index];
-                                // manifest/quarantine 簿记在主线程串行执行。
-                                if let Err(error) = self
-                                    .clear_quarantine_entry(&item.url)
-                                    .and_then(|_| {
-                                        self.record_download_manifest_entry(
-                                            &mut manifest,
-                                            &item.url,
-                                            &item.destination,
-                                        )
-                                    })
-                                    .and_then(|_| self.write_download_manifest(&manifest))
-                                {
-                                    // 簿记失败：记为该项错误并触发 fail-fast。
-                                    cancel.store(true, Ordering::Relaxed);
-                                    download_results[plan_index] = Some(Err(PullOneError::plain(
-                                        format!("记录下载 manifest 失败：{error}"),
-                                    )));
-                                    continue;
-                                }
-                                completed_count += 1;
-                                progress(OfficialResourcePullProgress::finished(
-                                    completed_count,
-                                    total,
-                                    item.url.clone(),
-                                    pull_result.status,
-                                    pull_result.bytes,
-                                    pull_result.transferred_bytes,
-                                ));
-                            }
-                            download_results[plan_index] = Some(result);
-                        }
-                    }
-                }
-            });
-
-            if should_cancel() && download_results.iter().any(Option::is_none) {
+        for &plan_index in &download_indices {
+            let item = &planned[plan_index];
+            progress(OfficialResourcePullProgress::started(
+                completed_count,
+                total,
+                item.url.clone(),
+            ));
+            if should_cancel() {
                 return Err("官方资源拉取已被停止请求中断".to_string().into());
             }
 
-            // fail-fast：按 plan 顺序取首个失败项，记 quarantine、发 failed 进度并中止。
-            for &plan_index in &download_indices {
-                if let Some(Some(Err(error))) = download_results.get(plan_index) {
-                    let item = &planned[plan_index];
-                    self.record_quarantine_entry(&item.url, &item.destination, error)?;
+            let result = self.pull_one(&item.url, &item.destination);
+            match result {
+                Ok(pull_result) => {
+                    if let Err(error) = self
+                        .clear_quarantine_entry(&item.url)
+                        .and_then(|_| {
+                            self.record_download_manifest_entry(
+                                &mut manifest,
+                                &item.url,
+                                &item.destination,
+                            )
+                        })
+                        .and_then(|_| self.write_download_manifest(&manifest))
+                    {
+                        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
+                            ),
+                        ));
+                    }
+
+                    completed_count += 1;
+                    progress(OfficialResourcePullProgress::finished(
+                        completed_count,
+                        total,
+                        item.url.clone(),
+                        pull_result.status,
+                        pull_result.bytes,
+                        pull_result.transferred_bytes,
+                    ));
+                    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,
+                        &error,
                     ));
                     return Err(DownloadError::new(
                         error.error_code(),
@@ -1771,17 +1703,6 @@ struct PlannedDownload {
     existing: Option<PullOneResult>,
 }
 
-/// worker 线程经 mpsc 送回主线程的消息。
-enum WorkerMessage {
-    /// worker 已领取某计划项、即将下载（主线程据此发 started 进度）。
-    Started { plan_index: usize },
-    /// 某计划项下载结束（成功或失败）。
-    Done {
-        plan_index: usize,
-        result: Result<PullOneResult, PullOneError>,
-    },
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct PullOneResult {
     bytes: u64,
@@ -3084,16 +3005,14 @@ exit 22
     }
 
     #[test]
-    fn downloads_run_concurrently_and_each_url_reports_once() {
+    fn downloads_run_sequentially_and_each_url_reports_once() {
         let out_dir = TempDir::new().unwrap();
         let bin_dir = TempDir::new().unwrap();
         let curl_path = bin_dir.path().join("curl");
         write_fake_curl(&curl_path);
 
-        // 并发度 8：每个 URL 恰好一次 started + 一次 finished，全部文件落盘。
-        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
-            .with_download_concurrency(8);
-        assert_eq!(service.download_concurrency(), 8);
+        // 顺序下载：每个 URL 恰好一次 started + 一次 finished，全部文件落盘。
+        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
         let plan = build_official_pull_plan_for_platforms(
             discovery_plan(),
             inventory(),
@@ -3127,24 +3046,6 @@ exit 22
         assert_eq!(manifest.entries.len(), all_urls.len());
     }
 
-    #[test]
-    fn download_concurrency_is_clamped() {
-        let service = OfficialResourcePullService::with_curl_command("/tmp/unused", "curl");
-        assert_eq!(service.download_concurrency(), DEFAULT_DOWNLOAD_CONCURRENCY);
-        assert_eq!(
-            OfficialResourcePullService::with_curl_command("/tmp/unused", "curl")
-                .with_download_concurrency(0)
-                .download_concurrency(),
-            1
-        );
-        assert_eq!(
-            OfficialResourcePullService::with_curl_command("/tmp/unused", "curl")
-                .with_download_concurrency(9999)
-                .download_concurrency(),
-            MAX_DOWNLOAD_CONCURRENCY
-        );
-    }
-
     #[test]
     fn retries_transient_download_failures() {
         let out_dir = TempDir::new().unwrap();