modify: worker 管理多进程副本

pyproject.toml

@@ -4,16 +4,16 @@ requires = ["hatchling"]
 
 [project]
 name = "chongming-worker"
-version = "0.4.0"
+version = "0.4.1"
 description = "chongming worker"
 readme = "README.md"
 requires-python = ">=3.12"
 dependencies = [
-    "eclipse-zenoh>=1.9.0",
-    "pydantic>=2.0",
-    "tomli>=2.4.1",
-    "tomli-w>=1.0.0",
-    "minio>=7.2.0",
+    "eclipse-zenoh",
+    "pydantic",
+    "tomli",
+    "minio",
+    "chongming-cache",
 ]
 
 [project.scripts]

src/chongming_worker/process.py

@@ -43,7 +43,7 @@ def load_plugin_config(worker_type: str, config_dir: str = "config/workers") ->
 
     Returns:
         {"type", "module", "factory", "cleanup", "count", "load_mode",
-         "mqtt_topics", "config": 展平后的运行时配置}
+         "mqtt_topics", "cache"([workers.cache] 段), "config": 展平后的运行时配置}
     """
     import pathlib
 
@@ -67,7 +67,7 @@ def load_plugin_config(worker_type: str, config_dir: str = "config/workers") ->
         mqtt_topics.append(t)
 
     workers_flat = {k: v for k, v in workers_section.items()
-                    if k not in ("config", "mqtt_topics", "load_mode")}
+                    if k not in ("config", "mqtt_topics", "load_mode", "cache")}
     runtime_config = {
         **plugin_conf,
         **workers_flat,
@@ -82,10 +82,24 @@ def load_plugin_config(worker_type: str, config_dir: str = "config/workers") ->
         "count": workers_section.get("count", 1),
         "load_mode": load_mode,
         "mqtt_topics": mqtt_topics,
+        "cache": dict(workers_section.get("cache") or {}),
         "config": runtime_config,
     }
 
 
+def build_worker_cache(cache_cfg: dict):
+    """按插件 [workers.cache] 段建立配置缓存（redis/diskcache），返回 (cache, mode)。
+
+    从配置中剔除内部控制键 launch_worker_count（决定配置优先来源 toml/cache），
+    避免其被当作后端连接参数传给 Cache.from_config。
+    """
+    from chongming_cache import Cache
+
+    mode = str((cache_cfg or {}).get("launch_worker_count", "toml")).lower()
+    clean = {k: v for k, v in (cache_cfg or {}).items() if k != "launch_worker_count"}
+    return Cache.from_config(clean), mode
+
+
 def worker_serve():
     """独立 Worker 进程入口（console script: chongming-worker / worker_serve）。"""
     # 独立进程不依赖网关的日志配置，兜底 basicConfig（已有 handler 时不覆盖）
@@ -99,6 +113,12 @@ def worker_serve():
     parser.add_argument("--type", required=True, help="插件类型（如 demo / example）")
     parser.add_argument("--count", type=int, default=None, help="副本数（覆盖配置中的 count）")
     parser.add_argument("--config", default="config.toml", help="全局配置文件路径（读取 [zenoh] 节）")
+    parser.add_argument(
+        "--instance", default=None,
+        help="外部进程实例标识（多实例全局分摊/逐实例精细扩缩用）。缺省用 "
+             "CHONGMING_WORKER_INSTANCE 或 {hostname}:{pid}（同机多实例默认即自动区分）；"
+             "跨重启稳定适逐实例缩放请显式指定",
+    )
     args = parser.parse_args()
 
     # console script 启动时 sys.path 不含当前目录，显式加入以保证
@@ -136,6 +156,36 @@ def worker_serve():
         session = zenoh.open(zconfig)
         logger.info("[%s] Zenoh 会话已建立", args.type)
 
+        # 1.5 配置缓存（[workers.cache] 段，进程插件自行管理）+ 配置优先来源登记
+        #     [workers.config] 的单一事实来源解析由 worker 运行时（create_workers）统一完成。
+        #     多实例场景下还登记「外部进程实例参与方」，使扩缩容按存活实例数分摊全局 count。
+        from chongming_worker.runtime import (
+            register_instance,
+            run_instance_heartbeat,
+            set_current_instance_id,
+            set_worker_cache,
+            set_worker_mode,
+            unregister_instance,
+        )
+        set_current_instance_id(args.instance)
+
+        cache = None
+        mode = "toml"
+        hb_task = None  # 实例参与方心跳任务（外部多实例全局分摊用）
+        if wc.get("cache"):
+            cache, mode = build_worker_cache(wc["cache"])
+            set_worker_cache(args.type, cache)
+            set_worker_mode(args.type, mode)
+            register_instance(args.type, cache, args.instance)
+            hb_task = asyncio.create_task(
+                run_instance_heartbeat(args.type, cache, instance_id=args.instance)
+            )
+            logger.info(
+                "[%s] 已启用配置缓存（backend=%s, mode=%s）并登记外部实例参与方",
+                args.type, getattr(getattr(cache, "backend", None), "name", "?"), mode,
+            )
+        worker_config = wc["config"]
+
         try:
             # 2. import 插件（触发 @rpc_method 注册），然后契约校验
             module = __import__(wc["module"], fromlist=[wc["factory"]])
@@ -145,9 +195,9 @@ def worker_serve():
             factory = getattr(module, wc["factory"])
             count = args.count or wc.get("count", 1)
             if inspect.iscoroutinefunction(factory):
-                tasks = await factory(session, wc["config"], count, worker_manager=None)
+                tasks = await factory(session, worker_config, count, worker_manager=None)
             else:
-                tasks = factory(session, wc["config"], count, worker_manager=None)
+                tasks = factory(session, worker_config, count, worker_manager=None)
 
             # 4. 发布路由描述符（网关 RouteRegistry 订阅后动态挂载 HTTP 路由）
             publisher = RoutePublisher(session, args.type)
@@ -173,6 +223,9 @@ def worker_serve():
             await stop_event.wait()
             announce_task.cancel()
             await asyncio.gather(announce_task, return_exceptions=True)
+            if hb_task is not None:
+                hb_task.cancel()
+                await asyncio.gather(hb_task, return_exceptions=True)
 
             # 6. 优雅退出：摘除路由 → cleanup → 取消副本任务 → 关闭会话
             publisher.unpublish()
@@ -187,9 +240,16 @@ def worker_serve():
                 t.cancel()
             if tasks:
                 await asyncio.gather(*tasks, return_exceptions=True)
+            if cache is not None:
+                unregister_instance(args.type, cache, args.instance)
             logger.info("[%s] Worker 进程已退出", args.type)
         finally:
             session.close()
+            if cache is not None:
+                try:
+                    cache.close()
+                except Exception:  # noqa: BLE001
+                    pass
 
     try:
         asyncio.run(_run())

src/chongming_worker/runtime.py

@@ -9,13 +9,18 @@ import inspect
 import logging
 import pathlib
 import random
+import socket
+import time
 import tomllib
 import orjson
-import tomli_w
 from typing import Any, Awaitable, Callable, Dict, List, Optional
 
 import zenoh
 
+# 配置缓存（chongming-cache）用于 [workers.config] 与 count 的单一事实来源读写；
+# 由 process.py build_worker_cache（进程插件）或网关 manager.cache（inline）注入，
+# 本模块不自建，仅按注入的 cache 句柄读写缓存键。
+
 from chongming_worker.decorators import (
     RPCMethod,
     get_rpc_methods,
@@ -42,6 +47,7 @@ ADMIN_CMD_RESTART = "restart"
 ADMIN_CMD_STOP = "stop"
 ADMIN_CMD_DRAIN = "drain"
 ADMIN_CMD_CONFIG_SET = "config_set"
+ADMIN_CMD_GET_CONFIG = "get_config"
 ADMIN_CMD_RESCALE = "rescale"
 
 # 优雅排空时，等待在途请求完成的最大时长（秒），超时则强制关闭该副本
@@ -77,65 +83,365 @@ def _plugin_config_path(worker_type: str) -> pathlib.Path:
     return pathlib.Path("config/workers") / worker_type / "config.toml"
 
 
-def _write_plugin_config_section(worker_type: str, updates: Dict[str, Any]) -> Dict[str, Any]:
-    """把 updates 合并写入插件 config.toml 的 [workers.config] 段（原子写回）。
+# ── 单事实来源缓存（chongming-cache）──────────────────────────────
+# 运行时可写配置（[workers.config]）与副本目标数（count）以缓存为准，不再回写 config.toml：
+#   - key = "worker:cm:configs:{type}"：全局一份 [workers.config] dict
+#   - key = "worker:cm:counts:{type}"：全局目标副本数（已有，由网关 scale 统一写）
+# TOML 仅在启动时作为种子（默认模式用 toml 修正缓存；cache 模式缓存缺失时用 toml 引导），
+# 之后读写都在缓存侧进行。缓存可选：网关 inline 用 manager.cache，进程插件用自身 [workers.cache]。
 
