fix: 拆分 bat 控制面

infrastructure/src/bin/bat/task_registry.rs

@@ -0,0 +1,587 @@
+use super::*;
+
+pub(super) const MAX_RETAINED_TASKS: usize = 64;
+/// 每个任务保留的进度日志行数上限。
+pub(super) const MAX_TASK_LOG_LINES: usize = 200;
+
+/// 任务类型：目前覆盖官方同步、校验与 catalog 更新检查。
+#[derive(Debug, Clone, Copy)]
+pub(super) enum TaskKind {
+    Sync,
+    Verify,
+    Repair,
+    /// catalog 更新检查：只做发现 + 拉取计划（dry-run），不下载不审计。
+    Refresh,
+}
+
+impl TaskKind {
+    pub(super) fn method(self) -> &'static str {
+        match self {
+            Self::Sync => RPC_METHOD_RESOURCE_SYNC,
+            Self::Verify => RPC_METHOD_RESOURCE_VERIFY,
+            Self::Repair => RPC_METHOD_RESOURCE_REPAIR,
+            Self::Refresh => RPC_METHOD_CATALOG_REFRESH,
+        }
+    }
+
+    /// 由 daemon 基准配置派生该任务的实际同步配置。
+    pub(super) fn build_config(
+        self,
+        base: &OfficialUpdateConfig,
+        force: bool,
+    ) -> OfficialUpdateConfig {
+        let mut config = base.clone();
+        match self {
+            Self::Sync => {
+                config.dry_run = false;
+                config.force = config.force || force;
+            }
+            Self::Verify => {
+                config.dry_run = true;
+                config.plan = true;
+                config.audit_local = true;
+                config.repair = false;
+                config.force = false;
+            }
+            Self::Repair => {
+                config.dry_run = false;
+                config.audit_local = true;
+                config.repair = true;
+                config.force = false;
+            }
+            Self::Refresh => {
+                config.dry_run = true;
+                config.plan = true;
+                config.audit_local = false;
+                config.repair = false;
+                config.force = force;
+            }
+        }
+        config
+    }
+}
+
+/// 请求取消任务的结果。
+pub(super) enum CancelOutcome {
+    Requested,
+    AlreadyFinished,
+    NotFound,
+}
+
+/// 单个任务的可轮询记录。
+#[derive(Debug, Clone, Serialize)]
+pub(super) struct TaskRecord {
+    pub(super) id: String,
+    pub(super) kind: &'static str,
+    /// `queued` | `running` | `succeeded` | `failed` | `cancelled`。
+    pub(super) status: &'static str,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) stage: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) message: Option<String>,
+    pub(super) created_at: u64,
+    pub(super) updated_at: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) started_at: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) finished_at: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) error: Option<ApiError>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(super) result: Option<serde_json::Value>,
+    /// 取消标志，worker 的 should_cancel 检查它；不参与序列化。
+    #[serde(skip)]
+    pub(super) cancel: Arc<AtomicBool>,
+    /// 进度日志（有界），经 task.logs 返回；不参与 task.status 序列化。
+    #[serde(skip)]
+    pub(super) log: Vec<String>,
+}
+
+impl TaskRecord {
+    pub(super) fn is_finished(&self) -> bool {
+        matches!(self.status, "succeeded" | "failed" | "cancelled")
+    }
+}
+
+struct TaskStore {
+    tasks: HashMap<String, TaskRecord>,
+    order: Vec<String>,
+    seq: u64,
+    /// 任务历史持久化文件路径；`None` 表示纯内存（测试等非 daemon 场景）。
+    persist_path: Option<PathBuf>,
+}
+
+/// daemon 任务历史持久化文件名（位于 state dir 内，`0600` 原子写）。
+pub(super) const TASKS_FILE_NAME: &str = "bat-tasks.json";
+/// 任务历史文件结构版本。
+pub(super) const TASKS_FILE_VERSION: u32 = 1;
+
+/// 任务历史文件的持久化形态（版本化；daemon 重启后恢复任务历史用）。
+#[derive(Debug, Serialize, Deserialize)]
+pub(super) struct PersistedTaskFile {
+    pub(super) version: u32,
+    /// 任务 ID 序号计数器；恢复它避免 pid 复用时新任务与历史任务撞 ID。
+    pub(super) seq: u64,
+    pub(super) tasks: Vec<PersistedTaskRecord>,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub(super) struct PersistedTaskRecord {
+    id: String,
+    kind: String,
+    pub(super) status: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    stage: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    message: Option<String>,
+    created_at: u64,
+    updated_at: u64,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    started_at: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    finished_at: Option<u64>,
+    /// `ApiError` 的序列化形态（code/kind/domain/location/message/retryable）。
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    error: Option<serde_json::Value>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    result: Option<serde_json::Value>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    log: Vec<String>,
+}
+
+/// 把持久化的任务类型映射回静态字符串；未识别（如未来版本新增）返回 `None`。
+fn task_kind_static(kind: &str) -> Option<&'static str> {
+    match kind {
+        RPC_METHOD_RESOURCE_SYNC => Some(RPC_METHOD_RESOURCE_SYNC),
+        RPC_METHOD_RESOURCE_VERIFY => Some(RPC_METHOD_RESOURCE_VERIFY),
+        RPC_METHOD_RESOURCE_REPAIR => Some(RPC_METHOD_RESOURCE_REPAIR),
+        RPC_METHOD_CATALOG_REFRESH => Some(RPC_METHOD_CATALOG_REFRESH),
+        _ => None,
+    }
+}
+
+/// 把持久化的任务状态映射回静态字符串；未识别返回 `None`。
+fn task_status_static(status: &str) -> Option<&'static str> {
+    match status {
+        "queued" => Some("queued"),
+        "running" => Some("running"),
+        "succeeded" => Some("succeeded"),
+        "failed" => Some("failed"),
+        "cancelled" => Some("cancelled"),
+        _ => None,
+    }
+}
+
+impl PersistedTaskRecord {
+    fn from_record(record: &TaskRecord) -> Self {
+        Self {
+            id: record.id.clone(),
+            kind: record.kind.to_string(),
+            status: record.status.to_string(),
+            stage: record.stage.clone(),
+            message: record.message.clone(),
+            created_at: record.created_at,
+            updated_at: record.updated_at,
+            started_at: record.started_at,
+            finished_at: record.finished_at,
+            error: record
+                .error
+                .as_ref()
+                .and_then(|error| serde_json::to_value(error).ok()),
+            result: record.result.clone(),
+            log: record.log.clone(),
+        }
+    }
+
+    /// 还原为内存任务记录；kind/status 未识别时返回 `None`（调用方计数跳过）。
+    fn into_record(self) -> Option<TaskRecord> {
+        let kind = task_kind_static(&self.kind)?;
+        let status = task_status_static(&self.status)?;
+        // 错误从序列化形态还原：code 经码表反查（未登记回退 internal），
+        // location 固定为任务执行器（当前全部任务错误的唯一来源）。
+        let error = self.error.as_ref().map(|value| {
+            let code = value
+                .get("code")
+                .and_then(serde_json::Value::as_str)
+                .and_then(ErrorCode::from_id)
+                .unwrap_or(ErrorCode::INTERNAL);
+            let message = value
+                .get("message")
+                .and_then(serde_json::Value::as_str)
+                .unwrap_or("<持久化错误信息缺失>")
+                .to_string();
+            ApiError::new(code, "task.executor", message)
+        });
+        Some(TaskRecord {
+            id: self.id,
+            kind,
+            status,
+            stage: self.stage,
+            message: self.message,
+            created_at: self.created_at,
+            updated_at: self.updated_at,
+            started_at: self.started_at,
+            finished_at: self.finished_at,
+            error,
+            result: self.result,
+            cancel: Arc::new(AtomicBool::new(false)),
+            log: self.log,
+        })
+    }
+}
+
+/// 读取任务历史文件。文件缺失返回 `Ok(None)`；symlink、解析失败或版本不支持返回 `Err`。
+fn load_persisted_tasks(path: &Path) -> Result<Option<PersistedTaskFile>, String> {
+    let Some(bytes) = read_file_no_symlink(path, "任务历史")? else {
+        return Ok(None);
+    };
+    let file: PersistedTaskFile = serde_json::from_slice(&bytes)
+        .map_err(|error| format!("解析任务历史失败 {}：{error}", path.display()))?;
+    if file.version != TASKS_FILE_VERSION {
+        return Err(format!(
+            "不支持的任务历史版本 {}，文件 {}",
+            file.version,
+            path.display()
+        ));
+    }
+    Ok(Some(file))
+}
+
+/// 任务注册表句柄：包住内存存储，供 RPC handler 与 worker 共享。
+///
+/// 通过方法访问（而非直接摸内部 map），便于将来换成 Redis 等持久化后端。
+#[derive(Clone)]
+pub(super) struct TaskRegistry {
+    inner: Arc<Mutex<TaskStore>>,
+}
+
+impl TaskRegistry {
+    /// 纯内存注册表（无持久化）；生产 daemon 走 [`Self::with_persistence`]。
+    #[cfg(test)]
+    pub(super) fn new() -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(TaskStore {
+                tasks: HashMap::new(),
+                order: Vec::new(),
+                seq: 0,
+                persist_path: None,
+            })),
+        }
+    }
+
+    /// 从 state dir 恢复任务历史并启用持久化。
+    ///
+    /// 中断时仍处于 queued/running 的任务标记为 `failed`（`TASK_INTERRUPTED`）；
+    /// 文件缺失按空历史处理；文件损坏或版本不支持时改名 `.corrupt` 留证并从
+    /// 空历史开始。返回注册表与恢复摘要（供 daemon 日志记录）。
+    pub(super) fn with_persistence(state_dir: &Path) -> (Self, String) {
+        let path = state_dir.join(TASKS_FILE_NAME);
+        let now = unix_seconds_now();
+        let mut seq = 0;
+        let mut tasks = HashMap::new();
+        let mut order = Vec::new();
+        let summary = match load_persisted_tasks(&path) {
+            Ok(None) => "无历史任务文件，从空任务历史开始".to_string(),
+            Ok(Some(file)) => {
+                seq = file.seq;
+                let total = file.tasks.len();
+                let mut interrupted = 0usize;
+                let mut skipped = 0usize;
+                for persisted in file.tasks {
+                    let Some(mut record) = persisted.into_record() else {
+                        skipped += 1;
+                        continue;
+                    };
+                    if !record.is_finished() {
+                        interrupted += 1;
+                        record.status = "failed";
+                        record.finished_at = Some(now);
+                        record.updated_at = now;
+                        record.error = Some(ApiError::new(
+                            ErrorCode::TASK_INTERRUPTED,
+                            "task.executor",
+                            "daemon 停止/重启导致任务中断",
+                        ));
+                        record
+                            .log
+                            .push("[daemon] 任务因 daemon 停止/重启而中断".to_string());
+                    }
+                    if tasks.insert(record.id.clone(), record.clone()).is_none() {
+                        order.push(record.id);
+                    } else {
+                        skipped += 1;
+                    }
+                }
+                format!("恢复任务历史 {total} 条（标记中断 {interrupted} 条，跳过无法识别 {skipped} 条）")
+            }
+            Err(error) => {
+                // 保留损坏文件供诊断（改名而非覆盖），从空历史开始。
+                let corrupt = path.with_extension("json.corrupt");
+                if fs::rename(&path, &corrupt).is_ok() {
+                    format!(
+                        "任务历史不可用（{error}）；原文件已改名保留为 {}",
+                        corrupt.display()
+                    )
+                } else {
+                    format!("任务历史不可用（{error}）；且无法改名保留原文件")
+                }
+            }
+        };
+        let registry = Self {
+            inner: Arc::new(Mutex::new(TaskStore {
+                tasks,
+                order,
+                seq,
+                persist_path: Some(path),
+            })),
+        };
+        // 把中断标记（或空历史）立即写回，保证文件与内存视图一致。
+        registry.lock().persist();
+        (registry, summary)
+    }
+
+    fn lock(&self) -> std::sync::MutexGuard<'_, TaskStore> {
+        self.inner
+            .lock()
+            .unwrap_or_else(|poison| poison.into_inner())
+    }
+
+    /// 创建 queued 任务并返回 task_id。
+    pub(super) fn create(&self, kind: TaskKind) -> String {
+        let now = unix_seconds_now();
+        let mut store = self.lock();
+        store.seq += 1;
+        let id = format!("task-{}-{}", std::process::id(), store.seq);
+        let record = TaskRecord {
+            id: id.clone(),
+            kind: kind.method(),
+            status: "queued",
+            stage: None,
+            message: None,
+            created_at: now,
+            updated_at: now,
+            started_at: None,
+            finished_at: None,
+            error: None,
+            result: None,
+            cancel: Arc::new(AtomicBool::new(false)),
+            log: Vec::new(),
+        };
+        store.tasks.insert(id.clone(), record);
+        store.order.push(id.clone());
+        store.prune();
+        store.persist();
+        id
+    }
+
+    pub(super) fn update<F: FnOnce(&mut TaskRecord)>(&self, id: &str, update: F) {
+        let mut store = self.lock();
+        let mut status_changed = false;
+        if let Some(record) = store.tasks.get_mut(id) {
+            let previous_status = record.status;
+            update(record);
+            record.updated_at = unix_seconds_now();
+            status_changed = record.status != previous_status;
+        }
+        // 只在生命周期转换时落盘；stage/message/log 的高频进度更新以内存为准，
+        // 随下一次转换一起写入（避免每个进度事件一次磁盘写）。
+        if status_changed {
+            store.persist();
+        }
+    }
+
+    pub(super) fn get(&self, id: &str) -> Option<TaskRecord> {
+        self.lock().tasks.get(id).cloned()
+    }
+
+    /// 返回任务的取消标志（与 worker 共享同一 Arc）。
+    pub(super) fn cancel_flag(&self, id: &str) -> Option<Arc<AtomicBool>> {
+        self.lock()
+            .tasks
+            .get(id)
+            .map(|record| Arc::clone(&record.cancel))
+    }
+
+    /// 追加一行进度日志，超出上限时丢弃最旧的。
+    pub(super) fn append_log(&self, id: &str, line: String) {
+        let mut store = self.lock();
+        if let Some(record) = store.tasks.get_mut(id) {
+            record.log.push(line);
+            if record.log.len() > MAX_TASK_LOG_LINES {
+                let overflow = record.log.len() - MAX_TASK_LOG_LINES;
+                record.log.drain(0..overflow);
+            }
+        }
+    }
+
+    /// 返回任务的进度日志。
+    pub(super) fn logs(&self, id: &str) -> Option<Vec<String>> {
+        self.lock().tasks.get(id).map(|record| record.log.clone())
+    }
+
+    /// 请求取消任务：未结束的置取消标志，已结束的原样返回，不存在返回 NotFound。
+    pub(super) fn request_cancel(&self, id: &str) -> CancelOutcome {
+        let store = self.lock();
+        match store.tasks.get(id) {
+            None => CancelOutcome::NotFound,
+            Some(record) if record.is_finished() => CancelOutcome::AlreadyFinished,
+            Some(record) => {
+                record.cancel.store(true, Ordering::Relaxed);
+                CancelOutcome::Requested
+            }
+        }
+    }
+
+    /// 返回全部任务，最新创建的在前。
+    pub(super) fn list(&self) -> Vec<TaskRecord> {
+        let store = self.lock();
+        store
+            .order
+            .iter()
+            .rev()
+            .filter_map(|id| store.tasks.get(id).cloned())
+            .collect()
+    }
+}
+
+impl TaskStore {
+    /// 把当前任务历史落盘（`0600` 原子写、不跟随 symlink）。
+    ///
+    /// 持久化未启用时为 no-op；写失败只记 stderr（进 daemon 日志），
+    /// 不让持久化故障拖垮任务执行本身。
+    fn persist(&self) {
+        let Some(path) = &self.persist_path else {
+            return;
+        };
+        let file = PersistedTaskFile {
+            version: TASKS_FILE_VERSION,
+            seq: self.seq,
+            tasks: self
+                .order
+                .iter()
+                .filter_map(|id| self.tasks.get(id))
+                .map(PersistedTaskRecord::from_record)
+                .collect(),
+        };
+        match serde_json::to_vec_pretty(&file) {
+            Ok(bytes) => {
+                if let Err(error) = write_file_atomic(path, &bytes, PRIVATE_FILE_MODE, "任务历史")
+                {
+                    eprintln!("[daemon] 任务历史落盘失败：{error}");
+                }
+            }
+            Err(error) => eprintln!("[daemon] 任务历史序列化失败：{error}"),
+        }
+    }
+
+    /// 裁剪最旧的已结束任务，把内存占用控制在上限内；运行中/排队中的任务不裁剪。
+    fn prune(&mut self) {
+        while self.order.len() > MAX_RETAINED_TASKS {
+            let Some(position) = self.order.iter().position(|id| {
+                self.tasks
+                    .get(id)
+                    .map(TaskRecord::is_finished)
+                    .unwrap_or(true)
+            }) else {
+                break;
+            };
+            let id = self.order.remove(position);
+            self.tasks.remove(&id);
+        }
+    }
+}
+
+/// 提交给任务 worker 的作业（配置已按任务类型派生完毕）。
+pub(super) struct TaskJob {
+    pub(super) id: String,
+    pub(super) config: OfficialUpdateConfig,
+    /// 与任务记录共享的取消标志。
+    pub(super) cancel: Arc<AtomicBool>,
+}
+
+/// daemon 任务上下文：RPC handler 借它创建任务、入队和读取。
+#[derive(Clone)]
+pub(super) struct DaemonTaskContext {
+    pub(super) registry: TaskRegistry,
+    pub(super) queue: mpsc::Sender<TaskJob>,
+    pub(super) base_config: OfficialUpdateConfig,
+    pub(super) restart_controller: DaemonRestartController,
+}
+
+pub(super) type DaemonRestartController = fn(&Path) -> anyhow::Result<u32>;
+
+/// 任务 worker：单线程 FIFO 消费任务队列，串行执行官方同步/校验。
+///
+/// 每个任务执行前获取进程内 `sync_lock`，与 watch 循环互斥（等待而非撞文件锁失败）；
+/// 进度写入任务记录；`should_cancel` 接 daemon 停止标志，停机时中止在途任务。
+pub(super) fn run_task_worker(
+    receiver: mpsc::Receiver<TaskJob>,
+    registry: TaskRegistry,
+    sync_lock: Arc<Mutex<()>>,
+    control: DaemonControl,
+) {
+    let service = OfficialUpdateService::new();
+    for job in receiver {
+        registry.update(&job.id, |record| {
+            record.status = "running";
+            record.started_at = Some(unix_seconds_now());
+        });
+
+        let cancel = Arc::clone(&job.cancel);
+        let run_result = {
+            let _sync_guard = sync_lock
+                .lock()
+                .unwrap_or_else(|poison| poison.into_inner());
+            let progress_registry = registry.clone();
+            let progress_id = job.id.clone();
+            let cancel_check = Arc::clone(&cancel);
+            let stop_control = Arc::clone(&control);
+            service.run_with_progress_and_cancellation(
+                &job.config,
+                |event| {
+                    progress_registry
+                        .append_log(&progress_id, format!("[{}] {}", event.stage, event.message));
+                    progress_registry.update(&progress_id, |record| {
+                        record.stage = Some(event.stage.to_string());
+                        record.message = Some(event.message.clone());
+                    });
+                },
+                || {
+                    cancel_check.load(Ordering::Relaxed)
+                        || daemon_control_stop_requested(Some(&stop_control))
+                },
+            )
+        };
+
+        match run_result {
+            Ok(report) => registry.update(&job.id, |record| {
+                record.status = "succeeded";
+                record.finished_at = Some(unix_seconds_now());
+                record.result = serde_json::to_value(&report).ok();
+            }),
+            Err(error) => {
+                let cancelled = cancel.load(Ordering::Relaxed);
+                // 下载失败携带类型化 DownloadError（含准确网络域码）；其余归 internal。
+                let code = error
+                    .downcast_ref::<bat_infrastructure::DownloadError>()
+                    .map(bat_infrastructure::DownloadError::code)
+                    .unwrap_or(ErrorCode::INTERNAL);
+                registry.update(&job.id, |record| {
+                    record.finished_at = Some(unix_seconds_now());
+                    if cancelled {
+                        record.status = "cancelled";
+                        record.error = Some(ApiError::new(
+                            ErrorCode::INTERNAL,
+                            "task.executor",
+                            "任务已取消",
+                        ));
+                    } else {
+                        record.status = "failed";
+                        record.error =
+                            Some(ApiError::new(code, "task.executor", error.to_string()));
+                    }
+                });
+            }
+        }
+    }
+}