modify: 内置插件 [workers.config] 两种模式语义修正(cache 模式初始化即读缓存)

- resolve_runtime_config 按模式落实单一事实来源:toml 模式以 config.toml 为准并覆盖
  写入缓存(前端读缓存 → 随之显示文件值);cache 模式以缓存为准,缓存缺失才用 toml
  引导并写回,此后跨重启保持。cache 模式下 toml 与缓存不一致时提示差异键与处置方式。
- 新增 resolve_startup_config:Plugin.create_workers 在 _bootstrap()/on_init() 之前调用,
  使插件实例读到与 get_config/前端一致的来源。此前只在 WorkerPool 里合并缓存,插件实例
  仍按 toml 建客户端(Ark/MQTT/DB 等),表现为“前端显示缓存值、实际跑 toml”,且只有
  config_set 那一刻是缓存值、进程重启即回退。
- get_config 如实回报来源:新增 _read_worker_config_section_with_source(值来自缓存 vs
  回退读 toml 种子),并在回复中带 mode(toml/cache),供网关与前端标注展示值来路。
- count 语义未改:toml 模式仍用文件值覆盖缓存,cache 模式缓存有值用缓存。
This commit is contained in:
2026-09-18 15:22:49 +08:00
parent 67f45edc3c
commit d85fbe567a
2 changed files with 105 additions and 34 deletions
+14 -2
View File
@@ -66,7 +66,10 @@ from pathlib import Path
from typing import Any, Awaitable, Callable from typing import Any, Awaitable, Callable
from chongming_worker.decorators import auto_bind_models 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: class Plugin:
@@ -152,6 +155,12 @@ class Plugin:
""" """
插件入口 — 由 app lifespan 调用,返回 Worker 任务列表。 插件入口 — 由 app lifespan 调用,返回 Worker 任务列表。
初始化前先按配置模式解析 [workers.config] 的单一事实来源(见
:func:`chongming_worker.runtime.resolve_startup_config`):
- cache 模式:插件必须读到缓存值(否则会按 toml 种子建客户端,
出现“前端显示缓存值、实际跑 toml”);
- toml 模式:用文件值并覆盖缓存,使前端显示与运行值一致。
Args: Args:
session: Zenoh 会话 session: Zenoh 会话
config: Worker 配置字典(含 [workers.config] 展平后的运行时配置) config: Worker 配置字典(含 [workers.config] 展平后的运行时配置)
@@ -160,7 +169,10 @@ class Plugin:
""" """
async with self._init_lock: async with self._init_lock:
if not self._initialized: 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) self.logger.info("[%s] 初始化开始 (count=%d)", self.type, count)
if self.auto_bind: if self.auto_bind:
auto_bind_models(self.type) auto_bind_models(self.type)
+91 -32
View File
@@ -13,7 +13,7 @@ import socket
import time import time
import tomllib import tomllib
import orjson import orjson
from typing import Any, Awaitable, Callable, Dict, List, Optional from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
import zenoh import zenoh
@@ -372,12 +372,8 @@ def _resolve_mode(worker_type: str, worker_manager=None) -> str:
return _worker_modes.get(worker_type, "toml") return _worker_modes.get(worker_type, "toml")
def _read_worker_config_section(cache, worker_type: str) -> Dict[str, Any]: def _read_toml_config_section(worker_type: str) -> Dict[str, Any]:
"""读该类型 [workers.config](缓存优先,缓存缺失回退读 toml 种子)。""" """读插件 config.toml 的 [workers.config] 段(缺失/损坏按空 dict)。"""
if cache is not None:
val = cache.get(_configs_cache_key(worker_type))
if isinstance(val, dict):
return dict(val)
try: try:
with open(_plugin_config_path(worker_type), "rb") as f: with open(_plugin_config_path(worker_type), "rb") as f:
data = tomllib.load(f) data = tomllib.load(f)
@@ -388,6 +384,26 @@ def _read_worker_config_section(cache, worker_type: str) -> Dict[str, Any]:
return {} 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]: def _persist_worker_config_section(cache, worker_type: str, updates: Dict[str, Any]) -> Dict[str, Any]:
"""把 updates 合并写入该类型 [workers.config](缓存侧,全局唯一)。 """把 updates 合并写入该类型 [workers.config](缓存侧,全局唯一)。
@@ -414,18 +430,14 @@ def resolve_runtime_config(
"""启动时按模式对齐 [workers.config] 的单一事实来源,返回覆盖后的运行时配置。 """启动时按模式对齐 [workers.config] 的单一事实来源,返回覆盖后的运行时配置。
缓存配置([workers.config])在传入插件前被展平进运行配置顶层(与 _discover_plugins 一致)。 缓存配置([workers.config])在传入插件前被展平进运行配置顶层(与 _discover_plugins 一致)。
- "toml"(默认):用 toml 的 [workers.config] 覆盖并修正缓存(缓存缺失时写入); - "toml"(默认):以 config/workers/<type>/config.toml 为准——启动时把 toml 的
- "cache":缓存有值则用缓存覆盖;缓存缺失才用 toml 引导并同步到缓存。 [workers.config] 覆盖写入缓存(前端读缓存,因此随之显示 toml 值);运行期经
config_set 的改动仅本次运行有效,下次启动被 toml 覆盖;
- "cache":以缓存为准——缓存有值就用缓存(toml 仅作差异提示);缓存缺失才用
toml 引导并同步到缓存,此后跨重启保持。
cache 为 None 时保持传入的 config 原样(不启用缓存语义)。 cache 为 None 时保持传入的 config 原样(不启用缓存语义)。
""" """
try: toml_cfg = _read_toml_config_section(worker_type)
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 = {}
if cache is None: if cache is None:
# 无缓存:保持 toml 种子(经典行为,无回写) # 无缓存:保持 toml 种子(经典行为,无回写)
@@ -434,18 +446,59 @@ def resolve_runtime_config(
key = _configs_cache_key(worker_type) key = _configs_cache_key(worker_type)
if mode == "cache": if mode == "cache":
cached = cache.get(key) cached = cache.get(key)
if not isinstance(cached, dict): if isinstance(cached, dict):
cached = toml_cfg _warn_config_divergence(worker_type, toml_cfg, cached)
cache[key] = cached return {**config, **cached}
return {**config, **cached} # 缓存缺失:用 toml 引导并写回,之后以缓存为准
# toml(默认):toml 仅作为「启动种子」。缓存缺失时写入; cache[key] = dict(toml_cfg)
# 缓存已存在(可能被运行时 config_set 更新过)则保留,不再用 toml 覆盖, return {**config, **toml_cfg}
# 否则会丢掉前端/管理指令对内置插件(inline)写入的配置——即“改了不生效”。 # toml(默认):以文件为准,覆盖写入缓存,使前端(读缓存)与插件运行值一致
cached = cache.get(key) cache[key] = dict(toml_cfg)
if not isinstance(cached, dict): return {**config, **toml_cfg}
cached = dict(toml_cfg)
cache[key] = cached
return {**config, **cached} 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]: 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 = "" self.current_worker_id: str = ""
# 该副本所属类型的配置缓存句柄(由 WorkerPool 注入,供 admin 读写缓存) # 该副本所属类型的配置缓存句柄(由 WorkerPool 注入,供 admin 读写缓存)
self.cache = None self.cache = None
# 该副本所属类型的配置优先来源模式("toml" 默认 / "cache",由 WorkerPool 注入)
self.mode = "toml"
# 配置变更后待应用的新运行时配置(由 config_set 设置,重启时生效) # 配置变更后待应用的新运行时配置(由 config_set 设置,重启时生效)
self.pending_config: Optional[Dict[str, Any]] = None self.pending_config: Optional[Dict[str, Any]] = None
self.pending_updates: 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 controller.draining = True
logger.info("[Worker] %s 收到优雅排空指令(将隔离新请求并等待在途请求完成)", worker_id) logger.info("[Worker] %s 收到优雅排空指令(将隔离新请求并等待在途请求完成)", worker_id)
elif cmd == ADMIN_CMD_GET_CONFIG: elif cmd == ADMIN_CMD_GET_CONFIG:
# 查询该类型的 [workers.config](缓存优先,无缓存回退读 toml 种子)。 # 查询该类型的 [workers.config](缓存优先,缓存缺失或无缓存回退读 toml 种子)。
# source/mode 如实回报值实际来自哪一侧,供网关与前端展示——存在缓存对象但值
# 来自 toml 种子时必须是 "toml",否则前端会把 toml 值当成缓存值。
cache = getattr(controller, "cache", None) 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( query.reply(
key_expr=query.key_expr, key_expr=query.key_expr,
payload=orjson.dumps({ payload=orjson.dumps({
@@ -548,7 +605,8 @@ async def _handle_admin_command(
"worker_type": worker_type, "worker_type": worker_type,
"worker_id": worker_id, "worker_id": worker_id,
"config": cfg, "config": cfg,
"source": "cache" if cache is not None else "toml", "source": source,
"mode": getattr(controller, "mode", "toml"),
}), }),
) )
return return
@@ -1123,6 +1181,7 @@ class WorkerPool:
self._next_index += 1 self._next_index += 1
controller = WorkerController(self.worker_type, index) controller = WorkerController(self.worker_type, index)
controller.cache = self.cache controller.cache = self.cache
controller.mode = self.mode
task = asyncio.create_task( task = asyncio.create_task(
_managed_worker( _managed_worker(
self.session, self.worker_type, controller, self.config, self.session, self.worker_type, controller, self.config,