-    仅改动 [workers.config] 下的键，保留文件其它段与段落。
-    返回写入后的 [workers.config] 完整 dict（供回显）。
 
-    Raises:
-        ValueError: 包含保留键 / config.toml 缺失 / 无 [workers] 段
+def _configs_cache_key(worker_type: str) -> str:
+    return f"worker:cm:configs:{worker_type}"
+
+
+# ── 全局副本数（chongming-cache 单一事实来源，与网关共享同一套键）──
+# 「count」= 全局目标总量；worker（inline/external 统一）从缓存读在线网关列表，把总量
+# 分摊成本网关应承担的副本数（external 无网关身份则全额承担），并写缓存。
+_SCALE_COUNTS_KEY = "worker:cm:counts:{type}"       # 全局目标总量（int）
+_SCALE_GATEWAYS_KEY = "worker:cm:gateways"          # 在线网关列表：[[id, ts], ...]（JSON）
+_SCALE_QUOTA_KEY = "worker:cm:quota:{type}:{gw}"    # 各网关分摊配额（int）
+_GATEWAY_LEASE = 60
+
+# ── 外部进程实例参与方（多实例全局协同）────────────────────
+# 同一类型的外部进程可能并行运行多个（多台机器/多个容器）。为避免"每个进程都全额起
+# 副本"导致副本翻倍，每个外部进程用稳定的 instance_id 登记到共享缓存，扩缩容时按存活
+# 实例数分摊全局 count（与 inline 按在线网关分摊同构）。租约/心跳复用网关参数。
+_SCALE_INSTANCES_KEY = "worker:cm:instances:{type}"   # 参与方注册：[[inst_id, ts], ...]（JSON）
+_INSTANCE_LEASE = 60
+# 逐实例绝对副本数覆盖表：{instance_id: count}。有记录的实例按绝对值扩缩（精细定制），
+# 无记录的回退到「全局 count 均分」；全局 /scale 会清除此表回到对称分摊。
+_SCALE_INSTCOUNTS_KEY = "worker:cm:instcounts:{type}"
+
+
+def _read_online_gateways(cache) -> List[str]:
+    """从缓存读在线网关 id（按心跳租约清理死网关），排序返回。无缓存返回 []。"""
+    if cache is None:
+        return []
+    now = time.time()
+    deadline = now - _GATEWAY_LEASE
+    alive = []
+    for entry in cache.get(_SCALE_GATEWAYS_KEY) or []:
+        if not isinstance(entry, (list, tuple)) or len(entry) != 2:
+            continue
+        gw_id, ts = entry
+        if isinstance(gw_id, str) and isinstance(ts, (int, float)) and ts >= deadline:
+            alive.append(gw_id)
+    return sorted(set(alive))
+
+
+# ── 外部进程实例参与方助手 ─────────────────────────────────
+def _read_instances_raw(cache, worker_type: str) -> List[List]:
+    """读同类型外部实例原始列表 [[inst_id, ts], ...]；无缓存返回 []。"""
+    if cache is None:
+        return []
+    return list(cache.get(_SCALE_INSTANCES_KEY.format(type=worker_type)) or [])
+
+
+def _prune_instances_raw(cache, worker_type: str, raw: List[List]) -> List[List]:
+    """剔除超过租约未心跳的死实例，写回清理结果以保持列表准确。"""
+    deadline = time.time() - _INSTANCE_LEASE
+    alive = []
+    for entry in raw:
+        if not isinstance(entry, (list, tuple)) or len(entry) != 2:
+            continue
+        iid, ts = entry
+        if isinstance(iid, str) and isinstance(ts, (int, float)) and ts >= deadline:
+            alive.append([iid, ts])
+    if len(alive) != len(raw):
+        cache[_SCALE_INSTANCES_KEY.format(type=worker_type)] = alive
+    return alive
+
+
+def _read_instances(cache, worker_type: str) -> List[str]:
+    """读同类型存活外部实例 id（按租约清理死实例、保序去重），排序返回。"""
+    raw = _prune_instances_raw(cache, worker_type, _read_instances_raw(cache, worker_type))
+    return sorted({iid for iid, _ in raw})
+
+
+# 进程内显式设定的实例身份（process.py 启动时注入）；None 则回退 env/hostname。
+# 与 register_instance/逐一绝对扩缩共用同一 id，保证命中判断一致。
+_current_instance_id: Optional[str] = None
+
+
+def set_current_instance_id(iid: Optional[str]) -> None:
+    """设定当前外部 worker 进程的实例身份（应与 register_instance 使用的 id 一致）。"""
+    global _current_instance_id
+    _current_instance_id = iid
+
+
+def current_instance_id() -> str:
+    """外部 worker 进程的稳定全局实例标识。
+
+    优先取进程内显式设定的身份（process.py --instance）；其次 env
+    CHONGMING_WORKER_INSTANCE（编排时显式给定、跨重启稳定）；缺省回退
+    "{hostname}:{pid}"——同一台机器上并行多个未指定身份的实例时，凭 pid 区分，
+    避免它们被误并为一个参与方、导致副本翻倍或 UI 只显示一个实例。
+    注意：缺省 id 随进程重启变化；若需跨重启稳定且被逐实例精细扩缩持续命中，
+    请显式设 --instance / CHONGMING_WORKER_INSTANCE。
     """
