From b95a3ac417d3aa076fccb639e522f32bf0e20451 Mon Sep 17 00:00:00 2001 From: Yuyi-Oak <1722157266@qq.com> Date: Fri, 17 Jul 2026 04:19:22 -0700 Subject: [PATCH] =?UTF-8?q?feat(daemon):=20RPC=20=E7=BB=9F=E4=B8=80=20enve?= =?UTF-8?q?lope=E3=80=81=E5=91=BD=E5=90=8D=E7=A9=BA=E9=97=B4=E5=88=86?= =?UTF-8?q?=E5=8F=91=E4=B8=8E=E5=8F=AA=E8=AF=BB=E6=96=B9=E6=B3=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 阶段 1a 的协议地基（承接错误码模型）： - 统一 envelope：所有响应装入 JSON-RPC 2.0 result 的 { ok, status, data, error, request_id }；error 用 bat_core::ApiError （含 BAT-ERR 码/kind/domain/location/retryable）。传输层解析失败仍走 JSON-RPC 顶层 error。request_id 为 req--。 - 命名空间分发：方法改用国际惯例 .；daemon.* 为规范名， bat.* 保留为兼容别名（canonical_rpc_method 解析）。未知方法 → 700001， 已规划未实现（resource.sync/verify、catalog.*/patch.*/unityfs.* 等）→ 700003。 - 只读方法：daemon.status/logs/stop/reload/refresh（迁移）、resource.state（新增， 返回资源发布根 + 版本状态 + 上次同步结果）、task.status/task.list（任务库暂空， status → 700004，list → 空列表；执行器留待阶段 1b）。 - 客户端 daemon_rpc_call 解包 envelope（检查 ok、返回 data、把 error 转 Err）， CLI 调用方零改动。 验证：新增分发/别名/envelope 单测（60 bin 测试全过）；真机起停 daemon 端到端确认 daemon.status/logs/resource.state/task.list 经 CLI 与裸 socket 均正确返回 envelope， bat.* 别名生效，未知/未实现返回对应 BAT-ERR 码。 对应 issue #1（协议基础设施 + 最小只读方法集）。 Co-Authored-By: Claude Fable 5 --- infrastructure/src/bin/bat_official_sync.rs | 420 +++++++++++++++++--- 1 file changed, 363 insertions(+), 57 deletions(-) diff --git a/infrastructure/src/bin/bat_official_sync.rs b/infrastructure/src/bin/bat_official_sync.rs index 1be917d..ef032ef 100644 --- a/infrastructure/src/bin/bat_official_sync.rs +++ b/infrastructure/src/bin/bat_official_sync.rs @@ -1,4 +1,5 @@ use bat_adapters::official::yostar_jp::PatchPlatform; +use bat_core::{ApiError, ErrorCode}; use bat_infrastructure::{ gc_orphan_staging, lexical_absolute, open_append_file, read_file_no_symlink, read_version_state, redact_proxy_url, resolve_curl_proxy, validate_output_root, @@ -602,11 +603,102 @@ struct DaemonRpcAck { force: Option, } -const RPC_METHOD_STATUS: &str = "bat.status"; -const RPC_METHOD_STOP: &str = "bat.stop"; -const RPC_METHOD_RELOAD: &str = "bat.reload"; -const RPC_METHOD_REFRESH: &str = "bat.refresh"; -const RPC_METHOD_LOGS: &str = "bat.logs"; +// 规范方法名采用国际惯例的 `.`。`bat.*` 保留为向后兼容别名。 +const RPC_METHOD_STATUS: &str = "daemon.status"; +const RPC_METHOD_STOP: &str = "daemon.stop"; +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_TASK_STATUS: &str = "task.status"; +const RPC_METHOD_TASK_LIST: &str = "task.list"; + +/// 把 `bat.*` 兼容别名解析为规范的 `daemon.*` 方法名；其余原样返回。 +fn canonical_rpc_method(method: &str) -> &str { + match method { + "bat.status" => RPC_METHOD_STATUS, + "bat.stop" => RPC_METHOD_STOP, + "bat.reload" => RPC_METHOD_RELOAD, + "bat.refresh" => RPC_METHOD_REFRESH, + "bat.logs" => RPC_METHOD_LOGS, + other => other, + } +} + +/// 判断方法是否属于已规划但尚未实现的命名空间/动作（返回 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.") + || method.starts_with("patch.") + || method.starts_with("unityfs.") +} + +/// RPC 应用层统一 envelope，装入 JSON-RPC 2.0 的 `result`。 +/// +/// 所有响应都带 `ok`/`status`/`data`/`error`/`request_id`；传输层错误（请求解析失败） +/// 仍走 JSON-RPC 顶层 `error`。 +#[derive(Debug, Serialize)] +struct RpcEnvelope { + ok: bool, + status: &'static str, + #[serde(skip_serializing_if = "Option::is_none")] + data: Option, + #[serde(skip_serializing_if = "Option::is_none")] + error: Option, + request_id: String, +} + +/// 生成进程内唯一的 request_id（`req--`）。 +fn next_request_id() -> String { + use std::sync::atomic::{AtomicU64, Ordering}; + static COUNTER: AtomicU64 = AtomicU64::new(1); + format!( + "req-{}-{}", + std::process::id(), + COUNTER.fetch_add(1, Ordering::Relaxed) + ) +} + +fn rpc_envelope_ok( + request_id: String, + status: &'static str, + data: serde_json::Value, +) -> RpcEnvelope { + RpcEnvelope { + ok: true, + status, + data: Some(data), + error: None, + request_id, + } +} + +fn rpc_envelope_error(request_id: String, error: ApiError) -> RpcEnvelope { + RpcEnvelope { + ok: false, + status: "error", + data: None, + error: Some(error), + request_id, + } +} + +/// 把内部 `anyhow::Result` 转成 envelope；成功为 ok，失败归为内部错误码。 +fn rpc_envelope_from_result( + request_id: String, + location: &'static str, + result: anyhow::Result, +) -> RpcEnvelope { + match result { + Ok(data) => rpc_envelope_ok(request_id, "ok", data), + Err(error) => rpc_envelope_error( + request_id, + ApiError::new(ErrorCode::INTERNAL, location, error.to_string()), + ), + } +} fn record_daemon_status( enabled: bool, @@ -1068,72 +1160,168 @@ fn handle_daemon_rpc_request( control: &DaemonControl, ) -> JsonRpcResponse { let id = request.id.clone(); - let result: anyhow::Result = match request.method.as_str() { - RPC_METHOD_STATUS => { - let report = build_daemon_status_report(state_dir); - report.and_then(|report| serde_json::to_value(report).map_err(anyhow::Error::from)) + let request_id = next_request_id(); + let envelope = dispatch_rpc_method(&request, state_dir, control, 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()), + } +} + +/// 命名空间分发：把请求路由到对应 handler，统一返回应用层 envelope。 +fn dispatch_rpc_method( + request: &JsonRpcRequest, + state_dir: &Path, + control: &DaemonControl, + request_id: String, +) -> RpcEnvelope { + match canonical_rpc_method(&request.method) { + RPC_METHOD_STATUS => rpc_envelope_from_result( + request_id, + "daemon.status", + build_daemon_status_report(state_dir) + .and_then(|report| serde_json::to_value(report).map_err(anyhow::Error::from)), + ), + RPC_METHOD_LOGS => { + let tail = match rpc_tail_param(request.params.as_ref(), 200) { + Ok(tail) => tail, + Err(error) => { + return rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::RPC_INVALID_PARAMS, + "daemon.logs", + error.to_string(), + ), + ) + } + }; + rpc_envelope_from_result( + request_id, + "daemon.logs", + build_logs_report(state_dir, tail) + .and_then(|report| serde_json::to_value(report).map_err(anyhow::Error::from)), + ) } RPC_METHOD_STOP => { daemon_control_mark_stop_requested(control); let _ = update_daemon_state_only(state_dir, "stopping"); - serde_json::to_value(DaemonRpcAck { - command: "stop", - status: "accepted", - message: "后台停止请求已发送", - state_dir: state_dir.to_path_buf(), - socket_path: daemon_socket_path(state_dir), - force: None, - }) - .map_err(anyhow::Error::from) + rpc_envelope_ok( + request_id, + "accepted", + rpc_ack_value("stop", "后台停止请求已发送", state_dir, None), + ) } RPC_METHOD_RELOAD => { daemon_control_request_reload(control); - serde_json::to_value(DaemonRpcAck { - command: "reload", - status: "accepted", - message: "后台重新加载请求已发送；空闲时会立即唤醒，忙碌时会在当前轮结束后重新执行自动发现和强制刷新", - state_dir: state_dir.to_path_buf(), - socket_path: daemon_socket_path(state_dir), - force: Some(true), - }) - .map_err(anyhow::Error::from) + rpc_envelope_ok( + request_id, + "accepted", + rpc_ack_value( + "reload", + "后台重新加载请求已发送；空闲时会立即唤醒，忙碌时会在当前轮结束后重新执行自动发现和强制刷新", + state_dir, + Some(true), + ), + ) } RPC_METHOD_REFRESH => { let force = rpc_bool_param(request.params.as_ref(), "force").unwrap_or(false); daemon_control_request_refresh(control, force); - serde_json::to_value(DaemonRpcAck { - command: "refresh", - status: "accepted", - message: if force { - "后台强制刷新请求已发送；空闲时会立即唤醒，忙碌时会在当前轮结束后执行" - } else { - "后台刷新检查请求已发送；空闲时会立即唤醒，忙碌时会在当前轮结束后执行" - }, - state_dir: state_dir.to_path_buf(), - socket_path: daemon_socket_path(state_dir), - force: Some(force), - }) - .map_err(anyhow::Error::from) - } - RPC_METHOD_LOGS => { - let tail = match rpc_tail_param(request.params.as_ref(), 200) { - Ok(tail) => tail, - Err(error) => return json_rpc_error(id, -32602, error.to_string()), + let message = if force { + "后台强制刷新请求已发送；空闲时会立即唤醒，忙碌时会在当前轮结束后执行" + } else { + "后台刷新检查请求已发送；空闲时会立即唤醒，忙碌时会在当前轮结束后执行" }; - let report = build_logs_report(state_dir, tail); - report.and_then(|report| serde_json::to_value(report).map_err(anyhow::Error::from)) + rpc_envelope_ok( + request_id, + "accepted", + rpc_ack_value("refresh", message, state_dir, Some(force)), + ) } - _ => { - return json_rpc_error(id, -32601, format!("未知后台 RPC 方法：{}", request.method)); + RPC_METHOD_RESOURCE_STATE => rpc_envelope_from_result( + request_id, + "resource.state", + build_resource_state_report(state_dir), + ), + 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(); + // 任务执行器尚未落地（阶段 1b）；任务库为空，任何 task_id 都不存在。 + rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::TASK_NOT_FOUND, + "task.status", + format!("任务不存在：{task_id}"), + ), + ) } - }; - - match result { - Ok(value) => json_rpc_result(id, value), - Err(error) => json_rpc_error(id, -32603, error.to_string()), + RPC_METHOD_TASK_LIST => { + rpc_envelope_ok(request_id, "ok", serde_json::json!({ "tasks": [] })) + } + pending if is_pending_rpc_method(pending) => rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::RPC_NOT_IMPLEMENTED, + "rpc.dispatch", + format!("方法尚未实现：{pending}"), + ), + ), + unknown => rpc_envelope_error( + request_id, + ApiError::new( + ErrorCode::RPC_UNKNOWN_METHOD, + "rpc.dispatch", + format!("未知 RPC 方法：{unknown}"), + ), + ), } } +fn rpc_ack_value( + command: &'static str, + message: &'static str, + state_dir: &Path, + force: Option, +) -> serde_json::Value { + serde_json::to_value(DaemonRpcAck { + command, + status: "accepted", + message, + state_dir: state_dir.to_path_buf(), + socket_path: daemon_socket_path(state_dir), + force, + }) + .unwrap_or(serde_json::Value::Null) +} + +/// 构建 `resource.state` 数据：资源发布根、版本状态、上次同步结果。 +fn build_resource_state_report(state_dir: &Path) -> anyhow::Result { + let status_file = read_daemon_status_file(&daemon_status_path(state_dir))?; + let resource_output_root = status_file + .as_ref() + .map(|status| status.resource_output_root.clone()); + let version_state = resource_output_root + .as_ref() + .map(|root| root.join("official-version-state.json")) + .and_then(|path| read_version_state(&path).ok().flatten()); + Ok(serde_json::json!({ + "resource_output_root": resource_output_root, + "version_state": version_state, + "last_update_status": status_file + .as_ref() + .and_then(|status| status.last_update_status.clone()), + "last_success_unix_seconds": status_file + .as_ref() + .and_then(|status| status.last_success_unix_seconds), + })) +} + #[cfg(unix)] fn write_json_rpc_response( stream: &mut UnixStream, @@ -1211,13 +1399,38 @@ fn daemon_rpc_call( let response: JsonRpcClientResponse = serde_json::from_str(&line)?; if let Some(error) = response.error { return Err(anyhow::anyhow!( - "后台 RPC {} 失败：{} ({})", + "后台 RPC {} 传输层失败：{} ({})", method, error.message, error.code )); } - Ok(response.result.unwrap_or(serde_json::Value::Null)) + // 解包应用层 envelope：检查 ok，成功返回 data，失败把 envelope.error 转为 Err。 + let envelope = response + .result + .ok_or_else(|| anyhow::anyhow!("后台 RPC {method} 未返回 result"))?; + let ok = envelope + .get("ok") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + if !ok { + let error = envelope.get("error"); + let code = error + .and_then(|error| error.get("code")) + .and_then(serde_json::Value::as_str) + .unwrap_or(""); + let message = error + .and_then(|error| error.get("message")) + .and_then(serde_json::Value::as_str) + .unwrap_or("未知错误"); + return Err(anyhow::anyhow!( + "后台 RPC {method} 失败：[{code}] {message}" + )); + } + Ok(envelope + .get("data") + .cloned() + .unwrap_or(serde_json::Value::Null)) } #[cfg(not(unix))] @@ -4668,6 +4881,97 @@ mod tests { assert!(error.to_string().contains(PROXY_URL_ENV_VAR)); } + fn rpc_request(method: &str, params: Option) -> JsonRpcRequest { + JsonRpcRequest { + jsonrpc: Some("2.0".to_string()), + id: Some(serde_json::json!(1)), + method: method.to_string(), + params, + } + } + + #[test] + fn canonical_rpc_method_resolves_aliases() { + assert_eq!(canonical_rpc_method("bat.status"), RPC_METHOD_STATUS); + assert_eq!(canonical_rpc_method("bat.refresh"), RPC_METHOD_REFRESH); + assert_eq!(canonical_rpc_method("daemon.status"), RPC_METHOD_STATUS); + assert_eq!(canonical_rpc_method("resource.state"), "resource.state"); + assert_eq!(canonical_rpc_method("unknown.method"), "unknown.method"); + } + + #[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("daemon.status")); + assert!(!is_pending_rpc_method("resource.state")); + } + + #[test] + fn dispatch_unknown_method_returns_unknown_error_envelope() { + let temp = tempfile::TempDir::new().unwrap(); + let control = new_daemon_control(); + let envelope = dispatch_rpc_method( + &rpc_request("nope.nope", None), + temp.path(), + &control, + "req-test-1".to_string(), + ); + let value = serde_json::to_value(&envelope).unwrap(); + assert_eq!(value["ok"], false); + assert_eq!(value["error"]["code"], "BAT-ERR-700001"); + assert_eq!(value["request_id"], "req-test-1"); + } + + #[test] + fn dispatch_pending_method_returns_not_implemented_envelope() { + let temp = tempfile::TempDir::new().unwrap(); + let control = new_daemon_control(); + let envelope = dispatch_rpc_method( + &rpc_request("resource.sync", None), + temp.path(), + &control, + "req-test-2".to_string(), + ); + let value = serde_json::to_value(&envelope).unwrap(); + assert_eq!(value["ok"], false); + assert_eq!(value["error"]["code"], "BAT-ERR-700003"); + } + + #[test] + fn dispatch_task_list_returns_empty_ok_envelope() { + let temp = tempfile::TempDir::new().unwrap(); + let control = new_daemon_control(); + let envelope = dispatch_rpc_method( + &rpc_request("task.list", None), + temp.path(), + &control, + "req-test-3".to_string(), + ); + let value = serde_json::to_value(&envelope).unwrap(); + assert_eq!(value["ok"], true); + assert_eq!(value["status"], "ok"); + assert_eq!(value["data"]["tasks"], serde_json::json!([])); + } + + #[test] + fn dispatch_task_status_reports_task_not_found() { + let temp = tempfile::TempDir::new().unwrap(); + let control = new_daemon_control(); + let envelope = dispatch_rpc_method( + &rpc_request("task.status", Some(serde_json::json!({ "task_id": "abc" }))), + temp.path(), + &control, + "req-test-4".to_string(), + ); + let value = serde_json::to_value(&envelope).unwrap(); + assert_eq!(value["ok"], false); + assert_eq!(value["error"]["code"], "BAT-ERR-700004"); + } + #[test] fn parses_explicit_watch_error_retry_interval() { let options = @@ -4873,7 +5177,9 @@ mod tests { "params": { "force": true } })) .unwrap(); - assert_eq!(request.method, RPC_METHOD_REFRESH); + // bat.* 是兼容别名，解析为规范的 daemon.* 方法名。 + assert_eq!(request.method, "bat.refresh"); + assert_eq!(canonical_rpc_method(&request.method), RPC_METHOD_REFRESH); assert_eq!(rpc_bool_param(request.params.as_ref(), "force"), Some(true)); let response = json_rpc_result(