feat(daemon): 补全 task.cancel 与 task.logs

完成 task.* 命名空间:

- task.cancel:每个任务带共享的取消标志(Arc<AtomicBool>),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 <noreply@anthropic.com>
This commit is contained in:
2026-07-17 06:00:26 -07:00
co-authored by Claude Fable 5
parent d7257db447
commit 9a5fef23af
2 changed files with 199 additions and 22 deletions
+4 -1
View File
@@ -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 生死;持久化(重启后仍可查)为后续工作。
+189 -15
View File
@@ -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<String>,
@@ -699,6 +711,18 @@ struct TaskRecord {
error: Option<ApiError>,
#[serde(skip_serializing_if = "Option::is_none")]
result: Option<serde_json::Value>,
/// 取消标志,worker 的 should_cancel 检查它;不参与序列化。
#[serde(skip)]
cancel: Arc<AtomicBool>,
/// 进度日志(有界),经 task.logs 返回;不参与 task.status 序列化。
#[serde(skip)]
log: Vec<String>,
}
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<Arc<AtomicBool>> {
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<Vec<String>> {
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<TaskRecord> {
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<AtomicBool>,
}
/// 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";
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<serde_json::Value>, 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<bool> {
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::<TaskJob>();
DaemonTaskContext {