fix(sync): 补齐发布清理与校验进度

2026-09-18 11:34:59 +08:00 · 2026-07-31 13:00:53 +08:00
parent 8ec0e12795
commit 16e73327b4
4 changed files with 592 additions and 62 deletions
@@ -11,14 +11,15 @@ use bat_infrastructure::{
    read_parse_cache_at, read_snapshot, read_textunit_index_at, read_version_state,
    redact_proxy_url, resolve_curl_proxy, validate_output_root, validate_runtime_state_dir,
    write_file_atomic, CurlProxyConfig, CurlProxyMode, OfficialEndpointMarkerRole,
-    OfficialFailedVersionRecord, OfficialServerInfoSource, OfficialTextUnitQuery,
-    OfficialUpdateConfig, OfficialUpdateProgress, OfficialUpdateReport, OfficialUpdateService,
-    OfficialUpdateSnapshot, OfficialUpdateStatus, OfficialVerificationSummary,
-    OfficialVersionRecord, OfficialVersionState, PatchApplyKind, PatchApplyParams,
-    PatchApplyReport, SqliteResourceRepository, UnityFsFieldPatchParams, UnityFsPatchReport,
-    UnityFsStringFieldPatchParams, UnityFsTextAssetPatchParams, LOCALIZED_CURRENT_LINK,
-    LOCALIZED_PATCH_MANIFEST_FILE, LOCALIZED_VERSIONS_DIR, LOCALIZED_VERSION_STATE_FILE,
-    OFFICIAL_PARSE_CACHE_FILE, OFFICIAL_TEXTUNIT_INDEX_FILE, PRIVATE_FILE_MODE,
+    OfficialFailedVersionRecord, OfficialResourceHashVerification, OfficialResourceVerification,
+    OfficialServerInfoSource, OfficialTextUnitQuery, OfficialUpdateConfig, OfficialUpdateProgress,
+    OfficialUpdateReport, OfficialUpdateService, OfficialUpdateSnapshot, OfficialUpdateStatus,
+    OfficialVerificationSummary, OfficialVersionRecord, OfficialVersionState, PatchApplyKind,
+    PatchApplyParams, PatchApplyReport, SqliteResourceRepository, UnityFsFieldPatchParams,
+    UnityFsPatchReport, UnityFsStringFieldPatchParams, UnityFsTextAssetPatchParams,
+    LOCALIZED_CURRENT_LINK, LOCALIZED_PATCH_MANIFEST_FILE, LOCALIZED_VERSIONS_DIR,
+    LOCALIZED_VERSION_STATE_FILE, OFFICIAL_PARSE_CACHE_FILE, OFFICIAL_TEXTUNIT_INDEX_FILE,
+    PRIVATE_FILE_MODE,
 };
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
@@ -48,6 +49,7 @@ const DAEMON_LOG_FILE: &str = "bat-daemon.log";
 const DAEMON_STRUCTURED_LOG_FILE: &str = "bat-events.jsonl";
 const DAEMON_SOCKET_FILE: &str = "bat.sock";
 const DAEMON_CONTROL_LOCK_FILE: &str = "bat-control.lock";
+const SHUTDOWN_SIGNAL_POLL_MILLISECONDS: u64 = 100;
 /// 后台进程保存代理凭据的专用文件，仅供 restart/reload 复用，永不序列化到状态输出。
 const DAEMON_PROXY_SECRET_FILE: &str = "bat-proxy.secret";
 /// 代理凭据下传子进程使用的环境变量；避免凭据出现在子进程 argv（/proc/<pid>/cmdline）。
@@ -354,21 +356,22 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
    if options.daemon_child {
        validate_runtime_state_dir(&options.state_dir).map_err(anyhow::Error::msg)?;
    }
+    install_shutdown_signal_handlers()?;
+    clear_shutdown_signal_request();
    let service = OfficialUpdateService::new();
    let mut logger = ProgressLogger::new(options.progress);
    let daemon_state_dir = options.state_dir.clone();
    if options.daemon_child {
        logger.attach_structured_log(daemon_structured_log_path(&daemon_state_dir));
    }
