diff --git a/README.md b/README.md index a8ab2f1..3aba1e4 100644 --- a/README.md +++ b/README.md @@ -14,10 +14,11 @@ uv pip install -e utils/worker ## 核心功能 -- **`@rpc_method` 装饰器**：将异步函数注册为 Zenoh Queryable 端点 +- **`@rpc_method` 装饰器**：将异步函数注册为 Zenoh Queryable 端点（同 `subject#METHOD` 冲突注册即报错） - **自动参数解析**：根据 HTTP 方法从 query 参数或 JSON body 解析请求 - **Pydantic 模型绑定**：`auto_bind_models()` 自动绑定生成的模型 -- **Worker 运行时**：`create_worker` / `create_workers` 管理生命周期 +- **信封协议**：请求负载统一为 `{"meta": {...}, "data": {...}}`，运行时按 `meta.method` 分派 +- **Worker 运行时**：`create_worker` / `create_workers` 管理生命周期（结构化日志） - **`Plugin` 基类**：统一插件初始化 / 清理生命周期（幂等、日志、数据库、后台任务、清理钩子） - **配置加载**：`load_worker_config` 读取 TOML 配置 @@ -34,7 +35,6 @@ uv pip install -e utils/worker # src/plugins/my_plugin/__init__.py import logging -from chongming_worker import auto_bind_models from chongming_worker.plugin import Plugin from . import router # noqa: F401 触发 @rpc_method 注册 @@ -42,8 +42,6 @@ from . import router # noqa: F401 触发 @rpc_method 注册 logger = logging.getLogger(__name__) -auto_bind_models("my_plugin") - class MyPlugin(Plugin): type = "my_plugin" @@ -79,13 +77,13 @@ cleanup = plugin.cleanup ### 框架自动处理 -- **幂等**：重复 `create_workers` 只初始化一次（旧写法需手写 `_initialized` 标志） +- **幂等**：重复 `create_workers` 只初始化一次（`asyncio.Lock` 守卫，并发调用安全） - **统一日志**：`logger` 自动带 `[type]` 前缀 - **数据库**：`init_database()` 统一建路径、建表、记录，`cleanup` 自动 `dispose()` -- **global_var**：`set_var()` / `get_var()` 统一读写插件 `global_var` 模块 +- **global_var**：`set_var()` / `get_var()` 统一读写插件 `global_var` 模块；`cleanup` 时自动将已发布的变量重置为 `None`，避免业务代码拿到已释放的资源 - **后台任务**：`spawn_task()` 跟踪，`cleanup` 时自动取消并 `gather` - **清理钩子**：`register_cleanup()` 注册，`cleanup` 时逆序执行 -- **委托**：自动调用 `runtime.create_workers` 创建 Worker 副本 +- **委托**：自动调用 `runtime.create_workers` 创建 Worker 副本（经 `_create_workers_impl`，测试可替换桩） --- @@ -114,11 +112,32 @@ async def add(request: AddGetRequest) -> AddGetResponse: **装饰器行为**： -| method | 参数来源 | 请求绑定 | -| ------ | ----------------- | ------------------------ | -| `GET` | URL query string | TOML `params` 定义 | -| `POST` | JSON body | 自动绑定到 Pydantic 模型 | -| `MQTT` | MQTT payload JSON | 同 POST | +| method | 参数来源 | 请求绑定 | +| ------ | ------------------ | ------------------------ | +| `GET` | URL query string | TOML `params` 定义 | +| `POST` | JSON body | 自动绑定到 Pydantic 模型 | +| `MQTT` | MQTT payload JSON | 同 POST | + +### 请求信封协议 + +Worker 运行时收到的请求负载统一为信封结构（网关 HTTP 路由 / MQTT Bridge / `chongming-request` Client 自动组装）： + +```json +{ + "meta": { "method": "GET", "user_info": {...}, "timeout": 30, "cacheable": false }, + "data": { "a": 1, "b": 2 } +} +``` + +- `meta.method` 决定分派到哪个 `@rpc_method`（GET/POST/.../MQTT），缺省 `GET` +- `meta.user_info` 注入到签名为 `user_info: dict | None = None` 的处理器参数 +- `data` 为业务参数，经请求模型校验后传给处理器 +- 非信封载荷（无 `meta`/`data`）按旧行为兜底：整包视为 `data`、method 取 GET，并输出一次告警 + +### 注册表冲突检测 + +`@rpc_method` 注册时以 `subject#METHOD` 为唯一键；同一键重复注册且函数不同时直接抛 +`ValueError`（含双方函数位置），不再静默覆盖——静默覆盖会悄悄改变运行时分派行为。 ### `auto_bind_models(worker_type)` @@ -162,10 +181,11 @@ cfg = load_worker_config("example") ## 插件集成模式 -> 推荐使用上文的 **Plugin 基类**；以下为不依赖基类的传统写法（仍受支持）。 +> 推荐使用上文的 **Plugin 基类**（`src/plugins/` 下所有插件均已迁移到该风格）； +> 以下为不依赖基类的传统写法，仅框架层面保留兼容，新插件请勿采用。 ```python -# src/plugins/my_plugin/__init__.py +# src/plugins/my_plugin/__init__.py（旧式写法，已不推荐） from chongming_worker import auto_bind_models from chongming_worker.runtime import create_workers as _create_workers from .router import * # 触发 @rpc_method 注册 @@ -188,7 +208,7 @@ async def create_workers(session, config, count, worker_manager): _initialized = True return _create_workers(session, config, count, worker_manager) -async def cleanup(worker_manager): +async def cleanup(worker_item: str): from .global_var import db await db.dispose() ``` diff --git a/src/chongming_worker/decorators.py b/src/chongming_worker/decorators.py index 90ba9bb..1b73f27 100644 --- a/src/chongming_worker/decorators.py +++ b/src/chongming_worker/decorators.py @@ -15,10 +15,13 @@ """ import inspect import importlib +import logging from typing import Any, Callable, Dict, List, Optional, Tuple, Type, get_type_hints, Union from pydantic import BaseModel, Field, create_model, ValidationError +logger = logging.getLogger("chongming_worker.decorators") + # ============================================================ # 类型系统 @@ -204,7 +207,8 @@ class RPCMethod: try: validated = req_model(**payload) result = validated.model_dump() - # 保留原始 payload 中不属于请求模型的额外字段（如 _method、user_info 等） + # 保留原始 payload 中不属于请求模型的额外字段（信封协议下通常为空， + # 框架字段 method/user_info 由 runtime 从 meta 信封提取注入） for k, v in payload.items(): if k not in result: result[k] = v @@ -342,6 +346,19 @@ def rpc_method( _RPC_REGISTRY[worker_type] = {} # 使用 subject + "#" + method 作为唯一 key，允许多个 HTTP 方法共享同一 subject registry_key = f"{subject}#{method.upper()}" + + # 冲突检测：同 key 重复注册 fail fast，避免静默覆盖改变运行时分派行为 + existing = _RPC_REGISTRY[worker_type].get(registry_key) + if existing is not None: + if existing.func is func: + logger.debug("重复注册同一函数，跳过: %s", registry_key) + return func + raise ValueError( + f"RPC 端点冲突: '{registry_key}' 已注册为 " + f"{existing.func.__module__}.{existing.func.__qualname__}，" + f"当前又注册 {func.__module__}.{func.__qualname__}。" + f"同一 subject + 同一 HTTP 方法只能注册一次。" + ) _RPC_REGISTRY[worker_type][registry_key] = rpc # 给函数附加元数据，方便外部访问 @@ -431,7 +448,10 @@ def auto_bind_models(plugin_name: str): try: models_module = importlib.import_module(f"src.plugins.{plugin_name}.models.pydantic") except ImportError: - print(f"⚠️ auto_bind_models: 未找到 models 模块 (plugins.{plugin_name}.models.pydantic)") + logger.warning( + "auto_bind_models: 未找到 models 模块 (plugins.%s.models.pydantic)", + plugin_name, + ) return # 获取模块中所有 BaseModel 子类 @@ -442,7 +462,7 @@ def auto_bind_models(plugin_name: str): model_classes[name] = obj if not model_classes: - print(f"⚠️ auto_bind_models: 未找到任何 Pydantic 模型") + logger.warning("auto_bind_models: 未找到任何 Pydantic 模型") return registry = _RPC_REGISTRY.get(plugin_name, {}) @@ -458,19 +478,19 @@ def auto_bind_models(plugin_name: str): if req_model_name in model_classes: rpc.request_model = model_classes[req_model_name] bound_count += 1 - print(f" ✓ 绑定请求模型: {req_model_name} -> {rpc.subject} [{rpc.method}]") + logger.debug("绑定请求模型: %s -> %s [%s]", req_model_name, rpc.subject, rpc.method) # 尝试匹配响应模型: Response resp_model_name = f"{op_camel}{method_tag}Response" if resp_model_name in model_classes: rpc.response_model = model_classes[resp_model_name] bound_count += 1 - print(f" ✓ 绑定响应模型: {resp_model_name} -> {rpc.subject} [{rpc.method}]") + logger.debug("绑定响应模型: %s -> %s [%s]", resp_model_name, rpc.subject, rpc.method) if bound_count > 0: - print(f"✅ auto_bind_models: 已绑定 {bound_count} 个模型到 {plugin_name} 的 RPC 端点") + logger.info("auto_bind_models: 已绑定 %d 个模型到 %s 的 RPC 端点", bound_count, plugin_name) else: - print(f"⚠️ auto_bind_models: 未找到匹配的模型绑定") + logger.warning("auto_bind_models: 未找到匹配的模型绑定 (%s)", plugin_name) def get_rpc_methods(worker_type: str) -> Dict[str, RPCMethod]: diff --git a/src/chongming_worker/plugin.py b/src/chongming_worker/plugin.py index 84661c0..d5e8b3c 100644 --- a/src/chongming_worker/plugin.py +++ b/src/chongming_worker/plugin.py @@ -81,6 +81,8 @@ class Plugin: #: Database 实例（由 init_database 填充；类型标注为 Any 以避免强依赖 chongming_database） self.db: Any = None self._initialized = False + #: 初始化互斥锁——同一进程内并发 create_workers（如测试模拟多 worker 启动）串行化 + self._init_lock = asyncio.Lock() self._tasks: list[asyncio.Task] = [] self._cleanup_hooks: list[Callable[[], Awaitable[None]]] = [] self._vars: dict[str, Any] = {} @@ -118,19 +120,29 @@ class Plugin: count: Worker 副本数 worker_manager: WorkerManager 实例（可选） """ - if not self._initialized: - self.config = config or {} - self.logger.info("[%s] 初始化开始 (count=%d)", self.type, count) - if self.auto_bind: - auto_bind_models(self.type) - await self._bootstrap() - await self.on_init() - self._initialized = True - self.logger.info("[%s] 初始化完成", self.type) - else: - self.logger.warning( - "[%s] create_workers 重复调用，跳过重复初始化", self.type - ) + async with self._init_lock: + if not self._initialized: + self.config = config or {} + self.logger.info("[%s] 初始化开始 (count=%d)", self.type, count) + if self.auto_bind: + auto_bind_models(self.type) + await self._bootstrap() + await self.on_init() + self._initialized = True + self.logger.info("[%s] 初始化完成", self.type) + else: + self.logger.warning( + "[%s] create_workers 重复调用，跳过重复初始化", self.type + ) + return self._create_workers_impl(session, config, count, worker_manager) + + def _create_workers_impl(self, session, config, count=1, worker_manager=None): + """ + 创建 Worker 副本任务（委托通用运行时）。 + + 独立方法便于测试用桩替换（如 test_auth_race / verify_example_plugin）， + 避免在测试中依赖真实 zenoh 会话。 + """ return _create_workers(session, config, count, worker_manager) async def cleanup(self, worker_item: str) -> None: @@ -161,6 +173,21 @@ class Plugin: if self.db is not None: await self.db.dispose() self.db = None + # 5. 清空通过 set_var 发布的全局变量，避免 global_var 继续持有已释放资源 + for name in list(self._vars): + try: + gv = self._load_global_var() + if gv is not None: + setter = getattr(gv, "set_global_var", None) + if setter is not None: + setter(name, None) + else: + setattr(gv, name, None) + except Exception as e: # noqa: BLE001 + self.logger.warning( + "[%s] 清理全局变量 %s 失败: %s", self.type, name, e + ) + self._vars.clear() finally: self._initialized = False self.logger.info("[%s] %s 清理完毕", self.type, worker_item) diff --git a/src/chongming_worker/runtime.py b/src/chongming_worker/runtime.py index 0d496b5..525d536 100644 --- a/src/chongming_worker/runtime.py +++ b/src/chongming_worker/runtime.py @@ -6,6 +6,7 @@ Worker 运行时 — 通用的生命周期管理 import os import asyncio import json +import logging import orjson from typing import Any, Dict, List, Optional @@ -16,6 +17,11 @@ from chongming_worker.decorators import ( get_rpc_methods, ) +logger = logging.getLogger("chongming_worker.runtime") + +# 是否已提示过"非信封载荷"警告（每个进程仅提示一次，避免刷屏） +_warned_legacy_payload = False + async def create_worker( session: zenoh.Session, worker_type: str, @@ -27,7 +33,7 @@ async def create_worker( 创建一个 Worker 协程 自动发现通过 @rpc_method 注册的端点，为每个 subject 声明 Zenoh Queryable， - 收到请求后按 _method 字段分派到具体方法。 + 收到请求后按信封 meta.method 字段分派到具体方法。 Args: session: Zenoh 会话 @@ -36,7 +42,7 @@ async def create_worker( config: 自定义配置参数 worker_manager: WorkerManager 实例（用于本地注册） """ - print(f"[Worker] {worker_id} 启动 (type={worker_type})") + logger.info("[Worker] %s 启动 (type=%s)", worker_id, worker_type) # 1. 注册存活令牌 token = session.liveliness().declare_token(f"worker/{worker_type}/{worker_id}") @@ -48,11 +54,11 @@ async def create_worker( # 3. 获取该 worker 类型的所有 RPC 方法 rpc_methods = get_rpc_methods(worker_type) if not rpc_methods: - print(f"[Worker] {worker_id} 警告: 未找到任何 RPC 方法") + logger.warning("[Worker] %s 警告: 未找到任何 RPC 方法", worker_id) else: - print(f"[Worker] {worker_id} 发现 {len(rpc_methods)} 个端点:") + logger.info("[Worker] %s 发现 %d 个端点:", worker_id, len(rpc_methods)) for _key, rpc in rpc_methods.items(): - print(f" {rpc.method} {rpc.subject} -> {rpc.name}({', '.join(rpc.params.keys())})") + logger.info(" %s %s -> %s(%s)", rpc.method, rpc.subject, rpc.name, ", ".join(rpc.params.keys())) # 4. 按 subject 分组（多个方法可共享同一 subject，如 GET + POST） subject_methods: Dict[str, List[RPCMethod]] = {} @@ -67,11 +73,25 @@ async def create_worker( loop = asyncio.get_running_loop() async def _handle_query(query: zenoh.Query): try: - # 解析请求负载 - # payload_str = str(query.payload) if query.payload else "{}" + # 解析请求负载（信封协议: {"meta": {...}, "data": {...}}） payload_bytes = bytes(query.payload) if query.payload else b"" - data = orjson.loads(payload_bytes) if payload_bytes else {} - http_method = data.get("_method", "GET").upper() + raw = orjson.loads(payload_bytes) if payload_bytes else {} + + if isinstance(raw, dict) and ("meta" in raw or "data" in raw): + meta = raw.get("meta") or {} + data = raw.get("data") or {} + else: + # 非信封载荷: 整包视为业务数据，method 默认 GET（无控制字段语义） + global _warned_legacy_payload + if not _warned_legacy_payload: + logger.warning( + "收到非信封格式载荷（缺少 meta/data），按 method=GET 处理: %s", + subject, + ) + _warned_legacy_payload = True + meta = {} + data = raw if isinstance(raw, dict) else {} + http_method = str(meta.get("method", "GET")).upper() # 匹配方法（按 HTTP 方法） matched = next((m for m in methods if m.method.upper() == http_method), None) @@ -85,10 +105,12 @@ async def create_worker( return # 调用 RPC 方法（支持同步和异步） + # user_info 从 meta 信封注入（框架参数，不进业务 payload） result = await matched.call( data, session=session, - worker_id=worker_id + worker_id=worker_id, + user_info=meta.get("user_info"), ) # 回复 @@ -96,7 +118,7 @@ async def create_worker( key_expr=query.key_expr, payload=orjson.dumps(result) ) - print(f"[Worker] {worker_id} 回复 {matched.method} {subject}: {result}") + logger.debug("[Worker] %s 回复 %s %s", worker_id, matched.method, subject) except Exception as e: error_response = { "result": 0.0, @@ -108,7 +130,7 @@ async def create_worker( key_expr=query.key_expr, payload=json.dumps(error_response).encode("utf-8") ) - print(f"[Worker] {worker_id} 异常 {subject}: {e}") + logger.error("[Worker] %s 异常 %s: %s", worker_id, subject, e) def callback(query: zenoh.Query): # 当前 create_worker 运行在事件循环中，可直接 create_task @@ -123,16 +145,16 @@ async def create_worker( # 使用回调方式声明 q = session.declare_queryable(key, make_callback(subject, methods)) queryables.append(q) - print(f"[Worker] {worker_id} Queryable: '{key}' [{', '.join(m.method for m in methods)}]") + logger.info("[Worker] %s Queryable: '%s' [%s]", worker_id, key, ", ".join(m.method for m in methods)) # 6. 保持 worker 存活直到被取消 stop_event = asyncio.Event() try: await stop_event.wait() except asyncio.CancelledError: - print(f"[Worker] {worker_id} 正在关闭...") + logger.info("[Worker] %s 正在关闭...", worker_id) except Exception as e: - print(f"[Worker] {worker_id} 发生异常: {e}") + logger.error("[Worker] %s 发生异常: %s", worker_id, e) finally: # 清理 if worker_manager is not None: @@ -140,7 +162,7 @@ async def create_worker( token.undeclare() for q in queryables: q.undeclare() - print(f"[Worker] {worker_id} 已关闭") + logger.info("[Worker] %s 已关闭", worker_id) def create_workers( session: zenoh.Session, @@ -177,7 +199,7 @@ def create_workers( ) ) tasks.append(task) - print(f"[Worker] 已创建任务: {worker_id} (type={worker_type})") + logger.info("[Worker] 已创建任务: %s (type=%s)", worker_id, worker_type) return tasks