diff --git a/src/chongming_worker/plugin.py b/src/chongming_worker/plugin.py index c78d496..ca0b437 100644 --- a/src/chongming_worker/plugin.py +++ b/src/chongming_worker/plugin.py @@ -66,7 +66,10 @@ from pathlib import Path from typing import Any, Awaitable, Callable from chongming_worker.decorators import auto_bind_models -from chongming_worker.runtime import create_workers as _create_workers +from chongming_worker.runtime import ( + create_workers as _create_workers, + resolve_startup_config, +) class Plugin: @@ -152,6 +155,12 @@ class Plugin: """ 插件入口 — 由 app lifespan 调用，返回 Worker 任务列表。 + 初始化前先按配置模式解析 [workers.config] 的单一事实来源（见 + :func:`chongming_worker.runtime.resolve_startup_config`）： + - cache 模式：插件必须读到缓存值（否则会按 toml 种子建客户端， + 出现“前端显示缓存值、实际跑 toml”）； + - toml 模式：用文件值并覆盖缓存，使前端显示与运行值一致。 + Args: session: Zenoh 会话 config: Worker 配置字典（含 [workers.config] 展平后的运行时配置） @@ -160,7 +169,10 @@ class Plugin: """ async with self._init_lock: if not self._initialized: - self.config = config or {} + config = resolve_startup_config( + (config or {}).get("type", self.type), config or {}, worker_manager + ) + self.config = config self.logger.info("[%s] 初始化开始 (count=%d)", self.type, count) if self.auto_bind: auto_bind_models(self.type) diff --git a/src/chongming_worker/runtime.py b/src/chongming_worker/runtime.py index 61ab40d..31f05f5 100644 --- a/src/chongming_worker/runtime.py +++ b/src/chongming_worker/runtime.py @@ -13,7 +13,7 @@ import socket import time import tomllib import orjson -from typing import Any, Awaitable, Callable, Dict, List, Optional +from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple import zenoh @@ -372,12 +372,8 @@ def _resolve_mode(worker_type: str, worker_manager=None) -> str: return _worker_modes.get(worker_type, "toml") -def _read_worker_config_section(cache, worker_type: str) -> Dict[str, Any]: - """读该类型 [workers.config]（缓存优先，缓存缺失回退读 toml 种子）。""" - if cache is not None: - val = cache.get(_configs_cache_key(worker_type)) - if isinstance(val, dict): - return dict(val) +def _read_toml_config_section(worker_type: str) -> Dict[str, Any]: + """读插件 config.toml 的 [workers.config] 段（缺失/损坏按空 dict）。""" try: with open(_plugin_config_path(worker_type), "rb") as f: data = tomllib.load(f) @@ -388,6 +384,26 @@ def _read_worker_config_section(cache, worker_type: str) -> Dict[str, Any]: return {} +def _read_worker_config_section_with_source( + cache, worker_type: str +) -> Tuple[Dict[str, Any], str]: + """读该类型 [workers.config] 并回报实际命中来源（"cache" / "toml"）。 + + 缓存存在但该键缺失时会回退读 toml 种子，此时来源是 "toml" 而非 "cache"—— + get_config 据此如实回报，前端才能显示值到底来自哪一侧。 + """ + if cache is not None: + val = cache.get(_configs_cache_key(worker_type)) + if isinstance(val, dict): + return dict(val), "cache" + return _read_toml_config_section(worker_type), "toml" + + +def _read_worker_config_section(cache, worker_type: str) -> Dict[str, Any]: + """读该类型 [workers.config]（缓存优先，缓存缺失回退读 toml 种子）。""" + return _read_worker_config_section_with_source(cache, worker_type)[0] + + def _persist_worker_config_section(cache, worker_type: str, updates: Dict[str, Any]) -> Dict[str, Any]: """把 updates 合并写入该类型 [workers.config]（缓存侧，全局唯一）。 @@ -414,18 +430,14 @@ def resolve_runtime_config( """启动时按模式对齐 [workers.config] 的单一事实来源，返回覆盖后的运行时配置。 缓存配置（[workers.config]）在传入插件前被展平进运行配置顶层（与 _discover_plugins 一致）。 - - "toml"（默认）：用 toml 的 [workers.config] 覆盖并修正缓存（缓存缺失时写入）； - - "cache"：缓存有值则用缓存覆盖；缓存缺失才用 toml 引导并同步到缓存。 + - "toml"（默认）：以 config/workers//config.toml 为准——启动时把 toml 的 + [workers.config] 覆盖写入缓存（前端读缓存，因此随之显示 toml 值）；运行期经 + config_set 的改动仅本次运行有效，下次启动被 toml 覆盖； + - "cache"：以缓存为准——缓存有值就用缓存（toml 仅作差异提示）；缓存缺失才用 + toml 引导并同步到缓存，此后跨重启保持。 cache 为 None 时保持传入的 config 原样（不启用缓存语义）。 """ - try: - with open(_plugin_config_path(worker_type), "rb") as f: - data = tomllib.load(f) - workers = data.get("workers") or {} - toml_cfg = workers.get("config") if isinstance(workers, dict) else None - toml_cfg = dict(toml_cfg) if isinstance(toml_cfg, dict) else {} - except Exception: # noqa: BLE001 - toml 缺失/损坏 - toml_cfg = {} + toml_cfg = _read_toml_config_section(worker_type) if cache is None: # 无缓存：保持 toml 种子（经典行为，无回写） @@ -434,18 +446,59 @@ def resolve_runtime_config( key = _configs_cache_key(worker_type) if mode == "cache": cached = cache.get(key) - if not isinstance(cached, dict): - cached = toml_cfg - cache[key] = cached - return {**config, **cached} - # toml（默认）：toml 仅作为「启动种子」。缓存缺失时写入； - # 缓存已存在（可能被运行时 config_set 更新过）则保留，不再用 toml 覆盖， - # 否则会丢掉前端/管理指令对内置插件（inline）写入的配置——即“改了不生效”。 - cached = cache.get(key) - if not isinstance(cached, dict): - cached = dict(toml_cfg) - cache[key] = cached - return {**config, **cached} + if isinstance(cached, dict): + _warn_config_divergence(worker_type, toml_cfg, cached) + return {**config, **cached} + # 缓存缺失：用 toml 引导并写回，之后以缓存为准 + cache[key] = dict(toml_cfg) + return {**config, **toml_cfg} + # toml（默认）：以文件为准，覆盖写入缓存，使前端（读缓存）与插件运行值一致 + cache[key] = dict(toml_cfg) + return {**config, **toml_cfg} + + +def resolve_startup_config( + worker_type: str, + config: Dict[str, Any], + worker_manager=None, +) -> Dict[str, Any]: + """插件初始化（_bootstrap/on_init）前解析出的运行时配置。 + + 把 [workers.config] 的单一事实来源按模式解析进来，供 Plugin.create_workers 在 + on_init 之前调用：cache 模式下插件必须读到缓存值（否则插件按 toml 建好客户端后， + 只有 RPC/管理通道读到缓存，出现“前端显示缓存值、实际跑 toml”）；toml 模式下 + 用文件值并覆盖缓存，使前端显示与运行值一致。 + + 无缓存（未配置缓存后端）时原样返回传入的 config。 + """ + conf = dict(config) if isinstance(config, dict) else {} + cache = _resolve_cache(worker_type, worker_manager) + if cache is None: + return conf + mode = _resolve_mode(worker_type, worker_manager) + logger.info("[Worker] %s 插件初始化配置已按 %s 模式解析", worker_type, mode) + return resolve_runtime_config(worker_type, conf, cache, mode) + + +# 已提示过 toml↔缓存差异的类型（进程内去重，避免每次解析重复刷屏） +_divergence_notified: set = set() + + +def _warn_config_divergence( + worker_type: str, toml_cfg: Dict[str, Any], cached: Dict[str, Any] +) -> None: + """cache 模式下 toml 与缓存不一致时提示一次（缓存为准，便于排查"改了 toml 不生效"）。""" + if worker_type in _divergence_notified: + return + diff = sorted(k for k in set(toml_cfg) | set(cached) if toml_cfg.get(k) != cached.get(k)) + if not diff: + return + _divergence_notified.add(worker_type) + logger.info( + "[Worker] %s 的 config.toml 与缓存存在差异（键: %s），cache 模式以缓存为准；" + "如需以 toml 为准，请删除缓存键 %s 后重启（或改用 toml 模式）", + worker_type, ", ".join(diff), _configs_cache_key(worker_type), + ) def _merge_runtime_config(config: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: @@ -473,6 +526,8 @@ class WorkerController: self.current_worker_id: str = "" # 该副本所属类型的配置缓存句柄（由 WorkerPool 注入，供 admin 读写缓存） self.cache = None + # 该副本所属类型的配置优先来源模式（"toml" 默认 / "cache"，由 WorkerPool 注入） + self.mode = "toml" # 配置变更后待应用的新运行时配置（由 config_set 设置，重启时生效） self.pending_config: Optional[Dict[str, Any]] = None self.pending_updates: Optional[Dict[str, Any]] = None @@ -537,9 +592,11 @@ async def _handle_admin_command( controller.draining = True logger.info("[Worker] %s 收到优雅排空指令（将隔离新请求并等待在途请求完成）", worker_id) elif cmd == ADMIN_CMD_GET_CONFIG: - # 查询该类型的 [workers.config]（缓存优先，无缓存回退读 toml 种子）。 + # 查询该类型的 [workers.config]（缓存优先，缓存缺失或无缓存回退读 toml 种子）。 + # source/mode 如实回报值实际来自哪一侧，供网关与前端展示——存在缓存对象但值 + # 来自 toml 种子时必须是 "toml"，否则前端会把 toml 值当成缓存值。 cache = getattr(controller, "cache", None) - cfg = _read_worker_config_section(cache, worker_type) + cfg, source = _read_worker_config_section_with_source(cache, worker_type) query.reply( key_expr=query.key_expr, payload=orjson.dumps({ @@ -548,7 +605,8 @@ async def _handle_admin_command( "worker_type": worker_type, "worker_id": worker_id, "config": cfg, - "source": "cache" if cache is not None else "toml", + "source": source, + "mode": getattr(controller, "mode", "toml"), }), ) return @@ -1123,6 +1181,7 @@ class WorkerPool: self._next_index += 1 controller = WorkerController(self.worker_type, index) controller.cache = self.cache + controller.mode = self.mode task = asyncio.create_task( _managed_worker( self.session, self.worker_type, controller, self.config,