From 01ea5d8fa59df23c388a442b5d24717f9b3dd2e0 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Fri, 17 Jul 2026 05:23:24 -0700 Subject: [PATCH] =?UTF-8?q?feat(daemon):=20=E4=BB=BB=E5=8A=A1=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E5=99=A8=20+=20=E5=BC=82=E6=AD=A5=20resource.sync/ver?= =?UTF-8?q?ify=EF=BC=88=E9=98=B6=E6=AE=B5=201b=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在 daemon 里加入任务模型，让 resource.sync/verify 经 RPC 触发即返回 task_id、 可轮询进度与最终结果： - 专用任务 worker 线程 + FIFO 队列（mpsc）；TaskRegistry 句柄包住内存态注册表 （HashMap，有界保留 64 条已结束任务，运行中/排队中不裁剪），方法访问便于将来 换 Redis 等持久化后端。 - 进程内 sync 锁：watch 循环与 worker 在跑同步前都获取它，互相等待而非撞 .official-sync.lock 文件锁失败；只在实际跑同步的那段持锁，空闲睡眠不持锁。 - resource.sync/resource.verify 入队返回 { task_id, kind }（envelope status=accepted）； worker 复用 run_with_progress_and_cancellation（sync 用 daemon config + force； verify 用 dry-run+plan+audit），进度写入任务记录，should_cancel 接 daemon 停止标志 （停机中止在途任务）。task.status/task.list 从注册表读真实数据。 验证：新增任务分发/入队/注册表单测（62 bin 测试全过）；真机端到端确认经裸 socket 触发 resource.sync 返回 task_id、轮询 task.status 走到最终态（fake curl 失败→ failed 且记录 error）、task.list 可见、resource.verify 亦入队。满足 issue #1 验收 「至少一个同步任务可经 RPC 触发并经 task.status 查询进度和最终结果」。 任务仅存内存（随 daemon 生死）；task.cancel/logs、错误码细分映射留待后续。 对应 issue #1（任务模型 + 最小方法集打通）。 Co-Authored-By: Claude Fable 5 --- infrastructure/src/bin/bat_official_sync.rs | 485 ++++++++++++++++++-- 1 file changed, 443 insertions(+), 42 deletions(-) diff --git a/infrastructure/src/bin/bat_official_sync.rs b/infrastructure/src/bin/bat_official_sync.rs index ef032ef..8320281 100644 --- a/infrastructure/src/bin/bat_official_sync.rs +++ b/infrastructure/src/bin/bat_official_sync.rs @@ -9,6 +9,7 @@ use bat_infrastructure::{ OfficialVerificationSummary, OfficialVersionRecord, OfficialVersionState, PRIVATE_FILE_MODE, }; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; use std::env; use std::fs::{self, OpenOptions}; use std::io::{BufRead, BufReader, Write}; @@ -18,7 +19,7 @@ use std::os::unix::fs::OpenOptionsExt; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; -use std::sync::{Arc, Condvar, Mutex}; +use std::sync::{mpsc, Arc, Condvar, Mutex}; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -267,13 +268,30 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { } else { None }; - let _rpc_server = if let Some(control) = daemon_control.as_ref() { - Some(start_daemon_rpc_server( - &daemon_state_dir, - Arc::clone(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 registry = TaskRegistry::new(); + let (task_tx, task_rx) = mpsc::channel::(); + let worker = { + let registry = registry.clone(); + let sync_lock = Arc::clone(&sync_lock); + let control = Arc::clone(control); + thread::Builder::new() + .name("bat-daemon-task-worker".to_string()) + .spawn(move || run_task_worker(task_rx, registry, sync_lock, control))? + }; + let context = DaemonTaskContext { + registry, + queue: task_tx, + base_config: options.config.clone(), + }; + let server = + start_daemon_rpc_server(&daemon_state_dir, Arc::clone(control), context.clone())?; + (Some(worker), Some(context), Some(server)) } else { - None + (None, None, None) }; let mut next_forced_refresh_at = next_forced_refresh_at_or_after(SystemTime::now()); let mut pending_scheduled_force = false; @@ -339,19 +357,26 @@ fn run_watch(options: CliOptions) -> anyhow::Result<()> { next_forced_refresh_at, }, ); - match service.run_with_progress_and_cancellation( - &iteration_config, - |event| { - record_daemon_progress( - options.daemon_child, - &daemon_state_dir, - &mut logger, - &event, - ); - logger.log(event); - }, - || daemon_control_stop_requested(daemon_control.as_ref()), - ) { + // 只在实际执行同步的这段持有 sync_lock，与任务 worker 互斥；空闲睡眠时不持锁。 + let run_result = { + let _sync_guard = sync_lock + .lock() + .unwrap_or_else(|poison| poison.into_inner()); + service.run_with_progress_and_cancellation( + &iteration_config, + |event| { + record_daemon_progress( + options.daemon_child, + &daemon_state_dir, + &mut logger, + &event, + ); + logger.log(event); + }, + || daemon_control_stop_requested(daemon_control.as_ref()), + ) + }; + match run_result { Ok(report) => { if pending_scheduled_force { pending_scheduled_force = false; @@ -610,9 +635,238 @@ const RPC_METHOD_RELOAD: &str = "daemon.reload"; const RPC_METHOD_REFRESH: &str = "daemon.refresh"; const RPC_METHOD_LOGS: &str = "daemon.logs"; const RPC_METHOD_RESOURCE_STATE: &str = "resource.state"; +const RPC_METHOD_RESOURCE_SYNC: &str = "resource.sync"; +const RPC_METHOD_RESOURCE_VERIFY: &str = "resource.verify"; const RPC_METHOD_TASK_STATUS: &str = "task.status"; const RPC_METHOD_TASK_LIST: &str = "task.list"; +/// 保留的已完成任务上限（内存态，超出后裁剪最旧的已结束任务）。 +const MAX_RETAINED_TASKS: usize = 64; + +/// 任务类型：目前覆盖官方同步与校验。 +#[derive(Debug, Clone, Copy)] +enum TaskKind { + Sync, + Verify, +} + +impl TaskKind { + fn method(self) -> &'static str { + match self { + Self::Sync => RPC_METHOD_RESOURCE_SYNC, + Self::Verify => RPC_METHOD_RESOURCE_VERIFY, + } + } + + /// 由 daemon 基准配置派生该任务的实际同步配置。 + 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; + } + } + config + } +} + +/// 单个任务的可轮询记录。 +#[derive(Debug, Clone, Serialize)] +struct TaskRecord { + id: String, + kind: &'static str, + /// `queued` | `running` | `succeeded` | `failed`。 + status: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + stage: Option, + #[serde(skip_serializing_if = "Option::is_none")] + message: Option, + created_at: u64, + updated_at: u64, + #[serde(skip_serializing_if = "Option::is_none")] + started_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + finished_at: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + #[serde(skip_serializing_if = "Option::is_none")] + result: Option, +} + +struct TaskStore { + tasks: HashMap, + order: Vec, + seq: u64, +} + +/// 任务注册表句柄：包住内存存储，供 RPC handler 与 worker 共享。 +/// +/// 通过方法访问（而非直接摸内部 map），便于将来换成 Redis 等持久化后端。 +#[derive(Clone)] +struct TaskRegistry { + inner: Arc>, +} + +impl TaskRegistry { + fn new() -> Self { + Self { + inner: Arc::new(Mutex::new(TaskStore { + tasks: HashMap::new(), + order: Vec::new(), + seq: 0, + })), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, TaskStore> { + self.inner + .lock() + .unwrap_or_else(|poison| poison.into_inner()) + } + + /// 创建 queued 任务并返回 task_id。 + 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, + }; + store.tasks.insert(id.clone(), record); + store.order.push(id.clone()); + store.prune(); + id + } + + fn update(&self, id: &str, update: F) { + let mut store = self.lock(); + if let Some(record) = store.tasks.get_mut(id) { + update(record); + record.updated_at = unix_seconds_now(); + } + } + + fn get(&self, id: &str) -> Option { + self.lock().tasks.get(id).cloned() + } + + /// 返回全部任务，最新创建的在前。 + fn list(&self) -> Vec { + let store = self.lock(); + store + .order + .iter() + .rev() + .filter_map(|id| store.tasks.get(id).cloned()) + .collect() + } +} + +impl TaskStore { + /// 裁剪最旧的已结束任务，把内存占用控制在上限内；运行中/排队中的任务不裁剪。 + fn prune(&mut self) { + while self.order.len() > MAX_RETAINED_TASKS { + let Some(position) = self.order.iter().position(|id| { + self.tasks + .get(id) + .map(|record| matches!(record.status, "succeeded" | "failed")) + .unwrap_or(true) + }) else { + break; + }; + let id = self.order.remove(position); + self.tasks.remove(&id); + } + } +} + +/// 提交给任务 worker 的作业（配置已按任务类型派生完毕）。 +struct TaskJob { + id: String, + config: OfficialUpdateConfig, +} + +/// daemon 任务上下文：RPC handler 借它创建任务、入队和读取。 +#[derive(Clone)] +struct DaemonTaskContext { + registry: TaskRegistry, + queue: mpsc::Sender, + base_config: OfficialUpdateConfig, +} + +/// 任务 worker：单线程 FIFO 消费任务队列，串行执行官方同步/校验。 +/// +/// 每个任务执行前获取进程内 `sync_lock`，与 watch 循环互斥（等待而非撞文件锁失败）； +/// 进度写入任务记录；`should_cancel` 接 daemon 停止标志，停机时中止在途任务。 +fn run_task_worker( + receiver: mpsc::Receiver, + registry: TaskRegistry, + sync_lock: Arc>, + 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 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(); + service.run_with_progress_and_cancellation( + &job.config, + |event| { + progress_registry.update(&progress_id, |record| { + record.stage = Some(event.stage.to_string()); + record.message = Some(event.message.clone()); + }); + }, + || daemon_control_stop_requested(Some(&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) => registry.update(&job.id, |record| { + record.status = "failed"; + record.finished_at = Some(unix_seconds_now()); + record.error = Some(ApiError::new( + ErrorCode::INTERNAL, + "task.executor", + error.to_string(), + )); + }), + } + } +} + /// 把 `bat.*` 兼容别名解析为规范的 `daemon.*` 方法名；其余原样返回。 fn canonical_rpc_method(method: &str) -> &str { match method { @@ -627,10 +881,8 @@ fn canonical_rpc_method(method: &str) -> &str { /// 判断方法是否属于已规划但尚未实现的命名空间/动作（返回 not_implemented 而非 unknown）。 fn is_pending_rpc_method(method: &str) -> bool { - matches!( - method, - "resource.sync" | "resource.verify" | "task.create" | "task.cancel" | "task.logs" - ) || method.starts_with("catalog.") + matches!(method, "task.create" | "task.cancel" | "task.logs") + || method.starts_with("catalog.") || method.starts_with("patch.") || method.starts_with("unityfs.") } @@ -1058,6 +1310,7 @@ fn take_daemon_wake(state: &mut DaemonControlState) -> Option { fn start_daemon_rpc_server( state_dir: &Path, control: DaemonControl, + tasks: DaemonTaskContext, ) -> anyhow::Result { validate_runtime_state_dir(state_dir).map_err(anyhow::Error::msg)?; fs::create_dir_all(state_dir)?; @@ -1087,9 +1340,12 @@ fn start_daemon_rpc_server( Ok(stream) => { let state_dir = server_state_dir.clone(); let control = Arc::clone(&control); + let tasks = tasks.clone(); let _ = thread::Builder::new() .name("bat-daemon-rpc-client".to_string()) - .spawn(move || handle_daemon_rpc_client(stream, state_dir, control)); + .spawn(move || { + handle_daemon_rpc_client(stream, state_dir, control, tasks) + }); } Err(error) => { eprintln!("[daemon] RPC socket accept 失败：{error}"); @@ -1106,12 +1362,18 @@ fn start_daemon_rpc_server( fn start_daemon_rpc_server( _state_dir: &Path, _control: DaemonControl, + _tasks: DaemonTaskContext, ) -> anyhow::Result { Err(anyhow::anyhow!("daemon RPC 目前只支持 Unix/Linux 平台")) } #[cfg(unix)] -fn handle_daemon_rpc_client(mut stream: UnixStream, state_dir: PathBuf, control: DaemonControl) { +fn handle_daemon_rpc_client( + mut stream: UnixStream, + state_dir: PathBuf, + control: DaemonControl, + tasks: DaemonTaskContext, +) { let Ok(reader_stream) = stream.try_clone() else { return; }; @@ -1137,7 +1399,7 @@ fn handle_daemon_rpc_client(mut stream: UnixStream, state_dir: PathBuf, control: let response = match serde_json::from_str::(&line) { Ok(request) => { notify_stop_after_response = request.method == RPC_METHOD_STOP; - handle_daemon_rpc_request(request, &state_dir, &control) + handle_daemon_rpc_request(request, &state_dir, &control, &tasks) } Err(error) => json_rpc_error(None, -32700, format!("JSON-RPC 请求解析失败：{error}")), }; @@ -1158,10 +1420,11 @@ fn handle_daemon_rpc_request( request: JsonRpcRequest, state_dir: &Path, control: &DaemonControl, + tasks: &DaemonTaskContext, ) -> JsonRpcResponse { let id = request.id.clone(); let request_id = next_request_id(); - let envelope = dispatch_rpc_method(&request, state_dir, control, request_id); + let envelope = dispatch_rpc_method(&request, state_dir, control, tasks, request_id); match serde_json::to_value(&envelope) { Ok(value) => json_rpc_result(id, value), Err(error) => json_rpc_error(id, -32603, error.to_string()), @@ -1173,6 +1436,7 @@ fn dispatch_rpc_method( request: &JsonRpcRequest, state_dir: &Path, control: &DaemonControl, + tasks: &DaemonTaskContext, request_id: String, ) -> RpcEnvelope { match canonical_rpc_method(&request.method) { @@ -1244,6 +1508,13 @@ fn dispatch_rpc_method( "resource.state", build_resource_state_report(state_dir), ), + RPC_METHOD_RESOURCE_SYNC => { + let force = rpc_bool_param(request.params.as_ref(), "force").unwrap_or(false); + enqueue_task_envelope(tasks, TaskKind::Sync, force, request_id) + } + RPC_METHOD_RESOURCE_VERIFY => { + enqueue_task_envelope(tasks, TaskKind::Verify, false, request_id) + } RPC_METHOD_TASK_STATUS => { let task_id = request .params @@ -1251,19 +1522,28 @@ fn dispatch_rpc_method( .and_then(|params| params.get("task_id")) .and_then(serde_json::Value::as_str) .unwrap_or_default(); - // 任务执行器尚未落地（阶段 1b）；任务库为空，任何 task_id 都不存在。 - rpc_envelope_error( - request_id, - ApiError::new( - ErrorCode::TASK_NOT_FOUND, + match tasks.registry.get(task_id) { + Some(record) => rpc_envelope_from_result( + request_id, "task.status", - format!("任务不存在：{task_id}"), + serde_json::to_value(record).map_err(anyhow::Error::from), ), - ) - } - RPC_METHOD_TASK_LIST => { - rpc_envelope_ok(request_id, "ok", serde_json::json!({ "tasks": [] })) + None => rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::TASK_NOT_FOUND, + "task.status", + format!("任务不存在：{task_id}"), + ), + ), + } } + RPC_METHOD_TASK_LIST => rpc_envelope_from_result( + request_id, + "task.list", + serde_json::to_value(serde_json::json!({ "tasks": tasks.registry.list() })) + .map_err(anyhow::Error::from), + ), pending if is_pending_rpc_method(pending) => rpc_envelope_error( request_id, ApiError::new( @@ -1283,6 +1563,42 @@ fn dispatch_rpc_method( } } +/// 创建任务、入队，返回 `accepted` + task_id 的 envelope。 +fn enqueue_task_envelope( + tasks: &DaemonTaskContext, + kind: TaskKind, + force: bool, + request_id: String, +) -> RpcEnvelope { + let config = kind.build_config(&tasks.base_config, force); + let task_id = tasks.registry.create(kind); + let job = TaskJob { + id: task_id.clone(), + config, + }; + if tasks.queue.send(job).is_err() { + // worker 已退出：把该任务标记为失败并返回错误。 + tasks.registry.update(&task_id, |record| { + record.status = "failed"; + record.finished_at = Some(unix_seconds_now()); + record.error = Some(ApiError::new( + ErrorCode::INTERNAL, + "task.enqueue", + "任务执行器不可用", + )); + }); + return rpc_envelope_error( + request_id, + ApiError::new(ErrorCode::INTERNAL, "task.enqueue", "任务执行器不可用"), + ); + } + rpc_envelope_ok( + request_id, + "accepted", + serde_json::json!({ "task_id": task_id, "kind": kind.method() }), + ) +} + fn rpc_ack_value( command: &'static str, message: &'static str, @@ -4901,13 +5217,23 @@ mod tests { #[test] fn is_pending_rpc_method_covers_planned_namespaces() { - assert!(is_pending_rpc_method("resource.sync")); - assert!(is_pending_rpc_method("resource.verify")); assert!(is_pending_rpc_method("catalog.status")); assert!(is_pending_rpc_method("patch.apply")); assert!(is_pending_rpc_method("unityfs.inspect")); + assert!(is_pending_rpc_method("task.cancel")); + // sync/verify 已由任务执行器实现，不再是 pending。 + assert!(!is_pending_rpc_method("resource.sync")); + assert!(!is_pending_rpc_method("resource.verify")); assert!(!is_pending_rpc_method("daemon.status")); - assert!(!is_pending_rpc_method("resource.state")); + } + + fn test_task_context() -> DaemonTaskContext { + let (queue, _rx) = mpsc::channel::(); + DaemonTaskContext { + registry: TaskRegistry::new(), + queue, + base_config: OfficialUpdateConfig::default(), + } } #[test] @@ -4918,6 +5244,7 @@ mod tests { &rpc_request("nope.nope", None), temp.path(), &control, + &test_task_context(), "req-test-1".to_string(), ); let value = serde_json::to_value(&envelope).unwrap(); @@ -4931,9 +5258,10 @@ mod tests { let temp = tempfile::TempDir::new().unwrap(); let control = new_daemon_control(); let envelope = dispatch_rpc_method( - &rpc_request("resource.sync", None), + &rpc_request("catalog.status", None), temp.path(), &control, + &test_task_context(), "req-test-2".to_string(), ); let value = serde_json::to_value(&envelope).unwrap(); @@ -4949,6 +5277,7 @@ mod tests { &rpc_request("task.list", None), temp.path(), &control, + &test_task_context(), "req-test-3".to_string(), ); let value = serde_json::to_value(&envelope).unwrap(); @@ -4965,6 +5294,7 @@ mod tests { &rpc_request("task.status", Some(serde_json::json!({ "task_id": "abc" }))), temp.path(), &control, + &test_task_context(), "req-test-4".to_string(), ); let value = serde_json::to_value(&envelope).unwrap(); @@ -4972,6 +5302,77 @@ mod tests { assert_eq!(value["error"]["code"], "BAT-ERR-700004"); } + #[test] + fn dispatch_resource_sync_enqueues_task() { + let temp = tempfile::TempDir::new().unwrap(); + let control = new_daemon_control(); + // 保留 rx 让 send 成功（不启动 worker，任务停留在 queued）。 + let (queue, rx) = mpsc::channel::(); + let context = DaemonTaskContext { + registry: TaskRegistry::new(), + queue, + base_config: OfficialUpdateConfig::default(), + }; + + let envelope = dispatch_rpc_method( + &rpc_request("resource.sync", Some(serde_json::json!({ "force": true }))), + temp.path(), + &control, + &context, + "req-test-5".to_string(), + ); + let value = serde_json::to_value(&envelope).unwrap(); + assert_eq!(value["ok"], true); + assert_eq!(value["status"], "accepted"); + let task_id = value["data"]["task_id"].as_str().unwrap().to_string(); + assert_eq!(value["data"]["kind"], "resource.sync"); + + // 注册表里应有一条 queued 任务。 + let record = context.registry.get(&task_id).unwrap(); + assert_eq!(record.status, "queued"); + assert_eq!(record.kind, "resource.sync"); + + // 作业已入队，force 已透传到派生配置。 + let job = rx.try_recv().unwrap(); + assert_eq!(job.id, task_id); + assert!(job.config.force); + assert!(!job.config.dry_run); + + // task.status 能查到这条任务。 + let status_envelope = dispatch_rpc_method( + &rpc_request( + "task.status", + Some(serde_json::json!({ "task_id": task_id })), + ), + temp.path(), + &control, + &context, + "req-test-6".to_string(), + ); + let status_value = serde_json::to_value(&status_envelope).unwrap(); + assert_eq!(status_value["ok"], true); + assert_eq!(status_value["data"]["status"], "queued"); + } + + #[test] + fn task_registry_create_update_and_prune() { + let registry = TaskRegistry::new(); + let id = registry.create(TaskKind::Verify); + assert_eq!(registry.get(&id).unwrap().status, "queued"); + registry.update(&id, |record| record.status = "running"); + assert_eq!(registry.get(&id).unwrap().status, "running"); + assert_eq!(registry.list().len(), 1); + + // 超过上限时裁剪最旧的已结束任务；未结束的保留。 + for _ in 0..MAX_RETAINED_TASKS { + let finished = registry.create(TaskKind::Sync); + registry.update(&finished, |record| record.status = "succeeded"); + } + assert!(registry.list().len() <= MAX_RETAINED_TASKS); + // 运行中的任务不被裁剪。 + assert_eq!(registry.get(&id).unwrap().status, "running"); + } + #[test] fn parses_explicit_watch_error_retry_interval() { let options =