-    bad = [k for k in updates if k in CONFIG_FORBIDDEN_KEYS]
+    if _current_instance_id:
+        return _current_instance_id
+    iid = os.environ.get("CHONGMING_WORKER_INSTANCE")
+    if not iid:
+        iid = f"{socket.gethostname()}:{os.getpid()}"
+    return iid
+
+
+def _write_instance_heartbeats(cache, worker_type: str, iid: str, *, keep_self: bool) -> None:
+    """按 keep_self 保留/移除本实例，并刷新本实例心跳 ts，写回参与方列表。"""
+    if cache is None:
+        return
+    now = time.time()
+    raw = _prune_instances_raw(cache, worker_type, _read_instances_raw(cache, worker_type))
+    alive = [[i, ts] for i, ts in raw if i != iid]
+    if keep_self:
+        alive.append([iid, now])
+    cache[_SCALE_INSTANCES_KEY.format(type=worker_type)] = alive
+
+
+def register_instance(worker_type: str, cache, instance_id: Optional[str] = None) -> None:
+    """外部进程启动时登记自身为参与方（带心跳时间戳）。进程退出后租约过期自动清除。"""
+    _write_instance_heartbeats(
+        cache, worker_type, instance_id or current_instance_id(), keep_self=True
+    )
+
+
+def heartbeat_instance(worker_type: str, cache, instance_id: Optional[str] = None) -> None:
+    """周期性刷新本实例心跳 ts，并剔除超租约死实例。"""
+    _write_instance_heartbeats(
+        cache, worker_type, instance_id or current_instance_id(), keep_self=True
+    )
+
+
+def unregister_instance(worker_type: str, cache, instance_id: Optional[str] = None) -> None:
+    """关闭时从参与方列表移除本实例。"""
+    _write_instance_heartbeats(
+        cache, worker_type, instance_id or current_instance_id(), keep_self=False
+    )
+
+
+# ── 逐实例绝对副本数（精细扩缩）────────────────────────────
+def _read_instance_counts(cache, worker_type: str) -> Dict[str, int]:
+    """读逐实例绝对副本数覆盖表 {instance_id: count}；无缓存/非对象返回 {}。"""
+    if cache is None:
+        return {}
+    v = cache.get(_SCALE_INSTCOUNTS_KEY.format(type=worker_type))
+    return dict(v) if isinstance(v, dict) else {}
+
+
+def _write_own_instance_count(
+    cache, worker_type: str, count: int, instance_id: Optional[str] = None
+) -> None:
+    """把本实例的绝对副本数写入覆盖表（精细扩缩；进程重启后依旧生效）。"""
+    if cache is None:
+        return
+    iid = instance_id or current_instance_id()
+    key = _SCALE_INSTCOUNTS_KEY.format(type=worker_type)
+    counts = dict(cache.get(key) or {})
+    counts[iid] = int(count)
+    cache[key] = counts
+
+
+def _clear_instance_counts(cache, worker_type: str) -> None:
+    """清除所有逐实例绝对覆盖，回到「全局 count 均分」对称语义。"""
+    if cache is None:
+        return
+    cache[_SCALE_INSTCOUNTS_KEY.format(type=worker_type)] = {}
+
+
+def _own_instance_count(
+    cache, worker_type: str, instance_id: Optional[str] = None
+) -> Optional[int]:
+    """读本实例在覆盖表中的绝对副本数；未定制返回 None（回退全局均分）。"""
+    if cache is None:
+        return None
+    counts = _read_instance_counts(cache, worker_type)
+    iid = instance_id or current_instance_id()
+    v = counts.get(iid)
+    return int(v) if isinstance(v, (int, float)) else None
+
+
+async def run_instance_heartbeat(
+    worker_type: str, cache, *,
+    interval: float = 15.0, instance_id: Optional[str] = None,
+) -> None:
+    """外部进程心跳常驻循环。启动后 create_task；无需显式反注册，租约过期自清。"""
+    if cache is None:
+        return
+    while True:
+        try:
+            heartbeat_instance(worker_type, cache, instance_id)
+        except Exception as e:  # noqa: BLE001
+            logger.warning("[instance] %s 心跳失败: %s", worker_type, e)
+        await asyncio.sleep(interval)
+
+
+def distribute_quota(cache, worker_type: str, total: int, gw_id: Optional[str]) -> int:
+    """在「count=全局总量」语义下，算当前主体应承担的副本数。
+
+    gw_id 非空（inline，具备网关身份）→ 按在线网关 id 排序均分，余量按序补前若干。
+    gw_id 为空（external 独立进程）→ 按同一类型「存活实例 N」分摊（余量补靠前实例），
+    避免每个外部进程都全额起副本导致翻倍；无参与方时兜底全额。
+    结果写回 cache（observability，与网关共享键）。
+    """
+    total = max(0, int(total))
+    if gw_id:
+        gws = _read_online_gateways(cache)
+        if not gws:
+            quota = total
+        else:
+            base, rem = divmod(total, len(gws))
+            idx = gws.index(gw_id) if gw_id in gws else 0
+            quota = base + (1 if idx < rem else 0)
+        quota_key = _SCALE_QUOTA_KEY.format(type=worker_type, gw=gw_id)
+    else:
+        iid = current_instance_id()
+        inst = _read_instances(cache, worker_type)
+        if not inst:
+            quota = total
+        else:
+            base, rem = divmod(total, len(inst))
+            idx = inst.index(iid) if iid in inst else 0
+            quota = base + (1 if idx < rem else 0)
+        quota_key = _SCALE_QUOTA_KEY.format(type=worker_type, gw=iid)
+    if cache is not None:
+        cache[quota_key] = quota
+    return quota
+
+
+def resolve_worker_count(
+    worker_type: str, default: int, cache, mode: str, gw_id: Optional[str] = None
+) -> int:
+    """启动时解析本进程应承载的副本数（count 收归 worker 后由 WorkerPool 调用）。
+
+      计算全局总量：toml（默认）用 default 覆盖写回缓存；cache 模式缓存有则用、缺则 default 引导写回。
+      再按在线网关分摊为写给本进程的副本数（external/no gw 按存活实例分摊；
+      外部实例若在 instcounts 有绝对覆盖，直接采用该值）。
+    """
+    if cache is None:
+        return int(default)
+    # 逐实例绝对覆盖（精细扩缩）：外部进程若被单独定制过，重启后应恢复到该绝对数，
+    # 而非被全局均分覆盖。inline（有网关身份）不适用。
+    if gw_id is None:
+        own = _own_instance_count(cache, worker_type)
+        if own is not None:
+            return own
+    total = max(0, int(default))
+    if mode == "cache":
+        v = cache.get(_SCALE_COUNTS_KEY.format(type=worker_type))
+        if isinstance(v, int) and v >= 0:
+            total = v
+    cache[_SCALE_COUNTS_KEY.format(type=worker_type)] = total
+    return distribute_quota(cache, worker_type, total, gw_id)
+
+
+# 当前进程内建立的缓存（key=worker_type）。进程插件在 process.py 启动时注入，
+# 供 admin（config_set/get_config/rescale）读写缓存，无需改 create_workers 签名。
+_worker_caches: Dict[str, Any] = {}
+
+# 各 worker 类型使用的「配置优先来源」模式（toml 默认 / cache）。
+# 进程插件在 build_worker_cache 时登记；网关 inline 走 manager 的 _launch_worker_count。
+_worker_modes: Dict[str, str] = {}
+
+
+def set_worker_cache(worker_type: str, cache) -> None:
+    """为某 worker 类型登记其配置缓存句柄（进程插件自建缓存后调用）。"""
+    _worker_caches[worker_type] = cache
+
+
+def set_worker_mode(worker_type: str, mode: str) -> None:
+    """登记某 worker 类型的配置优先来源模式（"toml" 默认 / "cache"）。"""
+    _worker_modes[worker_type] = ("cache" if str(mode).lower() == "cache" else "toml")
+
+
+def _resolve_cache(worker_type: str, worker_manager=None):
+    """取某 worker 类型的配置缓存：显式登记 > 网关 manager.cache > 无。"""
+    cache = _worker_caches.get(worker_type)
+    if cache is not None:
+        return cache
+    if worker_manager is not None:
+        return getattr(worker_manager, "cache", None)
+    return None
+
+
+def _resolve_mode(worker_type: str, worker_manager=None) -> str:
+    """取某 worker 类型的配置优先来源模式：网关 inline 用 manager > 进程插件登记 > toml。"""
+    if worker_manager is not None:
+        return "cache" if str(getattr(worker_manager, "_launch_worker_count", "toml")).lower() == "cache" else "toml"
+    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)
+    try:
+        with open(_plugin_config_path(worker_type), "rb") as f:
+            data = tomllib.load(f)
+        workers = data.get("workers") or {}
+        cfg = workers.get("config") if isinstance(workers, dict) else None
+        return dict(cfg) if isinstance(cfg, dict) else {}
+    except Exception:  # noqa: BLE001 - toml 缺失/损坏按空处理
+        return {}
+
+
+def _persist_worker_config_section(cache, worker_type: str, updates: Dict[str, Any]) -> Dict[str, Any]:
+    """把 updates 合并写入该类型 [workers.config]（缓存侧，全局唯一）。
+
+    仅改动 [workers.config] 下的键，不再写回 config.toml（取消回写机制）。
+    缓存缺失（未配置缓存后端）时不持久化，仅返回合并后的配置供热更/重启使用。
+    """
+    bad = sorted(k for k in updates if k in CONFIG_FORBIDDEN_KEYS)
     if bad:
         raise ValueError(f"禁止修改保留配置键: {bad}（路由/加载控制配置除外，不在此功能范围）")
 
