feat(daemon): RPC 统一 envelope、命名空间分发与只读方法

阶段 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-<pid>-<seq>。
- 命名空间分发:方法改用国际惯例 <namespace>.<action>;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 <noreply@anthropic.com>
This commit is contained in:
2026-07-17 04:19:22 -07:00
co-authored by Claude Fable 5
parent 1a25ea58dd
commit b95a3ac417
+360 -54
View File
@@ -1,4 +1,5 @@
use bat_adapters::official::yostar_jp::PatchPlatform; use bat_adapters::official::yostar_jp::PatchPlatform;
use bat_core::{ApiError, ErrorCode};
use bat_infrastructure::{ use bat_infrastructure::{
gc_orphan_staging, lexical_absolute, open_append_file, read_file_no_symlink, gc_orphan_staging, lexical_absolute, open_append_file, read_file_no_symlink,
read_version_state, redact_proxy_url, resolve_curl_proxy, validate_output_root, read_version_state, redact_proxy_url, resolve_curl_proxy, validate_output_root,
@@ -602,11 +603,102 @@ struct DaemonRpcAck {
force: Option<bool>, force: Option<bool>,
} }
const RPC_METHOD_STATUS: &str = "bat.status"; // 规范方法名采用国际惯例的 `<namespace>.<action>`。`bat.*` 保留为向后兼容别名。
const RPC_METHOD_STOP: &str = "bat.stop"; const RPC_METHOD_STATUS: &str = "daemon.status";
const RPC_METHOD_RELOAD: &str = "bat.reload"; const RPC_METHOD_STOP: &str = "daemon.stop";
const RPC_METHOD_REFRESH: &str = "bat.refresh"; const RPC_METHOD_RELOAD: &str = "daemon.reload";
const RPC_METHOD_LOGS: &str = "bat.logs"; 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_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
error: Option<ApiError>,
request_id: String,
}
/// 生成进程内唯一的 request_id`req-<pid>-<seq>`)。
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<Value>` 转成 envelope;成功为 ok,失败归为内部错误码。
fn rpc_envelope_from_result(
request_id: String,
location: &'static str,
result: anyhow::Result<serde_json::Value>,
) -> 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( fn record_daemon_status(
enabled: bool, enabled: bool,
@@ -1068,70 +1160,166 @@ fn handle_daemon_rpc_request(
control: &DaemonControl, control: &DaemonControl,
) -> JsonRpcResponse { ) -> JsonRpcResponse {
let id = request.id.clone(); let id = request.id.clone();
let result: anyhow::Result<serde_json::Value> = match request.method.as_str() { let request_id = next_request_id();
RPC_METHOD_STATUS => { let envelope = dispatch_rpc_method(&request, state_dir, control, request_id);
let report = build_daemon_status_report(state_dir); match serde_json::to_value(&envelope) {
report.and_then(|report| serde_json::to_value(report).map_err(anyhow::Error::from)) 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 => { RPC_METHOD_STOP => {
daemon_control_mark_stop_requested(control); daemon_control_mark_stop_requested(control);
let _ = update_daemon_state_only(state_dir, "stopping"); let _ = update_daemon_state_only(state_dir, "stopping");
serde_json::to_value(DaemonRpcAck { rpc_envelope_ok(
command: "stop", request_id,
status: "accepted", "accepted",
message: "后台停止请求已发送", rpc_ack_value("stop", "后台停止请求已发送", state_dir, None),
state_dir: state_dir.to_path_buf(), )
socket_path: daemon_socket_path(state_dir),
force: None,
})
.map_err(anyhow::Error::from)
} }
RPC_METHOD_RELOAD => { RPC_METHOD_RELOAD => {
daemon_control_request_reload(control); daemon_control_request_reload(control);
serde_json::to_value(DaemonRpcAck { rpc_envelope_ok(
command: "reload", request_id,
status: "accepted", "accepted",
message: "后台重新加载请求已发送;空闲时会立即唤醒,忙碌时会在当前轮结束后重新执行自动发现和强制刷新", rpc_ack_value(
state_dir: state_dir.to_path_buf(), "reload",
socket_path: daemon_socket_path(state_dir), "后台重新加载请求已发送;空闲时会立即唤醒,忙碌时会在当前轮结束后重新执行自动发现和强制刷新",
force: Some(true), state_dir,
}) Some(true),
.map_err(anyhow::Error::from) ),
)
} }
RPC_METHOD_REFRESH => { RPC_METHOD_REFRESH => {
let force = rpc_bool_param(request.params.as_ref(), "force").unwrap_or(false); let force = rpc_bool_param(request.params.as_ref(), "force").unwrap_or(false);
daemon_control_request_refresh(control, force); daemon_control_request_refresh(control, force);
serde_json::to_value(DaemonRpcAck { let message = if force {
command: "refresh",
status: "accepted",
message: if force {
"后台强制刷新请求已发送;空闲时会立即唤醒,忙碌时会在当前轮结束后执行" "后台强制刷新请求已发送;空闲时会立即唤醒,忙碌时会在当前轮结束后执行"
} else { } else {
"后台刷新检查请求已发送;空闲时会立即唤醒,忙碌时会在当前轮结束后执行" "后台刷新检查请求已发送;空闲时会立即唤醒,忙碌时会在当前轮结束后执行"
}, };
rpc_envelope_ok(
request_id,
"accepted",
rpc_ack_value("refresh", message, state_dir, Some(force)),
)
}
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}"),
),
)
}
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<bool>,
) -> serde_json::Value {
serde_json::to_value(DaemonRpcAck {
command,
status: "accepted",
message,
state_dir: state_dir.to_path_buf(), state_dir: state_dir.to_path_buf(),
socket_path: daemon_socket_path(state_dir), socket_path: daemon_socket_path(state_dir),
force: Some(force), force,
}) })
.map_err(anyhow::Error::from) .unwrap_or(serde_json::Value::Null)
} }
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 report = build_logs_report(state_dir, tail);
report.and_then(|report| serde_json::to_value(report).map_err(anyhow::Error::from))
}
_ => {
return json_rpc_error(id, -32601, format!("未知后台 RPC 方法:{}", request.method));
}
};
match result { /// 构建 `resource.state` 数据:资源发布根、版本状态、上次同步结果。
Ok(value) => json_rpc_result(id, value), fn build_resource_state_report(state_dir: &Path) -> anyhow::Result<serde_json::Value> {
Err(error) => json_rpc_error(id, -32603, error.to_string()), 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)] #[cfg(unix)]
@@ -1211,13 +1399,38 @@ fn daemon_rpc_call(
let response: JsonRpcClientResponse = serde_json::from_str(&line)?; let response: JsonRpcClientResponse = serde_json::from_str(&line)?;
if let Some(error) = response.error { if let Some(error) = response.error {
return Err(anyhow::anyhow!( return Err(anyhow::anyhow!(
"后台 RPC {} 失败:{} ({})", "后台 RPC {} 传输层失败:{} ({})",
method, method,
error.message, error.message,
error.code 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))] #[cfg(not(unix))]
@@ -4668,6 +4881,97 @@ mod tests {
assert!(error.to_string().contains(PROXY_URL_ENV_VAR)); assert!(error.to_string().contains(PROXY_URL_ENV_VAR));
} }
fn rpc_request(method: &str, params: Option<serde_json::Value>) -> 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] #[test]
fn parses_explicit_watch_error_retry_interval() { fn parses_explicit_watch_error_retry_interval() {
let options = let options =
@@ -4873,7 +5177,9 @@ mod tests {
"params": { "force": true } "params": { "force": true }
})) }))
.unwrap(); .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)); assert_eq!(rpc_bool_param(request.params.as_ref(), "force"), Some(true));
let response = json_rpc_result( let response = json_rpc_result(