-    let daemon_control = if options.daemon_child {
-        Some(new_daemon_control())
-    } else {
-        None
-    };
+    // 前台 watch 也使用控制状态，使信号停止和 daemon/RPC 停止共享同一条退出路径。
+    let daemon_control = Some(new_daemon_control());
    // 进程内同步锁：watch 循环与任务 worker 在跑同步前都获取它，互相等待而非撞文件锁失败。
    let sync_lock = Arc::new(Mutex::new(()));
-    let (_task_worker, _task_context, _rpc_server) = if let Some(control) = daemon_control.as_ref()
-    {
+    let (_task_worker, _task_context, _rpc_server) = if options.daemon_child {
+        let control = daemon_control
+            .as_ref()
+            .expect("daemon control must exist for daemon child");
        // 任务历史持久化在 state dir（此前已通过 validate_runtime_state_dir 校验）。
        let (registry, restore_summary) = TaskRegistry::with_persistence(&daemon_state_dir);
        logger.log_text("daemon", format!("任务历史：{restore_summary}"));
@@ -403,7 +406,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
        ),
    );
    loop {
-        if daemon_control_stop_requested(daemon_control.as_ref()) {
+        if watch_stop_requested(daemon_control.as_ref()) {
            logger.log_text("daemon", "收到停止请求，watch 循环准备退出");
            break;
        }
@@ -472,7 +475,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
                    );
                    logger.log(event);
                },
-                || daemon_control_stop_requested(daemon_control.as_ref()),
+                || watch_stop_requested(daemon_control.as_ref()),
            )
        };
        match run_result {
@@ -566,7 +569,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
                    next_retry_seconds: Some(sleep_for.as_secs()),
                };
                eprintln!("{}", serde_json::to_string_pretty(&payload)?);
