From 8b13ed17e6a80adb89d6b89cee50e4ae1416ee97 Mon Sep 17 00:00:00 2001 From: chakcy <947105045@qq.com> Date: Sat, 22 Aug 2026 03:53:31 +0800 Subject: [PATCH] =?UTF-8?q?modify:=20json=20=E8=A7=A3=E6=9E=90=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/multi-language-plugin-plan.md | 4 +- docs/plugin-wire-protocol.md | 30 +- docs/process-plugin.md | 11 + src/chongming/app/core/route_generator.py | 117 ++++++- src/chongming/app/core/route_registry.py | 9 +- src/chongming/app/core/worker_manager.py | 87 +++-- test_passthrough_optimization.py | 376 ++++++++++++++++++++++ test_route_registry.py | 5 +- test_stream_route.py | 8 +- 9 files changed, 603 insertions(+), 44 deletions(-) create mode 100644 test_passthrough_optimization.py diff --git a/docs/multi-language-plugin-plan.md b/docs/multi-language-plugin-plan.md index 13a2f01..d20e13d 100644 --- a/docs/multi-language-plugin-plan.md +++ b/docs/multi-language-plugin-plan.md @@ -15,7 +15,7 @@ | - | ---- | ---- | | P1 | **规范先行，协议是唯一契约** | 网关与插件之间只认线协议（[plugin-wire-protocol.md](./plugin-wire-protocol.md)）。任何语言实现均以规范为准，不得依赖其他语言 SDK 的内部行为 | | P2 | **网关零改动起步** | 阶段 1 的 Rust 插件按协议实现即可对接，网关代码不允许为适配单语言做改动 | -| P3 | **混合部署兼容** | 同一 `` 下可混布 Python 与 Rust 副本（协议相同、负载均衡透明）；但流式语义、错误格式、超时行为必须与规范完全一致 | +| P3 | **混合部署兼容** | 同一 `` 下可混布 Python 与 Rust 副本（协议相同、负载均衡透明）；但流式语义、错误格式（非流式经 `reply_err` 通道）、超时行为必须与规范完全一致 | | P4 | **契约校验不旁路** | Python SDK 的进程内契约校验（TOML ↔ 注册 ↔ 模型）是 Python 特有能力；其他语言实现须提供等价的自检（描述符字段完整性），网关侧强校验在阶段 2 补齐 | | P5 | **修改即更新规范** | 任何实现侧改动触及协议语义时，必须同步修订规范（含[附录 A 事实来源索引](./plugin-wire-protocol.md)） | | P6 | **性能收益只认实测** | 语言选型以基准测试为准（见 §6），不凭直觉切换 | @@ -75,7 +75,7 @@ chongming-worker-rs/ # 独立仓库（或 utils/worker-rs/，与 Pyth | §2 路由描述符 | `descriptor.rs` 构造 + `publisher.rs` PUT/DELETE；`route_id = subject.replace('.','_') + "__" + METHOD` | | §3 服务发现 | `session.liveliness().declare_token("worker//")`，`--count` 多副本 id 互异 | | §4 RPC | `declare_queryable("worker///")` 异步 handler；信封解析 + 分派；流式多次 `query.reply()` | -| §4.3 异常 | 非流式：`{"result":0.0,"operation":"error","error":...,"worker":...}`；流式：`{"error":...}` 收尾块 | +| §4.3 异常 | 非流式：`query.reply_err(...)`，载荷 `{"result":0.0,"operation":"error","error":...,"worker":...}`；流式：`{"error":...}` 收尾块（普通 reply） | | §5 配置 | `config.rs`：TOML + JSON5 双重解析 | | 生命周期 | `publisher.rs` 10s tokio interval 重发布；`tokio::signal` 监听 SIGTERM/SIGINT → DELETE 全部 key | diff --git a/docs/plugin-wire-protocol.md b/docs/plugin-wire-protocol.md index cf16b1e..c1a830a 100644 --- a/docs/plugin-wire-protocol.md +++ b/docs/plugin-wire-protocol.md @@ -81,6 +81,7 @@ chongming/routes//__meta__ # 插件级元信息（可选，仅在声 | `docstring` | string | `""` | OpenAPI 描述 | | `timeout` | number | `3.0` | 单次调用超时（秒）；流式端点为其**总寿命上限**，须配大（如 300） | | `cacheable` | bool | `false` | 网关侧结果缓存开关（成功响应按 `(type, subject, payload)` 缓存） | +| `validate` | bool | `true` | 网关是否解析并校验请求体（false = 原样透传，校验下沉 worker；仅对 POST/PUT/PATCH 非流式端点生效，见 §4.2） | | `auth_required` | bool | `true` | 网关是否注入鉴权依赖（认证用户经信封 `meta.user_info` 透传给插件） | | `internal` | bool | `false` | 内部端点：不注册 HTTP 路由，仅保留 Zenoh RPC 调用能力 | | `stream` | bool | `false` | 流式端点：网关以 SSE 逐块转发多回复 | @@ -147,17 +148,33 @@ worker/// **兼容规则**：非信封载荷（不含 `meta`/`data` 键的 JSON 对象）整包视为业务 `data`，按 `method = GET` 处理。 +**透传模式（`validate = false` 端点）**：`data` 为请求体的**原始 JSON 值**（网关零解析/校验， +信封由字节拼接构造，结构不变）。插件收到的 `data` 形态与校验模式一致（由 SDK 解析为 dict）； +参数类型转换/校验失败由插件侧暴露（框架异常经 `reply_err` → 网关 503）。空/空白请求体兜底为 `{}`。 + ### 4.3 响应 | 形态 | 行为 | | ---- | ---- | | 普通 | 单个回复，载荷为业务结果 JSON 对象 | -| 异常（非流式） | 单个回复：`{"result": 0.0, "operation": "error", "error": "<异常信息>", "worker": ""}` | -| 流式 | 对同一 Query **多次 reply** 逐块推送；生成器正常结束 = 流结束（无结束哨兵）；中途异常以 `{"error": "<异常信息>"}` **收尾块**结束流；查询已终止（客户端断开）时停止回复 | +| 框架异常（非流式） | 单个 **`reply_err`** 回复，载荷 `{"result": 0.0, "operation": "error", "error": "<异常信息>", "worker": ""}`；网关据此零解析判定失败 → 重试 → HTTP 503 | +| 业务错误 | 普通回复（如 `{"code": 1, "error": "..."}`），HTTP 200 透传，网关不解析 | +| 流式 | 对同一 Query **多次 reply** 逐块推送；生成器正常结束 = 流结束（无结束哨兵）；中途异常以 `{"error": "<异常信息>"}` **收尾块**结束流（流式错误仍为普通回复，不走 `reply_err`）；查询已终止（客户端断开）时停止回复 | + +> **错误通道分层**：框架异常（handler 抛错、无匹配方法、超时）→ `reply_err`； +> 业务错误（业务逻辑判定失败）→ 普通回复。插件作者语义：**返回即业务、抛出即框架**。 +> +> **兼容矩阵**（新旧网关/插件任意组合均语义完好）： +> +> | 网关 \ 插件 | 旧（普通回复错误信封） | 新（reply_err） | +> | ----------- | ---------------------- | --------------- | +> | 旧 | HTTP 200 + 错误信封（历史行为） | `call()` 已处理 reply.err → 重试 → 503 | +> | 新 | HTTP 200 + 错误信封（透传不解析，业务语义） | reply_err → 重试 → **503**（语义修正） | 网关侧对应行为： -- 普通调用 `WorkerManager.call`：取首个回复；空回复 / zenoh 层错误 / 超时 → **切换副本重试**（默认重试 1 次） +- 普通调用 `WorkerManager.call_raw`（HTTP 端点）：取首个回复，**原始字节零解析透传**给 HTTP 响应；空回复 / reply.err / 超时 → **切换副本重试**（默认重试 1 次），重试耗尽 → HTTP 503 +- `call()`（MQTT Bridge 等内部调用方）：同上述重试逻辑，回复解析为 dict 后使用 - 流式调用 `WorkerManager.call_stream`：逐块消费（`ConsolidationMode.NONE`）；收到含 `error` 键的块视为终止信号转发后结束；`reply.err`（zenoh 层错误）同样终止流 - 无可用副本 → HTTP 503 @@ -204,6 +221,13 @@ worker/// - **协议变更分级**： - 破坏性变更（keyexpr 结构、信封结构、错误格式、描述符语义）→ `schema_version` 递增，网关须先发布兼容处理，再切换 - 新增可选字段 / 新增端点形态 → 允许同版本内演进，但必须同步更新本规范 +- **本版同版本内演进说明（2026-08）**： + - 新增描述符字段 `validate`（§2.2，网关缺失时兜底 true，纯增量） + - 非流式框架异常新增 `reply_err` 错误通道（§4.3）：信封结构与错误载荷均不变， + 仅输出通道由普通 reply 改为 reply_err；网关 `call()/call_raw()` 历史上已处理 reply.err， + 新旧组合兼容（见 §4.3 兼容矩阵），故不递增 schema_version + - 网关普通调用由 `call`（dict 语义）切换为 `call_raw`（bytes 透传）：入参出参形态变化 + 为网关内部实现，不影响线上协议 - **修改即更新**：任何实现侧改动（`routes_publisher.py` / `runtime.py` / `route_registry.py` / `worker_manager.py` / `route_generator.py`）触及本规范描述的语义时，必须同步修订本规范 --- diff --git a/docs/process-plugin.md b/docs/process-plugin.md index da3be40..56a2397 100644 --- a/docs/process-plugin.md +++ b/docs/process-plugin.md @@ -240,6 +240,17 @@ from . import hello __all__ = ["hello"] ``` +**异常语义**（插件作者只需 return / raise，通道由 SDK 处理）： + +| 形态 | 插件写法 | 网关行为 | +| ---- | -------- | -------- | +| 业务错误 | `return {"code": 1, "error": "..."}` | HTTP 200 透传（业务语义，网关不解析） | +| 框架异常（非流式） | `raise ValueError(...)` | SDK 经 `reply_err` 通道回复 → 网关重试后 **HTTP 503** | +| 流式异常 | 生成器内 raise | SDK 发送 `{"error": ...}` 收尾块 → SSE error 事件终止 | + +即「**返回即业务、抛出即框架**」。注意：框架异常从历史行为「HTTP 200 + 错误信封」 +修正为 **503**（`reply_err` 通道，见 wire 协议 §4.3 兼容矩阵）。 + ### 第 6 步：本地调试 ```bash diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py index bc70de0..eae772b 100644 --- a/src/chongming/app/core/route_generator.py +++ b/src/chongming/app/core/route_generator.py @@ -5,12 +5,12 @@ 为每个端点生成带有正确 Pydantic 类型注解的处理函数，确保 OpenAPI 文档 正确反映参数类型、响应结构，并注册到 components/schemas。 """ -from typing import Any, AsyncGenerator, Dict, List, Optional +from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple from inspect import Parameter, Signature import orjson from fastapi import FastAPI, HTTPException, APIRouter, Query, Path as FastAPIPath, Request, Depends -from fastapi.responses import StreamingResponse +from fastapi.responses import StreamingResponse, Response from pydantic import create_model, Field, BaseModel from chongming_worker.route_config import WORKER_CONFIG_DIR, load_worker_routes_from_dir @@ -94,6 +94,9 @@ class RouteConfig: self.cacheable: bool = data.get("cacheable", False) self.auth_required: bool = data.get("auth_required", True) self.internal: bool = data.get("internal", False) + # validate=false 时网关不解析/校验请求体，原样透传（校验职责下沉 worker， + # 失败经 reply_err 以 503 暴露）；仅对 POST/PUT/PATCH 非流式端点生效 + self.validate: bool = data.get("validate", True) # 流式端点：以 SSE（text/event-stream）逐块输出 self.stream: bool = data.get("stream", False) # 流式空闲超时（秒）：连续未收到数据块的最大间隔（timeout 为流总寿命上限） @@ -125,6 +128,7 @@ def register_route_group( routes: List[Dict[str, Any]], prefix: str = "", extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, + raw_body_models: Optional[Dict[Tuple[str, str], str]] = None, ) -> List[Any]: """ 为一个 worker_type 的一组路由描述符创建并挂载 APIRouter。 @@ -139,6 +143,8 @@ def register_route_group( routes: 路由描述符列表（TOML 兼容格式，与 load_worker_routes_from_dir 输出一致） prefix: 全局路由前缀（如 /api/v1） extra_schemas: 额外 schema 收集 dict（GET 请求模型注入 OpenAPI 用） + raw_body_models: (path, method) -> 请求模型名 的收集 dict + （validate=false 端点：请求体文档经 requestBody $ref 注入，见 _inject_schemas_to_openapi） Returns: List[Any]: 本次新增到 app.router.routes 的 APIRoute 对象列表 @@ -146,7 +152,14 @@ def register_route_group( """ router = APIRouter(prefix=prefix) for route_conf in routes: - _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas) + _register_single_route(router, worker_manager, worker_type, route_conf, extra_schemas, raw_body_models) + + # raw_body_models 记录的 (path, method) 为端点声明路径（不含 prefix）， + # 而 OpenAPI paths 键为最终挂载路径（prefix + path），这里统一补前缀 + if raw_body_models is not None and prefix: + for key in list(raw_body_models.keys()): + path, method = key + raw_body_models[(prefix + path, method)] = raw_body_models.pop(key) before = set(id(r) for r in app.router.routes) app.include_router(router) @@ -178,6 +191,8 @@ def register_dynamic_routes( return extra_schemas: Dict[str, type[BaseModel]] = {} + # validate=false 端点的 (path, method) -> 请求模型名（OpenAPI requestBody 注入用） + raw_body_models: Dict[Tuple[str, str], str] = {} # 扫描所有 worker 子目录（每个子目录名 = worker_type） for worker_dir in sorted(workers_base.iterdir()): @@ -190,16 +205,16 @@ def register_dynamic_routes( if not routes: continue - register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas) + register_route_group(app, worker_manager, worker_type, routes, prefix, extra_schemas, raw_body_models) # 将额外 schema 注册到 OpenAPI 的 components/schemas if extra_schemas: - _inject_schemas_to_openapi(app, lambda: extra_schemas) + _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models) logger.info(f"动态路由注册完成，worker 配置目录: {WORKER_CONFIG_DIR}") -def _inject_schemas_to_openapi(app: FastAPI, schemas_provider): +def _inject_schemas_to_openapi(app: FastAPI, schemas_provider, raw_body_provider=None): """ 将自定义 Pydantic 模型注册到 FastAPI 的 OpenAPI components/schemas 中。 @@ -210,6 +225,9 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider): schemas_provider: 无参 callable，返回 {name: BaseModel} 字典。 盘扫路径传静态 dict；process 模式传 RouteRegistry 的动态 dict， 使路由表变化后重新生成的 OpenAPI 能反映最新 schema。 + raw_body_provider: 可选无参 callable，返回 {(path, method): model_name} 字典。 + validate=false 端点未声明 body 参数（FastAPI 不会自动生成 requestBody）， + 这里按 (path, method) 补注入 requestBody $ref，保持 OpenAPI 请求体文档完整。 """ original_openapi = app.openapi @@ -224,6 +242,19 @@ def _inject_schemas_to_openapi(app: FastAPI, schemas_provider): for name, model in schemas_provider().items(): if name not in schemas_dict: schemas_dict[name] = model.model_json_schema() + # validate=false 端点：补注入 requestBody（请求体文档，模型已注册到 components） + if raw_body_provider is not None: + for (path, method), model_name in raw_body_provider().items(): + op = schema.get("paths", {}).get(path, {}).get(method.lower()) + if op is not None: + op["requestBody"] = { + "required": True, + "content": { + "application/json": { + "schema": {"$ref": f"#/components/schemas/{model_name}"} + } + }, + } app.openapi_schema = schema return schema @@ -249,14 +280,57 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t if user_info: meta["user_info"] = user_info payload = {"meta": meta, "data": body.model_dump()} - result = await worker_manager.call( + result = await worker_manager.call_raw( + worker_type=worker_type, + subject=subject, + payload=orjson.dumps(payload), + timeout=timeout, + cacheable=cacheable, + ) + # 回复为 worker 原始字节，零解析直接转发（response_model 仅用于 OpenAPI 文档） + return Response(content=result, media_type="application/json") + except RuntimeError as e: + logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) + raise HTTPException(status_code=503, detail=str(e)) + except Exception as e: + logger.error(f"路由 {path} 调用异常: {e}") + raise HTTPException(status_code=500, detail="内部服务器错误") + + return endpoint + + +def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_type: str): + """ + 创建 validate=false 的 body 端点：请求体原样透传（网关零解析/校验）。 + + 信封结构不变（{"meta": ..., "data": <原始 JSON 值>}），通过字节拼接构造， + 网关既不解析 body 也不转义；参数校验/类型转换职责下沉到 worker + （失败经 reply_err 通道以 503 暴露）。空/空白 body 兜底为 {}。 + 请求模型（params TOML 生成）仅用于 OpenAPI 文档（extra_schemas + requestBody 注入）。 + """ + path = rc.path + subject = rc.subject + timeout = rc.timeout + http_method = rc.method + cacheable = rc.cacheable + + async def endpoint(request: Request) -> Any: + try: + meta = {"method": http_method} + user_info = getattr(request.state, "user", None) + if user_info: + meta["user_info"] = user_info + raw = await request.body() + data_part = raw.strip() or b"{}" + payload = b'{"meta":' + orjson.dumps(meta) + b',"data":' + data_part + b'}' + result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, payload=payload, timeout=timeout, cacheable=cacheable, ) - return result + return Response(content=result, media_type="application/json") except RuntimeError as e: logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) @@ -289,21 +363,21 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_ # 无参数情况 if not param_spec_combined: async def _simple_endpoint(request: Request) -> Any: - # 信封协议: {"meta": {...}, "data": {...}} + # 信封协议: {"meta": {...}, "data": {}} meta = {"method": http_method} user_info = getattr(request.state, "user", None) if user_info: meta["user_info"] = user_info payload = {"meta": meta, "data": {}} try: - result = await worker_manager.call( + result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, - payload=payload, + payload=orjson.dumps(payload), timeout=timeout, cacheable=cacheable, ) - return result + return Response(content=result, media_type="application/json") except RuntimeError as e: logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) @@ -323,14 +397,14 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_ meta["user_info"] = user_info payload = {"meta": meta, "data": {k: kwargs[k] for k in param_names}} try: - result = await worker_manager.call( + result = await worker_manager.call_raw( worker_type=worker_type, subject=subject, - payload=payload, + payload=orjson.dumps(payload), timeout=timeout, cacheable=cacheable, ) - return result + return Response(content=result, media_type="application/json") except RuntimeError as e: logger.warning("路由 %s [%s] 调用失败: %s", path, worker_type, e) raise HTTPException(status_code=503, detail=str(e)) @@ -475,6 +549,7 @@ def _register_single_route( worker_type: str, route_conf: Dict[str, Any], extra_schemas: Optional[Dict[str, type[BaseModel]]] = None, + raw_body_models: Optional[Dict[Tuple[str, str], str]] = None, ): """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档""" rc = RouteConfig(route_conf) @@ -537,7 +612,17 @@ def _register_single_route( RequestModel = generate_request_model( param_spec, request_model_name ) - endpoint = _make_body_endpoint(worker_manager, rc, worker_type, RequestModel) + if rc.validate: + endpoint = _make_body_endpoint(worker_manager, rc, worker_type, RequestModel) + else: + # validate=false：请求体原样透传（网关零解析/校验，校验下沉 worker）。 + # RequestModel 仅用于 OpenAPI 请求体文档： + # 经 extra_schemas 注册 components/schemas + requestBody $ref 注入 + if extra_schemas is not None and request_model_name not in extra_schemas: + extra_schemas[request_model_name] = RequestModel + if raw_body_models is not None: + raw_body_models[(rc.path, rc.method)] = request_model_name + endpoint = _make_raw_body_endpoint(worker_manager, rc, worker_type) else: ResponseModel = generate_response_model( rc.response_model, response_model_name diff --git a/src/chongming/app/core/route_registry.py b/src/chongming/app/core/route_registry.py index 483ef3f..578a2b9 100644 --- a/src/chongming/app/core/route_registry.py +++ b/src/chongming/app/core/route_registry.py @@ -20,7 +20,7 @@ import asyncio import hashlib import json -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple import zenoh @@ -63,6 +63,8 @@ class RouteRegistry: self._route_objs: Dict[str, List[Any]] = {} # OpenAPI 额外 schema（GET 请求模型），随路由表变化 self._schemas: Dict[str, Any] = {} + # validate=false 端点的 (path, method) -> 请求模型名（OpenAPI requestBody 注入用） + self._raw_body_models: Dict[Tuple[str, str], str] = {} # worker_type -> 防抖任务 self._debounce: Dict[str, asyncio.Task] = {} @@ -72,7 +74,7 @@ class RouteRegistry: self._inline_types: set = set() # 将动态 schema 注入 OpenAPI（provider 实时读取本注册表） - _inject_schemas_to_openapi(app, lambda: self._schemas) + _inject_schemas_to_openapi(app, lambda: self._schemas, lambda: self._raw_body_models) def set_inline_types(self, types: set): """设置 inline 插件类型集合（混合模式防双注册）。""" @@ -235,7 +237,8 @@ class RouteRegistry: ] if http_routes: self._route_objs[wtype] = register_route_group( - self._app, self._wm, wtype, http_routes, self._prefix, self._schemas + self._app, self._wm, wtype, http_routes, self._prefix, + self._schemas, self._raw_body_models, ) else: logger.info("[%s] 无 HTTP 路由可挂载（%d 个描述符，全部为 MQTT/internal）", diff --git a/src/chongming/app/core/worker_manager.py b/src/chongming/app/core/worker_manager.py index 3001b09..c780ec2 100644 --- a/src/chongming/app/core/worker_manager.py +++ b/src/chongming/app/core/worker_manager.py @@ -1,4 +1,5 @@ import asyncio +import hashlib import random import json import orjson @@ -264,25 +265,78 @@ class WorkerManager: cacheable: bool = False, ) -> Any: """ - 统一调用入口：缓存 > 负载均衡 > 超时重试 + 统一调用入口（dict 语义，MQTT Bridge 等调用方使用） Args: worker_type: Worker 类型标识（如 "example"） subject: 调用的主题（如 "calc.add"） - payload: 请求参数 + payload: 请求参数（信封格式，由路由层构造） timeout: 单次调用超时（秒） retries: 失败重试次数 cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写） Returns: - Dict[str, Any]: 调用结果 + Dict[str, Any]: 调用结果（worker 回复解析后的 dict） + """ + cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}" + reply_bytes = await self._call_impl( + worker_type, subject, orjson.dumps(payload), + timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key, + ) + return orjson.loads(reply_bytes) if reply_bytes else {} - Raises: - RuntimeError: 无可用 worker 或所有重试均失败 + async def call_raw( + self, + worker_type: str, + subject: str, + payload: bytes, + timeout: float = 3.0, + retries: int = 1, + cacheable: bool = False, + ) -> bytes: + """ + 统一调用入口（bytes 语义，HTTP 端点透传用）：返回 worker 回复的原始字节。 + + 与 call() 的区别: + - payload 为预序列化的请求字节（普通信封或字节拼接信封），网关不再二次序列化 + - 回复不做 JSON 解析，直接返回原始字节（配合 Response(content=bytes) 零开销转发） + - cacheable 时仅在缓存写入前解析一次做成功检查（业务错误不缓存）， + 缓存与返回均保留原始字节 + + Args: + worker_type: Worker 类型标识（如 "example"） + subject: 调用的主题（如 "calc.add"） + payload: 预序列化的请求字节（信封 JSON） + timeout: 单次调用超时（秒） + retries: 失败重试次数 + cacheable: 是否允许缓存结果（缓存原始字节） + + Returns: + bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致） + """ + cache_key = f"{worker_type}:{subject}:{hashlib.sha256(payload).hexdigest()}" + return await self._call_impl( + worker_type, subject, payload, + timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key, + ) + + async def _call_impl( + self, + worker_type: str, + subject: str, + payload_bytes: bytes, + timeout: float = 3.0, + retries: int = 1, + cacheable: bool = False, + cache_key: Optional[str] = None, + ) -> bytes: + """ + 调用实现：缓存 > 负载均衡 > 超时重试，返回回复的原始字节。 + + call() / call_raw() 共用本实现，仅入参出参形态不同（dict ↔ bytes）。 """ # 1. 缓存检查（仅在 cacheable=True 时检查） - cache_key = f"{worker_type}:{subject}:{json.dumps(payload, sort_keys=True)}" - if cacheable and self.cache is not None: + if cacheable and cache_key is not None and self.cache is not None: if cache_key in self.cache: logger.debug(f"命中缓存: {cache_key}") return self.cache[cache_key] @@ -293,7 +347,6 @@ class WorkerManager: await self._raise_no_worker(worker_type) full_key = f"{worker_expr}/{subject}" - payload_bytes = orjson.dumps(payload) last_error = None tried_workers: List[str] = [] @@ -335,16 +388,14 @@ class WorkerManager: reply = first_reply if reply.ok: # type: ignore - # result_str = str(reply.ok.payload) - payload_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"" # type: ignore - if not payload_bytes: - result = {} # 或根据业务返回默认值 - else: - result = orjson.loads(payload_bytes) - # 缓存结果（仅在 cacheable=True 时写入） - if cacheable and self.cache is not None and self.is_success_response(result): - self.cache[cache_key] = result - return result + reply_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"{}" + # 缓存写入（仅在 cacheable=True 时）：解析一次做成功检查， + # 但缓存与返回均保留原始字节（网关侧零二次序列化） + if cacheable and cache_key is not None and self.cache is not None and reply_bytes: + parsed = orjson.loads(reply_bytes) + if self.is_success_response(parsed): + self.cache[cache_key] = reply_bytes + return reply_bytes else: error_payload = reply.err.payload # type: ignore error_msg = str(error_payload) if error_payload else "unknown error" diff --git a/test_passthrough_optimization.py b/test_passthrough_optimization.py new file mode 100644 index 0000000..ea1b2f2 --- /dev/null +++ b/test_passthrough_optimization.py @@ -0,0 +1,376 @@ +""" +透传优化回归测试（依赖隔离：桩掉 chongming 包链、auth 与 chongming_cache） + +覆盖: + A. 回复路径 bytes 透传: + 1. 普通端点返回 worker 原始字节（byte-for-byte，零再序列化）+ application/json + 2. 框架错误（call_raw 抛 RuntimeError）-> HTTP 503 + 3. 业务错误（{"code":1,...} 普通回复）-> HTTP 200 透传（网关不解析） + B. validate=false 请求透传: + 4. POST 信封字节拼接精确断言（网关零解析/零转义） + 5. 空 body 兜底 {"meta":...,"data":{}} + 6. validate=true 端点仍走 Pydantic 校验（422）；validate=false 跳过校验（透传） + 7. GET 端点不受 validate 影响（validate 仅对 body 端点生效） + C. OpenAPI: + 8. validate=false 端点 requestBody $ref 注入 + components schema 完整 + D. WorkerManager 单测（FakeSession）: + 9. call_raw 零解析返回原始字节；call() 仍解析为 dict（MQTT 语义保留） + 10. cacheable: 缓存原始 bytes、二次命中不再发查询、业务错误不缓存 + 11. reply.err -> 重试 -> RuntimeError；空回复兜底 b"{}" + +运行: uv run python test_passthrough_optimization.py +""" +import asyncio +import importlib.util +import logging +import sys +import types +from pathlib import Path +from types import SimpleNamespace + +import orjson +from fastapi import FastAPI +from fastapi.testclient import TestClient + +REPO_ROOT = Path(__file__).resolve().parent +sys.path.insert(0, str(REPO_ROOT / "src")) +sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src")) + +# ---- 依赖隔离：桩掉 chongming 包链 / auth / chongming_cache ---- +_pkg = types.ModuleType("chongming") +_pkg.__path__ = [] +_app = types.ModuleType("chongming.app") +_app.__path__ = [] +_core = types.ModuleType("chongming.app.core") +_core.__path__ = [] +_logger_stub = types.ModuleType("chongming.app.core.logger") +_logger_stub.get_logger = lambda name: logging.getLogger(name) +_auth_stub = types.ModuleType("chongming.app.core.auth") +_auth_stub.store_user_to_request = lambda x: x # 仅用于 Depends，桩掉鉴权 +_cache_stub = types.ModuleType("chongming_cache") +_cache_stub.Cache = object +for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub): + sys.modules[mod.__name__] = mod + +# 加载真实 worker_manager（route_generator 依赖其 StreamIdleTimeoutError） +_wm_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "worker_manager.py" +_wm_spec = importlib.util.spec_from_file_location( + "chongming.app.core.worker_manager", _wm_path +) +_wm_mod = importlib.util.module_from_spec(_wm_spec) +sys.modules["chongming.app.core.worker_manager"] = _wm_mod +_wm_spec.loader.exec_module(_wm_mod) # type: ignore[union-attr] + +# 加载真实 route_generator +_rg_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "route_generator.py" +_rg_spec = importlib.util.spec_from_file_location( + "chongming.app.core.route_generator", _rg_path +) +_rg_mod = importlib.util.module_from_spec(_rg_spec) +sys.modules["chongming.app.core.route_generator"] = _rg_mod +_rg_spec.loader.exec_module(_rg_mod) # type: ignore[union-attr] + +register_route_group = _rg_mod.register_route_group +_inject_schemas_to_openapi = _rg_mod._inject_schemas_to_openapi +WorkerManager = _wm_mod.WorkerManager + +failures = [] + + +def check(name, cond, detail=""): + status = "PASS" if cond else "FAIL" + print(f"[{status}] {name}" + (f" — {detail}" if detail else "")) + if not cond: + failures.append(name) + + +class FakeWorkerManager: + """模拟 WorkerManager：call_raw 返回原始字节；error="no_worker" 时抛 RuntimeError""" + + def __init__(self, reply=b'{"result": 5}', error=None): + self.reply = reply + self.error = error + self.call_raw_payloads = [] # 捕获的预序列化请求字节 + + async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False): + self.call_raw_payloads.append(payload) + if self.error == "no_worker": + raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。") + return self.reply + + +ROUTES = [ + # validate=true 对照（默认） + { + "subject": "demo.echo", "path": "/echo", "method": "POST", + "params": ["msg: str"], "summary": "echo", + "timeout": 3.0, "auth_required": False, + }, + { + "subject": "demo.echo2", "path": "/echo2", "method": "POST", + "params": ["n: int"], "summary": "typed echo", + "timeout": 3.0, "auth_required": False, + }, + { + "subject": "demo.add", "path": "/add", "method": "GET", + "params": ["a: int", "b: int"], "summary": "add", + "timeout": 3.0, "auth_required": False, + }, + # validate=false：请求体原样透传 + { + "subject": "demo.raw", "path": "/raw/echo", "method": "POST", + "params": ["msg: str"], "summary": "raw echo", + "validate": False, "timeout": 3.0, "auth_required": False, + }, + # validate=false + GET 对照（validate 仅对 body 端点生效） + { + "subject": "demo.rawget", "path": "/raw/get", "method": "GET", + "params": ["a: int"], "summary": "raw get", + "validate": False, "timeout": 3.0, "auth_required": False, + }, +] + + +def build_app(fwm): + app = FastAPI() + extra_schemas = {} + raw_body_models = {} + register_route_group( + app, fwm, "demo", ROUTES, prefix="/api/v1", + extra_schemas=extra_schemas, raw_body_models=raw_body_models, + ) + if extra_schemas: + _inject_schemas_to_openapi(app, lambda: extra_schemas, lambda: raw_body_models) + return TestClient(app) + + +# ---- A. 回复路径 bytes 透传 ---- + +def test_reply_bytes_passthrough(): + fwm = FakeWorkerManager(reply=b'{"result": 5}') + client = build_app(fwm) + r = client.get("/api/v1/add?a=1&b=2") + check("GET 状态码 200", r.status_code == 200, str(r.status_code)) + check( + "回复为 worker 原始字节（byte-for-byte 零再序列化）", + r.content == b'{"result": 5}', + repr(r.content), + ) + check( + "Content-Type application/json", + r.headers["content-type"].startswith("application/json"), + r.headers.get("content-type", ""), + ) + data = orjson.loads(fwm.call_raw_payloads[0])["data"] + check("query 参数透传到信封 data", data == {"a": 1, "b": 2}, str(data)) + + +def test_framework_error_503(): + fwm = FakeWorkerManager(error="no_worker") + client = build_app(fwm) + r = client.get("/api/v1/add?a=1&b=2") + check("框架错误（无 worker）-> 503", r.status_code == 503, str(r.status_code)) + + +def test_business_error_passthrough_200(): + fwm = FakeWorkerManager(reply=b'{"code": 1, "error": "nope"}') + client = build_app(fwm) + r = client.post("/api/v1/echo", json={"msg": "hi"}) + check("业务错误（普通回复）-> 200 透传", r.status_code == 200, str(r.status_code)) + check("业务错误体原样透传", r.content == b'{"code": 1, "error": "nope"}', repr(r.content)) + + +# ---- B. validate=false 请求透传 ---- + +def test_validate_false_envelope_splice(): + fwm = FakeWorkerManager() + client = build_app(fwm) + r = client.post("/api/v1/raw/echo", json={"msg": "hello"}) + check("validate=false POST 状态码 200", r.status_code == 200, str(r.status_code)) + expected = b'{"meta":{"method":"POST"},"data":{"msg":"hello"}}' + check( + "信封字节拼接（网关零解析/零转义）", + fwm.call_raw_payloads[0] == expected, + repr(fwm.call_raw_payloads[0]), + ) + check("响应透传", r.content == b'{"result": 5}', repr(r.content)) + + +def test_validate_false_empty_body(): + fwm = FakeWorkerManager() + client = build_app(fwm) + r = client.post("/api/v1/raw/echo", content=b"") + check("空 body 状态码 200", r.status_code == 200, str(r.status_code)) + check( + "空 body 兜底 data:{}", + fwm.call_raw_payloads[0] == b'{"meta":{"method":"POST"},"data":{}}', + repr(fwm.call_raw_payloads[0]), + ) + + +def test_validate_true_still_422(): + fwm = FakeWorkerManager() + client = build_app(fwm) + r = client.post("/api/v1/echo2", json={"n": "not-a-number"}) + check("validate=true 仍 Pydantic 校验（422）", r.status_code == 422, str(r.status_code)) + check("校验失败不发起调用", len(fwm.call_raw_payloads) == 0) + + +def test_validate_false_skips_validation(): + fwm = FakeWorkerManager() + client = build_app(fwm) + r = client.post("/api/v1/raw/echo", json={"n": "not-a-number"}) + check("validate=false 跳过校验（200）", r.status_code == 200, str(r.status_code)) + check( + "原始 body 透传（含类型错误输入）", + fwm.call_raw_payloads[0] == b'{"meta":{"method":"POST"},"data":{"n":"not-a-number"}}', + repr(fwm.call_raw_payloads[0]), + ) + + +def test_validate_false_get_unaffected(): + fwm = FakeWorkerManager() + client = build_app(fwm) + r = client.get("/api/v1/raw/get?a=5") + check("validate=false GET 不受影响（200）", r.status_code == 200, str(r.status_code)) + data = orjson.loads(fwm.call_raw_payloads[0])["data"] + check("GET 参数正常透传", data == {"a": 5}, str(data)) + + +# ---- C. OpenAPI ---- + +def test_openapi_request_body_injected(): + fwm = FakeWorkerManager() + client = build_app(fwm) + schema = client.app.openapi() + + # validate=false 端点：requestBody $ref 注入 + raw_op = schema["paths"]["/api/v1/raw/echo"]["post"] + check( + "validate=false 端点 requestBody 已注入", + raw_op.get("requestBody", {}).get("content", {}).get("application/json", {}).get("schema", {}).get("$ref") + == "#/components/schemas/demo_raw_echo_POST_Request", + str(raw_op.get("requestBody")), + ) + check( + "请求模型注册到 components/schemas", + "demo_raw_echo_POST_Request" in schema["components"]["schemas"], + ) + # validate=true 端点：FastAPI 原生 requestBody（不受影响） + ok_op = schema["paths"]["/api/v1/echo"]["post"] + check( + "validate=true 端点原生 requestBody 保留", + "requestBody" in ok_op, + str(ok_op.get("requestBody")), + ) + # GET 端点无 requestBody + check( + "GET 端点无 requestBody", + "requestBody" not in schema["paths"]["/api/v1/raw/get"]["get"], + ) + + +# ---- D. WorkerManager 单测（FakeSession）---- + +class FakeSession: + """模拟 zenoh Session：get() 返回单回复迭代器""" + + def __init__(self, payload=b'{"result": 5}', err_payload=None): + self.payload = payload + self.err_payload = err_payload + self.get_calls = 0 + + def get(self, *args, **kwargs): + self.get_calls += 1 + if self.err_payload is not None: + reply = SimpleNamespace(ok=None, err=SimpleNamespace(payload=self.err_payload)) + else: + reply = SimpleNamespace(ok=SimpleNamespace(payload=self.payload), err=None) + return iter([reply]) + + +def make_wm(session, cache=None): + wm = WorkerManager(session, cache=cache if cache is not None else {}) + wm._workers_cache = {"demo": ["worker/demo/w1"]} + return wm + + +async def test_wm_call_raw_no_parse(): + session = FakeSession(payload=b'{"result": 5}') + wm = make_wm(session) + result = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}') + check( + "call_raw 返回原始字节（零解析）", + isinstance(result, bytes) and result == b'{"result": 5}', + repr(result), + ) + check("session.get 仅调用 1 次", session.get_calls == 1, str(session.get_calls)) + + +async def test_wm_call_dict_semantics(): + session = FakeSession(payload=b'{"result": 5}') + wm = make_wm(session) + result = await wm.call("demo", "calc.add", {"meta": {}, "data": {}}) + check("call() 仍解析为 dict（MQTT 语义保留）", result == {"result": 5}, repr(result)) + + +async def test_wm_cache_bytes(): + cache = {} + session = FakeSession(payload=b'{"result": 5}') + wm = make_wm(session, cache) + r1 = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True) + r2 = await wm.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True) + check("缓存命中返回原始 bytes", r1 == r2 == b'{"result": 5}', repr(r1)) + check("二次调用命中缓存（不再发查询）", session.get_calls == 1, str(session.get_calls)) + check("缓存值为原始 bytes", all(isinstance(v, bytes) for v in cache.values())) + + # 业务错误（code!=0）不缓存 + session2 = FakeSession(payload=b'{"code": 1, "error": "x"}') + wm2 = make_wm(session2, {}) + await wm2.call_raw("demo", "calc.add", b'{"meta":{},"data":{}}', cacheable=True) + check("业务错误不写入缓存", len(wm2.cache) == 0, str(wm2.cache)) + + +async def test_wm_reply_err_retry(): + session = FakeSession(err_payload=b'{"error": "boom"}') + wm = make_wm(session) + try: + await wm.call_raw("demo", "calc.add", b'{}') + check("reply.err 应抛 RuntimeError", False) + except RuntimeError as e: + check("reply.err 重试后抛 RuntimeError", "boom" in str(e), str(e)) + check("reply.err 触发重试（2 次查询）", session.get_calls == 2, str(session.get_calls)) + + +async def test_wm_empty_reply_fallback(): + session = FakeSession(payload=b"") + wm = make_wm(session) + check("call_raw 空回复兜底 b\"{}\"", await wm.call_raw("demo", "calc.add", b"{}") == b"{}") + check("call() 空回复 → {}", await wm.call("demo", "calc.add", {"meta": {}, "data": {}}) == {}) + + +def run(): + # HTTP 层（TestClient） + test_reply_bytes_passthrough() + test_framework_error_503() + test_business_error_passthrough_200() + test_validate_false_envelope_splice() + test_validate_false_empty_body() + test_validate_true_still_422() + test_validate_false_skips_validation() + test_validate_false_get_unaffected() + test_openapi_request_body_injected() + + # WorkerManager 层（FakeSession） + asyncio.run(test_wm_call_raw_no_parse()) + asyncio.run(test_wm_call_dict_semantics()) + asyncio.run(test_wm_cache_bytes()) + asyncio.run(test_wm_reply_err_retry()) + asyncio.run(test_wm_empty_reply_fallback()) + + print(f"\n结果: {len(failures)} 个失败") + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(run()) diff --git a/test_route_registry.py b/test_route_registry.py index 6896eaa..a554ad1 100644 --- a/test_route_registry.py +++ b/test_route_registry.py @@ -31,11 +31,14 @@ def check(name, cond, detail=""): class FakeWorkerManager: - """桩：无可用 worker，命中路由后 call 抛 RuntimeError（网关转 503）。""" + """桩：无可用 worker，命中路由后 call_raw 抛 RuntimeError（网关转 503）。""" async def call(self, **kwargs): raise RuntimeError("无可用 worker（测试桩）") + async def call_raw(self, **kwargs): + raise RuntimeError("无可用 worker（测试桩）") + class FakeBridge: """记录 update_dynamic_topics 的调用结果。""" diff --git a/test_stream_route.py b/test_stream_route.py index 719f3d0..2112d32 100644 --- a/test_stream_route.py +++ b/test_stream_route.py @@ -86,6 +86,12 @@ class FakeWorkerManager: data = payload["data"] return {"result": data.get("a", 0) + data.get("b", 0)} + async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False): + # 网关透传路径：payload 为预序列化字节，回复原始字节 + self.call_payloads.append((worker_type, subject, orjson.loads(payload))) + data = orjson.loads(payload)["data"] + return orjson.dumps({"result": data.get("a", 0) + data.get("b", 0)}) + async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0): if self.stream_error == "no_worker": raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。") @@ -203,7 +209,7 @@ def test_non_stream_compat(): f"status={r.status_code} body={r.text}") check("非流式调用未进入 call_stream", len(fwm.stream_payloads) == 0) - check("非流式调用仍走 call()", len(fwm.call_payloads) == 2) + check("非流式调用仍走 call_raw", len(fwm.call_payloads) == 2) def run():