From 9a5fef23afdaa41a1be63b959d613c2e16ddd340 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Fri, 17 Jul 2026 06:00:26 -0700 Subject: [PATCH] =?UTF-8?q?feat(daemon):=20=E8=A1=A5=E5=85=A8=20task.cance?= =?UTF-8?q?l=20=E4=B8=8E=20task.logs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 完成 task.* 命名空间： - task.cancel：每个任务带共享的取消标志（Arc），worker 的 should_cancel 同时检查它和 daemon 停止标志；请求取消后置标志，worker 在下一个同步检查点中止， 任务转为 cancelled（协作式，不硬杀正在执行的 curl）。已结束任务返回 already-finished， 不存在返回 700004。 - task.logs：worker 把每条 progress 追加到任务的有界日志（上限 200 行），task.logs 返回该历史；不存在返回 700004。TaskRecord 的 cancel/log 字段 serde(skip)，不进 task.status 输出。 - 新增 cancelled 任务状态；prune 与已结束判定纳入 cancelled。 - USERGUIDE RPC 方法表更新（cancel/logs 标为已实现）。 验证：新增取消/日志注册表单测（63 bin 测试全过）；真机端到端确认触发 resource.sync 后 task.cancel 使任务转为 cancelled、task.logs 返回进度历史。 对应 issue #1（task.* 命名空间补全）。 Co-Authored-By: Claude Fable 5 --- USERGUIDE.md | 5 +- infrastructure/src/bin/bat_official_sync.rs | 216 ++++++++++++++++++-- 2 files changed, 199 insertions(+), 22 deletions(-) diff --git a/USERGUIDE.md b/USERGUIDE.md index 1c945e5..b5d3f17 100644 --- a/USERGUIDE.md +++ b/USERGUIDE.md @@ -262,7 +262,9 @@ HTTPS_PROXY=http://user:pass@127.0.0.1:7890 bat --auto-discover --daemon | `resource.verify` | ✅ | 触发校验任务（dry-run + audit），返回 `task_id` | | `task.status` | ✅ | 查询任务（`params.task_id`） | | `task.list` | ✅ | 列出全部任务（最新在前） | -| `catalog.*` / `patch.*` / `unityfs.*` / `task.cancel` / `task.logs` | ⏳ | 已规划，返回 `BAT-ERR-700003`（not implemented） | +| `task.cancel` | ✅ | 请求取消任务（`params.task_id`）；协作式，在同步检查点生效 | +| `task.logs` | ✅ | 返回任务的进度日志（`params.task_id`，有界） | +| `catalog.*` / `patch.*` / `unityfs.*` / `task.create` | ⏳ | 已规划，返回 `BAT-ERR-700003`（not implemented） | | 未知方法 | — | `BAT-ERR-700001`（unknown method） | ### 任务模型 @@ -277,6 +279,7 @@ HTTPS_PROXY=http://user:pass@127.0.0.1:7890 bat --auto-discover --daemon "error": { … }, "result": { … } } ``` +- `task.cancel` 请求取消：置任务的取消标志，worker 在下一个同步检查点中止，任务转为 `cancelled`（协作式，不硬杀正在执行的 curl）。 - 与后台 watch 循环的定时同步通过进程内锁互斥（任务等待而非失败）。 - 任务目前**仅存内存**，随 daemon 生死；持久化（重启后仍可查）为后续工作。 diff --git a/infrastructure/src/bin/bat_official_sync.rs b/infrastructure/src/bin/bat_official_sync.rs index 8320281..63d7a1e 100644 --- a/infrastructure/src/bin/bat_official_sync.rs +++ b/infrastructure/src/bin/bat_official_sync.rs @@ -19,6 +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::atomic::{AtomicBool, Ordering}; use std::sync::{mpsc, Arc, Condvar, Mutex}; use std::thread; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; @@ -639,9 +640,13 @@ 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 RPC_METHOD_TASK_CANCEL: &str = "task.cancel"; +const RPC_METHOD_TASK_LOGS: &str = "task.logs"; /// 保留的已完成任务上限（内存态，超出后裁剪最旧的已结束任务）。 const MAX_RETAINED_TASKS: usize = 64; +/// 每个任务保留的进度日志行数上限。 +const MAX_TASK_LOG_LINES: usize = 200; /// 任务类型：目前覆盖官方同步与校验。 #[derive(Debug, Clone, Copy)] @@ -678,12 +683,19 @@ impl TaskKind { } } +/// 请求取消任务的结果。 +enum CancelOutcome { + Requested, + AlreadyFinished, + NotFound, +} + /// 单个任务的可轮询记录。 #[derive(Debug, Clone, Serialize)] struct TaskRecord { id: String, kind: &'static str, - /// `queued` | `running` | `succeeded` | `failed`。 + /// `queued` | `running` | `succeeded` | `failed` | `cancelled`。 status: &'static str, #[serde(skip_serializing_if = "Option::is_none")] stage: Option, @@ -699,6 +711,18 @@ struct TaskRecord { error: Option, #[serde(skip_serializing_if = "Option::is_none")] result: Option, + /// 取消标志，worker 的 should_cancel 检查它；不参与序列化。 + #[serde(skip)] + cancel: Arc, + /// 进度日志（有界），经 task.logs 返回；不参与 task.status 序列化。 + #[serde(skip)] + log: Vec, +} + +impl TaskRecord { + fn is_finished(&self) -> bool { + matches!(self.status, "succeeded" | "failed" | "cancelled") + } } struct TaskStore { @@ -750,6 +774,8 @@ impl TaskRegistry { 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()); @@ -769,6 +795,44 @@ impl TaskRegistry { self.lock().tasks.get(id).cloned() } + /// 返回任务的取消标志（与 worker 共享同一 Arc）。 + fn cancel_flag(&self, id: &str) -> Option> { + self.lock() + .tasks + .get(id) + .map(|record| Arc::clone(&record.cancel)) + } + + /// 追加一行进度日志，超出上限时丢弃最旧的。 + 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); + } + } + } + + /// 返回任务的进度日志。 + fn logs(&self, id: &str) -> Option> { + self.lock().tasks.get(id).map(|record| record.log.clone()) + } + + /// 请求取消任务：未结束的置取消标志，已结束的原样返回，不存在返回 NotFound。 + 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 + } + } + } + /// 返回全部任务，最新创建的在前。 fn list(&self) -> Vec { let store = self.lock(); @@ -788,7 +852,7 @@ impl TaskStore { let Some(position) = self.order.iter().position(|id| { self.tasks .get(id) - .map(|record| matches!(record.status, "succeeded" | "failed")) + .map(TaskRecord::is_finished) .unwrap_or(true) }) else { break; @@ -803,6 +867,8 @@ impl TaskStore { struct TaskJob { id: String, config: OfficialUpdateConfig, + /// 与任务记录共享的取消标志。 + cancel: Arc, } /// daemon 任务上下文：RPC handler 借它创建任务、入队和读取。 @@ -830,21 +896,29 @@ fn run_task_worker( 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()); }); }, - || daemon_control_stop_requested(Some(&control)), + || { + cancel_check.load(Ordering::Relaxed) + || daemon_control_stop_requested(Some(&stop_control)) + }, ) }; @@ -854,15 +928,27 @@ fn run_task_worker( 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(), - )); - }), + Err(error) => { + let cancelled = cancel.load(Ordering::Relaxed); + 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( + ErrorCode::INTERNAL, + "task.executor", + error.to_string(), + )); + } + }); + } } } } @@ -881,7 +967,7 @@ fn canonical_rpc_method(method: &str) -> &str { /// 判断方法是否属于已规划但尚未实现的命名空间/动作（返回 not_implemented 而非 unknown）。 fn is_pending_rpc_method(method: &str) -> bool { - matches!(method, "task.create" | "task.cancel" | "task.logs") + matches!(method, "task.create") || method.starts_with("catalog.") || method.starts_with("patch.") || method.starts_with("unityfs.") @@ -1516,12 +1602,7 @@ fn dispatch_rpc_method( enqueue_task_envelope(tasks, TaskKind::Verify, false, request_id) } RPC_METHOD_TASK_STATUS => { - let task_id = request - .params - .as_ref() - .and_then(|params| params.get("task_id")) - .and_then(serde_json::Value::as_str) - .unwrap_or_default(); + let task_id = rpc_task_id_param(request); match tasks.registry.get(task_id) { Some(record) => rpc_envelope_from_result( request_id, @@ -1544,6 +1625,48 @@ fn dispatch_rpc_method( serde_json::to_value(serde_json::json!({ "tasks": tasks.registry.list() })) .map_err(anyhow::Error::from), ), + RPC_METHOD_TASK_CANCEL => { + let task_id = rpc_task_id_param(request); + match tasks.registry.request_cancel(task_id) { + CancelOutcome::Requested => rpc_envelope_ok( + request_id, + "accepted", + serde_json::json!({ "task_id": task_id, "cancel_requested": true }), + ), + CancelOutcome::AlreadyFinished => rpc_envelope_ok( + request_id, + "ok", + serde_json::json!({ "task_id": task_id, "cancel_requested": false, "note": "任务已结束" }), + ), + CancelOutcome::NotFound => rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::TASK_NOT_FOUND, + "task.cancel", + format!("任务不存在：{task_id}"), + ), + ), + } + } + RPC_METHOD_TASK_LOGS => { + let task_id = rpc_task_id_param(request); + match tasks.registry.logs(task_id) { + Some(lines) => rpc_envelope_from_result( + request_id, + "task.logs", + serde_json::to_value(serde_json::json!({ "task_id": task_id, "lines": lines })) + .map_err(anyhow::Error::from), + ), + None => rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::TASK_NOT_FOUND, + "task.logs", + format!("任务不存在：{task_id}"), + ), + ), + } + } pending if is_pending_rpc_method(pending) => rpc_envelope_error( request_id, ApiError::new( @@ -1572,9 +1695,14 @@ fn enqueue_task_envelope( ) -> RpcEnvelope { let config = kind.build_config(&tasks.base_config, force); let task_id = tasks.registry.create(kind); + let cancel = tasks + .registry + .cancel_flag(&task_id) + .unwrap_or_else(|| Arc::new(AtomicBool::new(false))); let job = TaskJob { id: task_id.clone(), config, + cancel, }; if tasks.queue.send(job).is_err() { // worker 已退出：把该任务标记为失败并返回错误。 @@ -1667,6 +1795,15 @@ fn json_rpc_error(id: Option, code: i32, message: String) -> } } +fn rpc_task_id_param(request: &JsonRpcRequest) -> &str { + request + .params + .as_ref() + .and_then(|params| params.get("task_id")) + .and_then(serde_json::Value::as_str) + .unwrap_or_default() +} + fn rpc_bool_param(params: Option<&serde_json::Value>, key: &str) -> Option { params .and_then(|params| params.get(key)) @@ -5220,13 +5357,50 @@ mod tests { 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("task.create")); + // sync/verify 与 task.cancel/logs 已实现，不再是 pending。 assert!(!is_pending_rpc_method("resource.sync")); assert!(!is_pending_rpc_method("resource.verify")); + assert!(!is_pending_rpc_method("task.cancel")); + assert!(!is_pending_rpc_method("task.logs")); assert!(!is_pending_rpc_method("daemon.status")); } + #[test] + fn task_registry_cancel_and_logs() { + let registry = TaskRegistry::new(); + let id = registry.create(TaskKind::Sync); + + // 取消未结束任务：置标志。 + assert!(matches!( + registry.request_cancel(&id), + CancelOutcome::Requested + )); + assert!(registry.cancel_flag(&id).unwrap().load(Ordering::Relaxed)); + + // 追加日志（超上限时丢弃最旧）。 + for i in 0..(MAX_TASK_LOG_LINES + 5) { + registry.append_log(&id, format!("line {i}")); + } + let logs = registry.logs(&id).unwrap(); + assert_eq!(logs.len(), MAX_TASK_LOG_LINES); + assert_eq!( + logs.last().unwrap(), + &format!("line {}", MAX_TASK_LOG_LINES + 4) + ); + + // 已结束任务：AlreadyFinished；不存在：NotFound。 + registry.update(&id, |record| record.status = "succeeded"); + assert!(matches!( + registry.request_cancel(&id), + CancelOutcome::AlreadyFinished + )); + assert!(matches!( + registry.request_cancel("nope"), + CancelOutcome::NotFound + )); + } + fn test_task_context() -> DaemonTaskContext { let (queue, _rx) = mpsc::channel::(); DaemonTaskContext {