-                if daemon_control_stop_requested(daemon_control.as_ref()) {
+                if watch_stop_requested(daemon_control.as_ref()) {
                    logger.log_text("daemon", "停止请求已中断本轮同步，watch 循环准备退出");
                    break;
                }
@@ -576,7 +579,7 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> {
        match wait_for_daemon_wake(daemon_control.as_ref(), sleep_for) {
            DaemonWake::Timeout => {}
            DaemonWake::Stop => {
-                logger.log_text("daemon", "收到停止请求，watch 循环准备退出");
+                logger.log_text("daemon", "收到停止请求或进程信号，watch 循环准备退出");
                break;
            }
            DaemonWake::Refresh { force } => {
@@ -672,6 +675,10 @@ struct DaemonDownloadProgress {
    failure_attempts: Option<usize>,
    #[serde(default)]
    quarantined: Option<bool>,
+    #[serde(default)]
+    verification: Option<OfficialResourceVerification>,
+    #[serde(default)]
+    official_hash: Option<OfficialResourceHashVerification>,
 }

 #[derive(Debug)]
@@ -1737,10 +1744,68 @@ fn lexically_normalize_path(path: &Path) -> PathBuf {
    }
 }

+#[cfg(unix)]
+static SHUTDOWN_SIGNAL_REQUESTED: AtomicBool = AtomicBool::new(false);
+
+#[cfg(unix)]
+extern "C" fn handle_shutdown_signal(_signal: libc::c_int) {
+    // Signal handlers may only touch lock-free process state; the watch loop
+    // performs logging, cancellation, and cleanup after observing this flag.
+    SHUTDOWN_SIGNAL_REQUESTED.store(true, Ordering::SeqCst);
+}
+
+#[cfg(unix)]
+fn install_shutdown_signal_handlers() -> anyhow::Result<()> {
+    unsafe {
+        if libc::signal(
+            libc::SIGINT,
+            handle_shutdown_signal as *const () as libc::sighandler_t,
+        ) == libc::SIG_ERR
+        {
+            return Err(anyhow::anyhow!("注册 SIGINT 优雅退出处理器失败"));
+        }
+        if libc::signal(
+            libc::SIGTERM,
+            handle_shutdown_signal as *const () as libc::sighandler_t,
+        ) == libc::SIG_ERR
+        {
+            return Err(anyhow::anyhow!("注册 SIGTERM 优雅退出处理器失败"));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(not(unix))]
+fn install_shutdown_signal_handlers() -> anyhow::Result<()> {
+    Ok(())
+}
+
+#[cfg(unix)]
+fn clear_shutdown_signal_request() {
+    SHUTDOWN_SIGNAL_REQUESTED.store(false, Ordering::SeqCst);
+}
+
+#[cfg(not(unix))]
+fn clear_shutdown_signal_request() {}
+
+#[cfg(unix)]
+fn shutdown_signal_requested() -> bool {
+    SHUTDOWN_SIGNAL_REQUESTED.load(Ordering::SeqCst)
+}
+
+#[cfg(not(unix))]
+fn shutdown_signal_requested() -> bool {
+    false
+}
+
 fn new_daemon_control() -> DaemonControl {
    Arc::new((Mutex::new(DaemonControlState::default()), Condvar::new()))
 }

+fn watch_stop_requested(control: Option<&DaemonControl>) -> bool {
+    shutdown_signal_requested() || daemon_control_stop_requested(control)
+}
+
 fn daemon_control_stop_requested(control: Option<&DaemonControl>) -> bool {
    let Some(control) = control else {
        return false;
@@ -1781,21 +1846,42 @@ fn daemon_control_request_reload(control: &DaemonControl) {
 }

 fn wait_for_daemon_wake(control: Option<&DaemonControl>, timeout: Duration) -> DaemonWake {
+    let started_at = Instant::now();
    let Some(control) = control else {
-        thread::sleep(timeout);
-        return DaemonWake::Timeout;
+        loop {
+            if shutdown_signal_requested() {
+                return DaemonWake::Stop;
+            }
+            let elapsed = started_at.elapsed();
+            if elapsed >= timeout {
+                return DaemonWake::Timeout;
+            }
+            let remaining = timeout.saturating_sub(elapsed);
+            thread::sleep(remaining.min(Duration::from_millis(SHUTDOWN_SIGNAL_POLL_MILLISECONDS)));
+        }
    };
    let (lock, cvar) = &**control;
    let Ok(mut state) = lock.lock() else {
        return DaemonWake::Stop;
    };
-    if let Some(wake) = take_daemon_wake(&mut state) {
-        return wake;
+    loop {
+        if shutdown_signal_requested() {
+            return DaemonWake::Stop;
+        }
+        if let Some(wake) = take_daemon_wake(&mut state) {
+            return wake;
+        }
+        let elapsed = started_at.elapsed();
+        if elapsed >= timeout {
+            return DaemonWake::Timeout;
+        }
+        let remaining = timeout.saturating_sub(elapsed);
+        let wait_for = remaining.min(Duration::from_millis(SHUTDOWN_SIGNAL_POLL_MILLISECONDS));
+        let Ok((next_state, _timeout)) = cvar.wait_timeout(state, wait_for) else {
+            return DaemonWake::Stop;
+        };
+        state = next_state;
    }
-    let Ok((mut state, _timeout)) = cvar.wait_timeout(state, timeout) else {
-        return DaemonWake::Stop;
-    };
-    take_daemon_wake(&mut state).unwrap_or(DaemonWake::Timeout)
 }

 fn take_daemon_wake(state: &mut DaemonControlState) -> Option<DaemonWake> {
@@ -4499,6 +4585,18 @@ fn print_list(label: &str, values: &[String], limit: usize) {

 fn format_daemon_download_progress(progress: &DaemonDownloadProgress) -> String {
    let status = progress.status.as_deref().unwrap_or("running");
+    if let Some(hash) = progress.official_hash.as_ref() {
+        return format!(
+            "{}/{} official_hash algorithm={} expected={} actual={} data={} hash={}",
+            progress.index,
+            progress.total,
+            hash.algorithm.as_str(),
+            hash.expected,
+            hash.actual,
+            hash.data_url,
+            hash.hash_url
+        );
+    }
    if status == "failed" {
        return format!(
            "{}/{} failed kind={} http={} retryable={} attempts={} quarantined={} {}",
@@ -4524,6 +4622,19 @@ fn format_daemon_download_progress(progress: &DaemonDownloadProgress) -> String
            progress.url
        );
    }
+    if let Some(verification) = progress.verification.as_ref() {
+        return format!(
+            "{}/{} {} bytes={} blake3={} zip_checked={} zip_verified={} {}",
+            progress.index,
+            progress.total,
+            status,
+            verification.actual_bytes,
+            verification.actual_blake3,
+            verification.zip_checked,
+            verification.zip_structure_verified,
+            progress.url
+        );
+    }
    format!(
        "{}/{} {} {}",
        progress.index, progress.total, status, progress.url
@@ -5631,6 +5742,8 @@ fn update_daemon_progress(state_dir: &Path, event: &OfficialUpdateProgress) -> a
            failure_retryable: event.download_failure_retryable,
            failure_attempts: event.download_failure_attempts,
            quarantined: event.download_quarantined,
+            verification: event.download_verification.clone(),
+            official_hash: event.official_hash_verification.clone(),
        }),
        _ if event.stage != "download" => None,
        _ => status.download_progress,
@@ -6169,6 +6282,8 @@ impl RotatingStructuredLogger {
                "failure_retryable": event.download_failure_retryable,
                "failure_attempts": event.download_failure_attempts,
                "quarantined": event.download_quarantined,
+                "verification": event.download_verification,
+                "official_hash": event.official_hash_verification,
            })),
        });
        let mut line = serde_json::to_vec(&payload)?;
@@ -9020,6 +9135,31 @@ mod tests {
        );
    }

+    #[cfg(unix)]
+    #[test]
+    fn shutdown_signals_wake_watch_without_mutating_daemon_control() {
+        clear_shutdown_signal_request();
+        let control = new_daemon_control();
+
+        handle_shutdown_signal(libc::SIGINT);
+        assert!(watch_stop_requested(Some(&control)));
+        assert_eq!(
+            wait_for_daemon_wake(Some(&control), Duration::from_secs(60)),
+            DaemonWake::Stop
+        );
+
+        clear_shutdown_signal_request();
+        handle_shutdown_signal(libc::SIGTERM);
+        assert!(watch_stop_requested(Some(&control)));
+        assert_eq!(
+            wait_for_daemon_wake(Some(&control), Duration::from_secs(60)),
+            DaemonWake::Stop
+        );
+
+        clear_shutdown_signal_request();
+        assert!(!watch_stop_requested(Some(&control)));
+    }
+
    #[test]
    fn daemon_control_lock_rejects_active_owner() {
        let temp = tempfile::TempDir::new().unwrap();
@@ -9151,6 +9291,21 @@ mod tests {
        event.download_index = Some(1);
        event.download_total = Some(2);
        event.download_url = Some("https://prod-clientpatch.bluearchiveyostar.com/a.zip".into());
+        event.download_verification = Some(OfficialResourceVerification {
+            expected_bytes: Some(123),
+            actual_bytes: 123,
+            expected_blake3: Some("expected-blake3".to_string()),
+            actual_blake3: "expected-blake3".to_string(),
+            zip_checked: true,
+            zip_structure_verified: true,
+        });
+        event.official_hash_verification = Some(OfficialResourceHashVerification {
+            data_url: "https://prod-clientpatch.bluearchiveyostar.com/a.bytes".to_string(),
+            hash_url: "https://prod-clientpatch.bluearchiveyostar.com/a.hash".to_string(),
+            algorithm: bat_infrastructure::OfficialResourceHashAlgorithm::XxHash32Decimal,
+            expected: "2044170421".to_string(),
+            actual: "2044170421".to_string(),
+        });

        for _ in 0..8 {
            logger
@@ -9170,6 +9325,16 @@ mod tests {
        assert_eq!(value["stage"], "download");
        assert_eq!(value["download"]["index"], 1);
        assert_eq!(value["download"]["total"], 2);
+        assert_eq!(value["download"]["verification"]["actual_bytes"], 123);
+        assert_eq!(
+            value["download"]["verification"]["zip_structure_verified"],
+            true
+        );
+        assert_eq!(
+            value["download"]["official_hash"]["algorithm"],
+            "xxhash32_decimal"
+        );
+        assert_eq!(value["download"]["official_hash"]["expected"], "2044170421");
    }

    #[cfg(unix)]