-    path = _plugin_config_path(worker_type)
-    with open(path, "rb") as f:
-        data = tomllib.load(f)
-
-    workers = data.get("workers")
-    if not isinstance(workers, dict):
-        raise ValueError(f"{path} 缺少 [workers] 节")
-
-    cfg = workers.setdefault("config", {})
-    if not isinstance(cfg, dict):
-        raise ValueError(f"{path} 的 [workers.config] 不是表，无法更新")
-    cfg.update(updates)
-
-    # 原子写回：临时文件 + rename，避免写一半被读
-    tmp = path.with_suffix(".toml.tmp")
-    with open(tmp, "w", encoding="utf-8") as f:
-        f.write(tomli_w.dumps(data))
-    os.replace(tmp, path)
-    return dict(cfg)
+    current = _read_worker_config_section(cache, worker_type)
+    merged = {**current, **updates}
+    if cache is not None:
+        cache[_configs_cache_key(worker_type)] = merged
+    return dict(merged)
 
 
-def _write_plugin_count(worker_type: str, count: int) -> int:
-    """把某类型的副本目标数写入插件 config.toml 的 [workers].count（原子写回）。
+def resolve_runtime_config(
+    worker_type: str,
+    config: Dict[str, Any],
+    cache,
+    mode: str = "toml",
+) -> Dict[str, Any]:
+    """启动时按模式对齐 [workers.config] 的单一事实来源，返回覆盖后的运行时配置。
 
-    由 rescale 指令的处理副本在**其所在进程的工作目录**下执行，因此 inline（网关进程）
-    与 external（独立进程/容器、配置目录已挂载）都能持久化到自己那侧的配置，重启后保持。
-    返回写回的计数值。
-
-    Raises:
-        ValueError: 文件缺失 / 无 [workers] 节
+    缓存配置（[workers.config]）在传入插件前被展平进运行配置顶层（与 _discover_plugins 一致）。
+      - "toml"（默认）：用 toml 的 [workers.config] 覆盖并修正缓存（缓存缺失时写入）；
+      - "cache"：缓存有值则用缓存覆盖；缓存缺失才用 toml 引导并同步到缓存。
+    cache 为 None 时保持传入的 config 原样（不启用缓存语义）。
     """
-    path = _plugin_config_path(worker_type)
-    with open(path, "rb") as f:
+    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 = {}
 
-    workers = data.get("workers")
-    if not isinstance(workers, dict):
-        raise ValueError(f"{path} 缺少 [workers] 节")
+    if cache is None:
+        # 无缓存：保持 toml 种子（经典行为，无回写）
+        return {**config, **toml_cfg}
 
-    workers["count"] = int(count)
-
-    tmp = path.with_suffix(".toml.tmp")
-    with open(tmp, "w", encoding="utf-8") as f:
-        f.write(tomli_w.dumps(data))
-    os.replace(tmp, path)
-    return int(count)
+    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 覆盖并修正缓存
+    toml_cfg = dict(toml_cfg)
+    cache[key] = toml_cfg
+    return {**config, **toml_cfg}
 
 
 def _merge_runtime_config(config: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]:
@@ -161,6 +467,8 @@ class WorkerController:
         self.epoch = 0
         self.exit_request: str = ""
         self.current_worker_id: str = ""
+        # 该副本所属类型的配置缓存句柄（由 WorkerPool 注入，供 admin 读写缓存）
+        self.cache = None
         # 配置变更后待应用的新运行时配置（由 config_set 设置，重启时生效）
         self.pending_config: Optional[Dict[str, Any]] = None
         self.pending_updates: Optional[Dict[str, Any]] = None
@@ -224,22 +532,44 @@ async def _handle_admin_command(
         controller.exit_request = ADMIN_CMD_STOP
         controller.draining = True
         logger.info("[Worker] %s 收到优雅排空指令（将隔离新请求并等待在途请求完成）", worker_id)
+    elif cmd == ADMIN_CMD_GET_CONFIG:
+        # 查询该类型的 [workers.config]（缓存优先，无缓存回退读 toml 种子）。
+        cache = getattr(controller, "cache", None)
+        cfg = _read_worker_config_section(cache, worker_type)
+        query.reply(
+            key_expr=query.key_expr,
+            payload=orjson.dumps({
+                "status": "ok",
+                "cmd": cmd,
+                "worker_type": worker_type,
+                "worker_id": worker_id,
+                "config": cfg,
+                "source": "cache" if cache is not None else "toml",
+            }),
+        )
+        return
     elif cmd == ADMIN_CMD_CONFIG_SET:
         updates = raw.get("config")
         if not isinstance(updates, dict) or not updates:
             setup_error = "config_set 需要 data.config 字段（非空对象）"
         else:
             try:
-                # 仅 index0 落盘写回，其余副本不落盘（同一进程多副本/共享文件避免并发写）
+                # 写回改为缓存侧（不再回写 config.toml）。缓存为全局唯一事实来源，
+                # 仅 index0 执行写+热重载，避免多副本并发 read-modify-write；其余副本只应用。
+                cache = getattr(controller, "cache", None)
                 if controller.index == 0:
-                    new_section = _write_plugin_config_section(worker_type, updates)
+                    new_section = _persist_worker_config_section(cache, worker_type, updates)
                     applied_updates = {
                         k: v for k, v in updates.items()
                         if k not in CONFIG_FORBIDDEN_KEYS
                     }
+                    persist_note = (
+                        f"{len(new_section)} 键已写缓存（{_configs_cache_key(worker_type)}）"
+                        if cache is not None else "缓存未启用，仅内存热更"
+                    )
                     logger.info(
-                        "[Worker] %s 已将配置写入 %s: %s",
-                        worker_id, _plugin_config_path(worker_type), new_section,
+                        "[Worker] %s 已将配置写入缓存 %s: %s",
+                        worker_id, _configs_cache_key(worker_type), new_section,
                     )
                 else:
                     applied_updates = {
@@ -264,12 +594,16 @@ async def _handle_admin_command(
                     "[Worker] %s 收到配置变更指令，将以新配置重启（%d 个键）",
                     worker_id, len(applied_updates),
                 )
-            except Exception as e:  # noqa: BLE001 - 写回/校验失败按指令失败返回
+            except Exception as e:  # noqa: BLE001 - 校验/写缓存失败按指令失败返回
                 setup_error = f"配置写入失败: {e}"
                 logger.error("[Worker] %s %s", worker_id, setup_error)
     elif cmd == ADMIN_CMD_RESCALE:
-        # 调整当前进程内该类型的副本槽位数（缩容走优雅排空）。作用于本进程 Worker 池，
-        # 不退出本副本自身；跨进程（external）指令经 Zenoh 到达其容器进程内的指定副本 → 其进程池扩缩。
+        # 调整本进程内该类型的副本槽位数（缩容走优雅排空）。支持两种语义：
+        #   - 逐实例绝对（data.instance_id + data.absolute=true 且命中本实例 id）：
+        #     直接把本实例池设为 data.scale，并写入 instcounts 覆盖（精细扩缩，
+        #     进程重启后仍生效）；目标是其它实例的指令被安全忽略，避免误缩。
+        #   - 全局（无 instance_id）：先清除所有逐实例覆盖回到「均分」对称语义，再把
+        #     全局 target 按在线网关/存活实例分摊为本进程应承载数并写回缓存。
         try:
             target = int(raw.get("scale", -1))
         except (TypeError, ValueError):
@@ -281,18 +615,47 @@ async def _handle_admin_command(
             if pool is None:
                 setup_error = f"当前进程无 {worker_type} 类型的 Worker 池，无法缩放"
             else:
-                asyncio.ensure_future(pool.set_target(target))
+                cache = getattr(controller, "cache", None)
+                gw_id = getattr(pool.worker_manager, "_gateway_id", None) \
+                    if getattr(pool, "worker_manager", None) is not None else None
+                total = max(0, target)
+                target_inst = raw.get("instance_id")
+                if target_inst and raw.get("absolute"):
+                    # 逐实例绝对扩缩：只对命中本实例生效
+                    if target_inst == current_instance_id():
+                        _write_own_instance_count(cache, worker_type, total, target_inst)
+                        app_count = total
+                        asyncio.ensure_future(pool.set_target(app_count))
+                        persist_note = f"实例 {target_inst} 绝对 count={total}"
                         logger.info(
-                    "[Worker] %s 收到缩放指令，目标副本数=%d（当前池=%d）",
-                    worker_id, target, len(pool._slots),
+                            "[Worker] %s 实例级缩放 count=%d（当前池=%d）",
+                            worker_id, total, len(pool._slots),
+                        )
+                    else:
+                        # 目标为其它实例：不缩放本池，安全忽略
+                        app_count = len(pool._slots)
+                        persist_note = f"实例级缩放目标 {target_inst} 非本实例，忽略"
+                        logger.info(
+                            "[Worker] %s 忽略其它实例的缩放指令（target=%s）",
+                            worker_id, target_inst,
+                        )
+                else:
+                    # 全局对称：清除逐实例覆盖，回到均分
+                    _clear_instance_counts(cache, worker_type)
+                    if cache is not None:
+                        cache[_SCALE_COUNTS_KEY.format(type=worker_type)] = total
+                        app_count = distribute_quota(cache, worker_type, total, gw_id)
+                        persist_note = (
+                            f"全局 count={total}，本进程应用={app_count} 已写缓存"
+                        )
+                    else:
+                        app_count = total
+                        persist_note = "缓存未启用，count 仅内存缩放"
+                    asyncio.ensure_future(pool.set_target(app_count))
+                    logger.info(
+                        "[Worker] %s 收到缩放指令，全局目标=%d，本进程应用=%d（当前池=%d）",
+                        worker_id, total, app_count, len(pool._slots),
                     )
-                # 持久化目标数量到本进程侧配置（重启后保持），失败不影响本次内存缩放
-                try:
-                    _write_plugin_count(worker_type, target)
-                    persist_note = f"count={target} 已持久化"
-                except Exception as e:  # noqa: BLE001
-                    persist_note = f"count 持久化失败: {e}"
-                    logger.warning("[Worker] %s %s", worker_id, persist_note)
     else:
         query.reply(
             key_expr=query.key_expr,
@@ -301,6 +664,7 @@ async def _handle_admin_command(
                 "worker_id": worker_id,
                 "index": controller.index,
                 "epoch": controller.epoch,
+                "instance_id": current_instance_id(),
                 "status": "running",
             }),
         )
@@ -682,6 +1046,10 @@ def create_workers(
             session, worker_type, config, worker_manager=worker_manager, reload_cb=reload_cb
         )
         _register_process_pool(worker_type, pool)
+    # count 收归 worker：按缓存全局总量与分摊后启动本进程副本数（inline 分摊/external 全额）
+    if pool.cache is not None:
+        gw_id = getattr(worker_manager, "_gateway_id", None) if worker_manager is not None else None
+        count = resolve_worker_count(worker_type, count, pool.cache, pool.mode, gw_id)
     return pool.start(count)
 
 
@@ -710,9 +1078,17 @@ class WorkerPool:
     ):
         self.session = session
         self.worker_type = worker_type
-        self.config = config or {}
+        self.config = dict(config) if isinstance(config, dict) else {}
         self.worker_manager = worker_manager
         self.reload_cb = reload_cb
+        # 本类型配置缓存句柄：进程插件经 set_worker_cache 注入 > 网关 manager.cache > 无
+        self.cache = _resolve_cache(worker_type, worker_manager)
+        self.mode = _resolve_mode(worker_type, worker_manager)
+        # [workers.config] 单一事实来源：启动时按模式解析进运行配置（不再由网关/进程入口直写缓存）
+        if self.cache is not None:
+            self.config = resolve_runtime_config(
+                worker_type, self.config, self.cache, self.mode
+            )
         # index -> controller / task（仅记录当前活动槽位；drain 结束后由 done 回调移除）
         self._slots: Dict[int, WorkerController] = {}
         self._tasks: Dict[int, asyncio.Task] = {}
@@ -724,6 +1100,7 @@ class WorkerPool:
         index = self._next_index
         self._next_index += 1
         controller = WorkerController(self.worker_type, index)
+        controller.cache = self.cache
         task = asyncio.create_task(
             _managed_worker(
                 self.session, self.worker_type, controller, self.config,
@@ -794,4 +1171,6 @@ __all__ = [
     "create_workers",
     "WorkerPool",
     "get_process_pool",
+    "set_worker_cache",
+    "resolve_runtime_config",
 ]