diff --git a/.gitignore b/.gitignore
index 37c5793..4b8930c 100644
--- a/.gitignore
+++ b/.gitignore
@@ -41,4 +41,7 @@ config
# lib
lib
auto.crt
-auto.key
\ No newline at end of file
+auto.key
+
+*.pem
+*.key
\ No newline at end of file
diff --git a/README.md b/README.md
index 42838f9..2aad3df 100644
--- a/README.md
+++ b/README.md
@@ -11,7 +11,7 @@
| 特性 | 说明 |
| --------------------- | -------------------------------------------------------------------- |
| **FastAPI + Uvicorn** | 高性能异步 Web 框架，自动生成 OpenAPI 文档 |
-| **Zenoh 服务发现** | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试 |
+| **Zenoh 服务发现** | 基于 liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 实例定向 + 超时重试 |
| **MQTT 集成** | amqtt Broker（内嵌或外部），MQTT ↔ Worker 双向桥接 |
| **动态路由生成** | 扫描 `config/workers/*/*.toml` 自动注册 HTTP 路由 |
| **Worker 框架** | `@rpc_method` 装饰器 + `Plugin` 基类统一生命周期 + Pydantic 模型绑定 |
@@ -194,7 +194,7 @@ graph TB
MQTT[MQTT 客户端] -->|chongming/+/+| BRIDGE[MQTT Bridge
纯客户端模式]
REG[RouteRegistry
订阅 chongming/routes/**] -->|动态挂载/摘除 HTTP 路由| API
- API --> WM[WorkerManager
随机负载均衡 + 超时重试]
+ API --> WM[WorkerManager
随机负载均衡 / 实例定向 + 超时重试]
BRIDGE --> WM
subgraph GW[网关进程 chongming]
@@ -328,6 +328,30 @@ worker///__admin__
- **REST / UI**：`POST /admin/workers//scale_instance`（body `{instance_id, count}`）下发；`GET /admin/workers//instances` 供前端按实例分组展示当前副本数。前端用「调整数量」（全局对称）和「逐实例精细扩缩」表格（实例 id / 模式 / 当前副本数 / 目标 + 应用）两个入口。
- **协调前提**：与外部广播缩放一致，需共享 `[cache]`，且外部进程用稳定的 `--instance` / `CHONGMING_WORKER_INSTANCE`（缺省 `{hostname}:{pid}` 虽唯一但重启会变，跨重启持续命中请显式指定）。
+### 实例定向调用（把请求固定到某个进程实例）
+
+默认情况下，同 type 的多个副本（含 inline 与 external 进程）逐请求**随机负载均衡**（`random.choice`），这是绝大多数无状态插件想要的。若某插件持有**只存在于某个进程本地**的状态（本地 SQLite/文件、串口或其他设备连接），需要请求固定打到那个进程上，可用**实例定向**：
+
+- **指定方式（三种入口，语义一致）**：
+ - HTTP：请求头 `X-Chongming-Instance: <实例id>`
+ - MQTT：信封载荷顶层加控制字段 `target_instance`（网关读后**摘除**，业务 side 看不到该字段）
+ - 程序内调用：`worker_manager.call(...)` / `call_raw(...)` / `call_stream(...)` 的 `instance_id=` 参数（`utils/request` 的独立 Client 不支持定向）
+- **实例 id 取值**：worker 侧按 `--instance` → `CHONGMING_WORKER_INSTANCE` → `CHONGMING_GATEWAY_ID` → `{hostname}:{pid}` 的顺序解析（`current_instance_id()`）；网关侧定向寻址 inline 副本时用网关自身实例 id（`CHONGMING_GATEWAY_ID`，缺省 hostname）。两侧缺省值在容器环境（K8s 给网关进程注入 `CHONGMING_GATEWAY_ID`）下**自动一致**，无需额外配置；裸机同一台跑多个网关时请显式指定，否则默认实例 id 里的 pid 不同会导致对不上。人可读的取值列表见 `GET /api/v1/admin/workers//instances`。
+- **粒度**：定向到**进程实例**；若该实例内有多个副本，请求仍在这几个副本之间随机均衡（即“实例固定、副本仍均衡”）。
+- **实例离线时严格报错**：返回 HTTP 503 并附已识别实例列表，**不会静默回退到其它实例**——回退会让粘滞状态的插件跑在错误副本上，产生难查的数据错乱。MQTT 同理（响应主题收到 `{"error": ...}`）。
+- **重试不跨实例**：调用失败重试只在同实例内换副本；实例内已无可换副本时立即失败，不重复打同一个副本。流式端点本身不重试，实例离线即失败。
+- **缓存按实例隔离**：定向且 `cacheable=true` 时缓存键带实例前缀，避免取到其它实例的结果。
+- **身份来源与开销**：副本的实例身份取自其 admin status 回复的 `instance_id`（Python/Rust 两种运行时都会返回），网关**按需查询并缓存**——未使用定向功能时零额外开销，不定向的请求路径与行为完全不变；首次定向到某实例时会对该类型的新副本各发一次轻量 status 查询（每个副本只查一次，inline 副本无需查询）。副本重启后 id 变化，会重新查询一次。
+- **版本要求**：worker 端需运行带 `instance_id` 上报的运行时（当前 Python/Rust SDK 均满足）；读不出身份的副本会被排除在定向之外（不定向的调用不受影响）。
+
+```bash
+# 等价于把这次请求固定到实例 edge-01 的那个进程
+curl -H 'X-Chongming-Instance: edge-01' http://127.0.0.1:8000/api/v1/demo/state
+
+# MQTT：载荷顶层带 target_instance（网关转发前会摘掉它）
+# topic: chongming/demo/state payload: {"meta":{}, "data":{}, "target_instance":"edge-01"}
+```
+
### inline worker 宿主级领导选举（gunicorn 多 worker 兼容）
inline worker 的生命周期交给**基于分布式锁的宿主级领导选举**（取代“每 gunicorn worker 各自拉起”），key `inline_owner:` 按网关隔离：
diff --git a/src/chongming/app/__init__.py b/src/chongming/app/__init__.py
index e2012d1..21a877e 100644
--- a/src/chongming/app/__init__.py
+++ b/src/chongming/app/__init__.py
@@ -20,10 +20,9 @@ from fastapi.middleware.cors import CORSMiddleware
from chongming_config import get_config
from .core.logger import get_logger
-# from .core.static_files import SVFSStaticFiles
from sqlite_vfs.contrib.fastapi_static import SVFSStaticFiles
from .core import lifespan_setup, call_stats
-from .routes import storage_router, stats_router, worker_admin_router, auth_router
+from .routes import stats_router, worker_admin_router, auth_router
config = get_config()
@@ -311,7 +310,6 @@ if vfs_db_path := config[env].get("file_system", {}).get("path", None):
static_files_handler = SVFSStaticFiles(vfs_db_path=vfs_db_path)
app.mount("/static", static_files_handler, name="static")
-app.include_router(storage_router)
app.include_router(stats_router)
app.include_router(worker_admin_router)
app.include_router(auth_router)
diff --git a/src/chongming/app/core/mqtt_bridge.py b/src/chongming/app/core/mqtt_bridge.py
index e6112c7..25d1f03 100644
--- a/src/chongming/app/core/mqtt_bridge.py
+++ b/src/chongming/app/core/mqtt_bridge.py
@@ -169,6 +169,15 @@ class MQTTBridge:
except orjson.JSONDecodeError:
return
+ # 实例定向控制字段：与 HTTP 的 X-Chongming-Instance 头同义，指定请求发往的
+ # worker 进程实例。必须在下面归一化之前取出——归一化只保留 meta/data 两键，
+ # 之后控制字段就不在载荷里了。
+ target_instance: Optional[str] = None
+ if isinstance(pl, dict):
+ ti = pl.pop("target_instance", None)
+ if isinstance(ti, str) and ti.strip():
+ target_instance = ti.strip()
+
# 信封协议: {"meta": {...}, "data": {...}}
# 兼容非信封载荷（如第三方设备原始报文）: 整包视为业务数据、meta 置空，
# 与 worker 运行时 runtime.py 的容错语义一致（method 默认 GET）——
@@ -194,7 +203,7 @@ class MQTTBridge:
for ct, compiled in self._compiled:
matched = compiled.match(t)
if matched:
- await self._route_custom(ct, matched, pl, t)
+ await self._route_custom(ct, matched, pl, t, target_instance)
return
# 2. 内置接口 chongming/{worker}/{operation}
@@ -204,7 +213,7 @@ class MQTTBridge:
wt, op = parts[1], parts[2]
subj = f"{wt}.{op}"
rt = f"{MQTT_TOPIC_PREFIX}/{wt}/{op}/response"
- await self._route(wt, subj, pl, rt)
+ await self._route(wt, subj, pl, rt, target_instance)
def _fill(self, template: str, matched: re.Match) -> str:
"""用匹配到的通配符值填充模板中的 {1} {2}... 占位符"""
@@ -214,7 +223,8 @@ class MQTTBridge:
template = template.replace("{" + str(i) + "}", g)
return template
- async def _route_custom(self, ct: dict, matched: re.Match, pl: dict, topic: str):
+ async def _route_custom(self, ct: dict, matched: re.Match, pl: dict, topic: str,
+ target_instance: Optional[str] = None):
worker = ct.get("worker", "")
# subject 默认 "."，与内置 <插件>.<操作> 约定一致
subject_tpl = ct.get("subject")
@@ -225,15 +235,22 @@ class MQTTBridge:
# response_topic 默认 "<实际匹配主题>/response"
rt_tpl = ct.get("response_topic")
rt = self._fill(rt_tpl, matched) if rt_tpl else f"{topic}/response"
- await self._route(worker, subject, pl, rt)
+ await self._route(worker, subject, pl, rt, target_instance)
- async def _route(self, wt: str, subj: str, pl: dict, rt: str):
+ async def _route(self, wt: str, subj: str, pl: dict, rt: str,
+ target_instance: Optional[str] = None):
+ """把一次 MQTT 请求转成 worker RPC。
+
+ target_instance 为实例定向目标（来自载荷顶层的 target_instance 控制字段，
+ 已在 _handle 归一化前摘除）；缺省时跨实例随机均衡，行为与以往一致。
+ """
logger.info(f"MQTT -> {subj}")
meta = pl.get("meta") or {}
try:
r = await self._wm.call(worker_type=wt, subject=subj, payload=pl,
timeout=float(meta.get("timeout", 30)),
- cacheable=bool(meta.get("cacheable", False)))
+ cacheable=bool(meta.get("cacheable", False)),
+ instance_id=target_instance or None)
if r is not None:
await self._pub(rt, r)
except RuntimeError as e:
diff --git a/src/chongming/app/core/route_generator.py b/src/chongming/app/core/route_generator.py
index 21924f5..e9b2236 100644
--- a/src/chongming/app/core/route_generator.py
+++ b/src/chongming/app/core/route_generator.py
@@ -24,6 +24,21 @@ from .call_stats import spawn, record_call
logger = get_logger("route_generator")
+# 实例定向请求头：给定后该请求固定发往指定 worker 进程实例（同一实例下的多个副本
+# 之间仍随机均衡），不参与跨实例的随机负载均衡。取值 = 实例身份（外部进程启动参数
+# --instance / CHONGMING_WORKER_INSTANCE；inline 副本为网关自身实例 id）。
+# 未携带该头时行为与以往完全一致（全类型随机均衡）。
+INSTANCE_HEADER = "X-Chongming-Instance"
+
+
+def _instance_from_request(request: Request) -> Optional[str]:
+ """从请求头读取定向实例 id；空串/未携带视为不定向。"""
+ value = request.headers.get(INSTANCE_HEADER)
+ if value:
+ value = value.strip()
+ return value or None
+
+
# 简单类型映射
TYPE_MAP: Dict[str, Any] = {
"float": float,
@@ -348,6 +363,7 @@ def _make_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_t
payload=payload_bytes,
timeout=timeout,
cacheable=cacheable,
+ instance_id=_instance_from_request(request),
)
duration_ms = (time.perf_counter() - start) * 1000
_spawn_call_record(
@@ -409,6 +425,7 @@ def _make_raw_body_endpoint(worker_manager: WorkerManager, rc: RouteConfig, work
payload=payload,
timeout=timeout,
cacheable=cacheable,
+ instance_id=_instance_from_request(request),
)
duration_ms = (time.perf_counter() - start) * 1000
_spawn_call_record(
@@ -472,6 +489,7 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
payload=payload_bytes,
timeout=timeout,
cacheable=cacheable,
+ instance_id=_instance_from_request(request),
)
duration_ms = (time.perf_counter() - start) * 1000
_spawn_call_record(
@@ -515,6 +533,7 @@ def _make_query_endpoint(worker_manager: WorkerManager, rc: RouteConfig, worker_
payload=payload_bytes,
timeout=timeout,
cacheable=cacheable,
+ instance_id=_instance_from_request(request),
)
duration_ms = (time.perf_counter() - start) * 1000
_spawn_call_record(
@@ -648,6 +667,7 @@ def _make_stream_endpoint(
payload=payload,
timeout=timeout,
idle_timeout=idle_timeout,
+ instance_id=_instance_from_request(request),
)
except RuntimeError as e:
duration_ms = (time.perf_counter() - start) * 1000
diff --git a/src/chongming/app/core/worker_manager.py b/src/chongming/app/core/worker_manager.py
index b763b73..1a3e178 100644
--- a/src/chongming/app/core/worker_manager.py
+++ b/src/chongming/app/core/worker_manager.py
@@ -112,6 +112,15 @@ class WorkerManager:
self._launch_worker_count = launch_worker_count
self._workers_cache: Dict[str, List[str]] = {} # type -> list of worker key_exprs
self._local_workers: Dict[str, List[str]] = {} # type -> list of locally registered workers
+ # ── 实例级定向（把请求固定到某个进程实例而非随机负载均衡）──
+ # 副本经 liveliness 只暴露 worker//，实例身份取自副本 admin status
+ # 的 instance_id（Python/Rust 运行时均返回），故定向索引按需查询并缓存：
+ # _instance_index[type][instance_id] = [worker//, ...]
+ # _instance_unknown[type] = [已查询但身份不可得的副本 key, ...]
+ # 只缓存已查询过的副本；每次按需刷新时整表重建并对新副本补查询、对离线副本剪枝。
+ self._instance_index: Dict[str, Dict[str, List[str]]] = {}
+ self._instance_unknown: Dict[str, List[str]] = {}
+ self._instance_locks: Dict[str, asyncio.Lock] = {}
self._lock = asyncio.Lock()
self._refresh_interval = 15 # seconds
# 当前网关心跳标识：优先环境变量（多实例需唯一，尤其同宿主多网关须显式指定），
@@ -406,23 +415,112 @@ class WorkerManager:
logger.error(f"刷新 worker 列表失败: {e}")
await asyncio.sleep(self._refresh_interval)
- async def get_worker(self, worker_type: str) -> Optional[str]:
- """获取一个可用的 worker 地址（简单随机负载均衡）
+ def _all_workers(self, worker_type: str) -> List[str]:
+ """该类型当前全部候选副本（liveliness 列表优先，启动初期回退本地注册表）。"""
+ workers = list(self._workers_cache.get(worker_type, []))
+ if not workers:
+ workers = list(self._local_workers.get(worker_type, []))
+ return workers
+
+ async def _query_instance_id(self, worker_type: str, worker_key: str) -> Optional[str]:
+ """问单个副本要它的实例身份（admin status 的 instance_id）。
+
+ 复用单副本管理协议 worker///__admin__；查询失败或副本运行旧版
+ 运行时（回复无 instance_id）返回 None，由调用方归入 unknown 组。
+ """
+ wid = worker_key.split("/")[2] if "/" in worker_key else worker_key
+ try:
+ reply = await self._send_admin_cmd(worker_type, wid, "query", timeout=2.0)
+ except Exception as e: # noqa: BLE001 - 单副本查询失败不影响其它副本
+ logger.warning("[instance] %s/%s 身份查询失败: %s", worker_type, wid, e)
+ return None
+ iid = reply.get("instance_id") if isinstance(reply, dict) else None
+ return str(iid) if iid else None
+
+ async def _refresh_instance_map(self, worker_type: str) -> Dict[str, List[str]]:
+ """重建某类型的「实例 → 副本」索引，并按需补查询新副本。
+
+ 只对从未查询过的副本发一次 admin status（本地注册的 inline 副本直接返回
+ 网关自身实例，零查询）；已查询过的副本沿用缓存，因此常规路径是纯内存查询，
+ 仅新上线副本会引入一次额外 RPC。离线副本在重建时被剪枝。
+ """
+ workers = self._all_workers(worker_type)
+ known = self._instance_index.get(worker_type, {})
+ unknown = set(self._instance_unknown.get(worker_type, []))
+ # 反查已有归属：副本 -> 实例（副本 key 唯一，一个副本只属一个实例）
+ owner = {w: iid for iid, ws in known.items() for w in ws}
+
+ index: Dict[str, List[str]] = {}
+ still_unknown: List[str] = []
+ queried = owner.keys() | unknown
+ for w in workers:
+ if w in queried:
+ iid = owner.get(w)
+ else:
+ # 本地注册（inline）副本天然属于本网关实例，无需查询
+ iid = self._gateway_id if w in self._local_workers.get(worker_type, []) else None
+ if iid is None:
+ iid = await self._query_instance_id(worker_type, w)
+ if iid is None:
+ still_unknown.append(w)
+ if iid:
+ index.setdefault(iid, []).append(w)
+
+ self._instance_index[worker_type] = index
+ self._instance_unknown[worker_type] = still_unknown
+ return index
+
+ async def _instance_workers(self, worker_type: str, instance_id: str) -> List[str]:
+ """取某实例下该类型的在线副本 key 列表（无则空表）。"""
+ index = self._instance_index.get(worker_type)
+ if index is None:
+ # 同类型并发请求只让一个跑刷新，其余等结果，避免重复查询同一批副本
+ lock = self._instance_locks.setdefault(worker_type, asyncio.Lock())
+ async with lock:
+ index = self._instance_index.get(worker_type)
+ if index is None:
+ index = await self._refresh_instance_map(worker_type)
+ return index.get(instance_id, [])
+
+ async def get_worker(
+ self, worker_type: str, instance_id: Optional[str] = None
+ ) -> Optional[str]:
+ """获取一个可用的 worker 地址。
+
+ instance_id 为 None 时按简单随机负载均衡挑选（默认，行为不变）；
+ 指定实例时只在该实例的在线副本中挑选，把请求固定到那个进程实例。
+ 实例无在线副本（或该实例在旧版运行时下无法识别身份）返回 None。
读路径无锁：_workers_cache 由 refresh_workers 整表原子替换，
_local_workers 由同步方法修改，单事件循环内读取安全。
"""
- workers = self._workers_cache.get(worker_type, [])
- # 如果缓存为空，回退到本地注册表（适用于启动初期 refresh_workers 尚未执行时）
- if not workers:
- local = self._local_workers.get(worker_type, [])
- if local:
- return random.choice(local)
- return None
- return random.choice(workers)
+ if instance_id:
+ return await self._pick_from_instance(worker_type, instance_id)
+ workers = self._all_workers(worker_type)
+ return random.choice(workers) if workers else None
+
+ async def _pick_from_instance(
+ self, worker_type: str, instance_id: str
+ ) -> Optional[str]:
+ """在指定实例的在线副本中随机挑一个（定向路径，永不跨实例回退）。
+
+ 定向的语义就是"必须打在同一个进程实例上"，因此实例离线时返回 None 让
+ 调用方报错，而不是悄悄换成别的实例——后者会让粘滞状态类的插件跑在错误
+ 副本上，产生难查的数据错乱。
+ """
+ candidates = await self._instance_workers(worker_type, instance_id)
+ if candidates:
+ return random.choice(candidates)
+ logger.warning(
+ "[instance] 目标实例 %s 下无 %s 的在线副本（已识别实例: %s；%d 个副本读不出身份）",
+ instance_id, worker_type, sorted(self._instance_index.get(worker_type, {})),
+ len(self._instance_unknown.get(worker_type, [])),
+ )
+ return None
async def _pick_other_worker(
- self, worker_type: str, exclude: List[str]
+ self, worker_type: str, exclude: List[str],
+ instance_id: Optional[str] = None,
) -> Optional[str]:
"""
选择一个不同于已尝试列表的 worker。
@@ -430,10 +528,17 @@ class WorkerManager:
Args:
worker_type: Worker 类型
exclude: 已尝试（需排除）的 worker key_expr 列表
+ instance_id: 定向实例；给定时只在同实例内换副本，绝不跨实例
Returns:
可用的 worker key_expr，若无其他可用则返回 None
"""
+ if instance_id:
+ candidates = [
+ w for w in await self._instance_workers(worker_type, instance_id)
+ if w not in exclude
+ ]
+ return random.choice(candidates) if candidates else None
workers = self._workers_cache.get(worker_type, [])
candidates = [w for w in workers if w not in exclude]
if candidates:
@@ -455,24 +560,37 @@ class WorkerManager:
return False
return True
- async def _raise_no_worker(self, worker_type: str) -> None:
+ async def _no_worker_message(
+ self, worker_type: str, instance_id: Optional[str] = None
+ ) -> str:
"""
- 无可用 Worker 时的统一诊断：列出当前已知的所有 worker 类型并抛 RuntimeError。
+ 无可用 Worker 时的统一诊断文案。
- 供 call() / call_stream() 共用，保证两类调用错误信息一致。
+ 供 call() / call_stream() 共用，保证两类调用错误信息一致；定向（instance_id）
+ 调用额外列出当前已识别实例，便于区分"实例名写错"与"实例离线"。
"""
all_known = list(self._workers_cache.keys())
local_known = list(self._local_workers.keys())
all_types = sorted(set(all_known + local_known))
- msg = (
- f"类型 '{worker_type}' 没有可用 Worker。"
- f"当前已知的 Worker 类型: {all_types if all_types else '(无)'}。"
- f"请检查: 1) config/workers/config.toml 是否配置了该类型 "
- f"2) 插件初始化是否成功（查看启动日志中的 '加载 Worker 插件失败' 错误）"
- f"3) Worker 进程是否正常运行"
- )
- logger.warning(msg)
- raise RuntimeError(msg)
+ msg = f"类型 '{worker_type}' 没有可用 Worker。"
+ if instance_id:
+ msg = (
+ f"类型 '{worker_type}' 的实例 '{instance_id}' 没有在线副本（定向调用不"
+ f"跨实例回退）。已识别实例: "
+ f"{sorted(self._instance_index.get(worker_type, {})) or '(无)'}。"
+ f"请检查: 1) 该实例 id 是否写错（取值见 GET "
+ f"/admin/workers/{worker_type}/instances 或副本启动时的 --instance）"
+ f" 2) 该实例进程是否在线 3) 该实例的 worker 是否运行旧版运行时"
+ f"（不支持实例身份上报）"
+ )
+ else:
+ msg += (
+ f"当前已知的 Worker 类型: {all_types if all_types else '(无)'}。"
+ f"请检查: 1) config/workers/config.toml 是否配置了该类型 "
+ f"2) 插件初始化是否成功（查看启动日志中的 '加载 Worker 插件失败' 错误）"
+ f"3) Worker 进程是否正常运行"
+ )
+ return msg
async def call_stream(
self,
@@ -481,6 +599,7 @@ class WorkerManager:
payload: Dict[str, Any],
timeout: float = 300.0,
idle_timeout: float = 10.0,
+ instance_id: Optional[str] = None,
) -> AsyncGenerator[bytes, None]:
"""
流式调用：消费 Worker 的多回复流，逐块产出 JSON 字节。
@@ -499,19 +618,21 @@ class WorkerManager:
payload: 请求参数（信封格式，由路由层构造）
timeout: 流总寿命上限（秒），需大于业务最长生成时长
idle_timeout: 空闲超时（秒），超时抛 StreamIdleTimeoutError
+ instance_id: 定向进程实例（None=随机负载均衡）；流本身不可重试，
+ 故实例离线时直接报错，不会中途改派到别的实例
Yields:
bytes: 每个回复块的 JSON 字节
Raises:
- RuntimeError: 无可用 worker
+ RuntimeError: 无可用 worker（含定向实例无在线副本）
StreamIdleTimeoutError: 空闲超时
"""
# 急切解析 worker：无可用时立即抛错，让路由层映射 HTTP 503
# （异步生成器函数体在首次迭代才执行，故这里用普通 async 函数返回内部生成器）
- worker_expr = await self.get_worker(worker_type)
+ worker_expr = await self.get_worker(worker_type, instance_id=instance_id)
if not worker_expr:
- await self._raise_no_worker(worker_type)
+ raise RuntimeError(await self._no_worker_message(worker_type, instance_id))
full_key = f"{worker_expr}/{subject}"
reply_iter = self.session.get(
@@ -602,6 +723,7 @@ class WorkerManager:
timeout: float = 3.0,
retries: int = 1,
cacheable: bool = False,
+ instance_id: Optional[str] = None,
) -> Any:
"""
统一调用入口（dict 语义，MQTT Bridge 等调用方使用）
@@ -613,6 +735,8 @@ class WorkerManager:
timeout: 单次调用超时（秒）
retries: 失败重试次数
cacheable: 是否允许缓存结果（若为 False 则跳过缓存读写）
+ instance_id: 定向进程实例；给定时请求固定发往该实例（不参与随机
+ 负载均衡，重试也只在该实例内换副本），None=默认随机均衡
Returns:
Dict[str, Any]: 调用结果（worker 回复解析后的 dict）
@@ -620,10 +744,18 @@ class WorkerManager:
# cacheable=False 时不计算 cache_key，省一次排序序列化
cache_key = None
if cacheable:
- cache_key = f"{worker_type}:{subject}:{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}"
+ # 定向调用的缓存键含实例名：定向通常用于读取该实例的本地状态
+ # （本地库/设备连接），共用缓存会拿到其它实例的结果
+ cache_key = (
+ f"{worker_type}:{subject}"
+ f":{orjson.dumps(payload, option=orjson.OPT_SORT_KEYS).decode()}"
+ )
+ if instance_id:
+ cache_key = f"{instance_id}@{cache_key}"
reply_bytes = await self._call_impl(
worker_type, subject, orjson.dumps(payload),
timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
+ instance_id=instance_id,
)
return orjson.loads(reply_bytes) if reply_bytes else {}
@@ -635,6 +767,7 @@ class WorkerManager:
timeout: float = 3.0,
retries: int = 1,
cacheable: bool = False,
+ instance_id: Optional[str] = None,
) -> bytes:
"""
统一调用入口（bytes 语义，HTTP 端点透传用）：返回 worker 回复的原始字节。
@@ -652,6 +785,7 @@ class WorkerManager:
timeout: 单次调用超时（秒）
retries: 失败重试次数
cacheable: 是否允许缓存结果（缓存原始字节）
+ instance_id: 定向进程实例（见 call()）；None=默认随机均衡
Returns:
bytes: worker 回复的原始字节（空回复兜底 b"{}"，与 call() 的 {} 语义一致）
@@ -660,9 +794,12 @@ class WorkerManager:
cache_key = None
if cacheable:
cache_key = f"{worker_type}:{subject}:{hashlib.sha256(payload).hexdigest()}"
+ if instance_id:
+ cache_key = f"{instance_id}@{cache_key}"
return await self._call_impl(
worker_type, subject, payload,
timeout=timeout, retries=retries, cacheable=cacheable, cache_key=cache_key,
+ instance_id=instance_id,
)
async def _call_impl(
@@ -674,11 +811,13 @@ class WorkerManager:
retries: int = 1,
cacheable: bool = False,
cache_key: Optional[str] = None,
+ instance_id: Optional[str] = None,
) -> bytes:
"""
- 调用实现：缓存 > 负载均衡 > 超时重试，返回回复的原始字节。
+ 调用实现：缓存 > 选副本（随机均衡/实例定向）> 超时重试，返回回复的原始字节。
call() / call_raw() 共用本实现，仅入参出参形态不同（dict ↔ bytes）。
+ instance_id 给定时选副本与重试都被约束在该实例内（定向语义，不跨实例回退）。
"""
# 1. 缓存检查（仅在 cacheable=True 时检查）
if cacheable and cache_key is not None:
@@ -695,16 +834,39 @@ class WorkerManager:
logger.debug(f"命中共享缓存: {cache_key}")
return disk_val
- # 2. 获取可用 worker
- worker_expr = await self.get_worker(worker_type)
+ # 2. 获取可用 worker（instance_id 给定时固定到该实例，否则随机均衡）
+ worker_expr = await self.get_worker(worker_type, instance_id=instance_id)
if not worker_expr:
- await self._raise_no_worker(worker_type)
+ raise RuntimeError(await self._no_worker_message(worker_type, instance_id))
full_key = f"{worker_expr}/{subject}"
last_error = None
tried_workers: List[str] = []
+ async def _switch_worker(why: str) -> bool:
+ """换一个副本重试；返回是否值得继续（定向调用只在同实例内换副本）。
+
+ 换完更新闭包外层的 worker_expr / full_key。定向实例内已无未试副本时
+ 返回 False —— 继续循环只会重复打同一个坏副本，白等一整轮超时。
+ """
+ nonlocal worker_expr, full_key, last_error
+ new_worker = await self._pick_other_worker(
+ worker_type, tried_workers, instance_id=instance_id
+ )
+ if new_worker:
+ if new_worker != worker_expr:
+ logger.info(f"{why}切换 Worker: {worker_expr} -> {new_worker}")
+ worker_expr = new_worker
+ full_key = f"{worker_expr}/{subject}"
+ return True
+ # 没有任何未试过的副本可换：继续循环只会重复打同一个副本（若它已死，
+ # 白等一整轮超时；若它已回业务错误，结果必然相同），直接结束重试。
+ # 保留具体失败原因（last_error 已由调用方按场景写入），便于定位。
+ scope = f"实例 '{instance_id}' 内" if instance_id else ""
+ last_error = f"{scope}已无其它未试副本可换（{why}）：{last_error}"
+ return False
+
for attempt in range(retries + 1):
# 记录已尝试的 worker
if worker_expr not in tried_workers:
@@ -735,11 +897,8 @@ class WorkerManager:
f"Worker {worker_expr} 无回复，尝试重试 {attempt+1}/{retries+1}"
)
last_error = f"Worker {worker_expr} 无回复"
- # 换一个 worker 重试
- new_worker = await self._pick_other_worker(worker_type, tried_workers)
- if new_worker:
- worker_expr = new_worker
- full_key = f"{worker_expr}/{subject}"
+ if not await _switch_worker("无回复"):
+ break
continue
reply = first_reply
@@ -760,11 +919,8 @@ class WorkerManager:
logger.warning(f"Worker 返回错误: {error_msg} (尝试 {attempt+1}/{retries+1})")
last_error = f"Worker {worker_expr} 返回错误: {error_msg}"
# 错误回复（含 Zenoh 超时）也换一个 worker 重试
- new_worker = await self._pick_other_worker(worker_type, tried_workers)
- if new_worker:
- logger.info(f"错误后切换 Worker: {worker_expr} -> {new_worker}")
- worker_expr = new_worker
- full_key = f"{worker_expr}/{subject}"
+ if not await _switch_worker("错误后"):
+ break
continue # 重试
except asyncio.TimeoutError:
logger.warning(
@@ -772,11 +928,8 @@ class WorkerManager:
)
last_error = f"Worker {worker_expr} 超时 (>{timeout}s)"
# 超时后换一个 worker 重试，避免打到仍在处理中的同一个 worker
- new_worker = await self._pick_other_worker(worker_type, tried_workers)
- if new_worker:
- logger.info(f"超时后切换 Worker: {worker_expr} -> {new_worker}")
- worker_expr = new_worker
- full_key = f"{worker_expr}/{subject}"
+ if not await _switch_worker("超时后"):
+ break
continue
except Exception as e:
logger.error(f"调用 worker 异常: {e}")
@@ -784,7 +937,7 @@ class WorkerManager:
raise RuntimeError(
f"Worker '{worker_type}' 调用失败 (已尝试 {len(tried_workers)} 个 Worker, "
- f"共 {retries + 1} 次, timeout={timeout}s): {last_error}"
+ f"实际尝试 {attempt + 1} 次/上限 {retries + 1} 次, timeout={timeout}s): {last_error}"
)
# ------------------------------------------------------------------
diff --git a/src/chongming/app/routes/__init__.py b/src/chongming/app/routes/__init__.py
index 8599271..194527a 100644
--- a/src/chongming/app/routes/__init__.py
+++ b/src/chongming/app/routes/__init__.py
@@ -1,6 +1,5 @@
-from .storage import router as storage_router
from .stats import router as stats_router
from .worker_admin import router as worker_admin_router
from .auth import router as auth_router
-__all__ = ["storage_router", "stats_router", "worker_admin_router", "auth_router"]
\ No newline at end of file
+__all__ = ["stats_router", "worker_admin_router", "auth_router"]
\ No newline at end of file
diff --git a/src/chongming/app/routes/storage.py b/src/chongming/app/routes/storage.py
deleted file mode 100644
index 7a85fee..0000000
--- a/src/chongming/app/routes/storage.py
+++ /dev/null
@@ -1,53 +0,0 @@
-# src/chongming/app/routes/storage.py
-import os
-import uuid
-import urllib.parse
-import aiofiles
-
-from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Request
-from fastapi.responses import FileResponse
-
-
-router = APIRouter(prefix="/api/v1/storage", tags=["Storage"])
-
-
-@router.post("/upload", summary="上传文件")
-async def upload_file(
- request: Request,
- file: UploadFile = File(...),
- worker_type: str = Form(...),
-):
- """上传文件到本地存储"""
- upload_root = request.app.state.upload_root
- object_key = f"{worker_type}/{uuid.uuid4()}/{file.filename}"
- dest = os.path.join(upload_root, object_key)
- os.makedirs(os.path.dirname(dest), exist_ok=True)
-
- async with aiofiles.open(dest, "wb") as afp:
- while chunk := await file.read(1024 * 1024):
- await afp.write(chunk)
-
- file_size = os.path.getsize(dest)
- return {
- "file_key": object_key,
- "bucket": "local",
- "filename": file.filename,
- "content_type": file.content_type,
- "size": file_size,
- }
-
-@router.get("/download/{file_key:path}")
-async def download_file(request: Request, file_key: str):
- """从本地存储下载文件"""
- upload_root = request.app.state.upload_root
- path = os.path.join(upload_root, file_key)
- if not os.path.isfile(path):
- raise HTTPException(404, "文件不存在")
-
- raw_filename = file_key.split('/')[-1]
- encoded = urllib.parse.quote(raw_filename, safe='')
- return FileResponse(
- path,
- filename=raw_filename,
- headers={"Content-Disposition": f"attachment; filename*=UTF-8''{encoded}"},
- )
diff --git a/test.png b/test.png
deleted file mode 100644
index 9e9527b..0000000
Binary files a/test.png and /dev/null differ
diff --git a/test/test_call_stats.py b/test/test_call_stats.py
index b6ae633..d723909 100644
--- a/test/test_call_stats.py
+++ b/test/test_call_stats.py
@@ -22,7 +22,7 @@ import httpx
import orjson
from fastapi import FastAPI
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "src"))
sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
@@ -53,6 +53,11 @@ _cs_mod = importlib.util.module_from_spec(_cs_spec)
sys.modules["chongming.app.core.call_stats"] = _cs_mod
_cs_spec.loader.exec_module(_cs_mod) # type: ignore[union-attr]
+# 审计采样率固定为全量：生产 config 的 access_sample_rate=0.1 下，成功请求的审计行
+# 本就是按概率写的，断言"几条请求 = 几条审计行"会随机失败。本测试验证的是审计链路
+# 是否工作，与采样策略无关，故固定为 1.0。
+_cs_mod._AUDIT_SAMPLE_RATE = 1.0
+
# 加载真实 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)
@@ -95,12 +100,14 @@ class FakeWorkerManager:
self.reply = reply
self.error = error
- async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+ async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+ instance_id=None):
if self.error == "no_worker":
raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
return self.reply
- async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0):
+ async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0,
+ instance_id=None):
if self.error == "no_worker":
raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
diff --git a/test/test_instance_targeting.py b/test/test_instance_targeting.py
new file mode 100644
index 0000000..319af7a
--- /dev/null
+++ b/test/test_instance_targeting.py
@@ -0,0 +1,220 @@
+"""
+worker 实例定向（把请求固定到某个进程实例）测试
+
+验证 WorkerManager 的实例定向语义（HTTP 的 X-Chongming-Instance 头 / MQTT 的
+target_instance 字段 / 程序调用 instance_id 参数最终都走这里）：
+ 1. 不指定实例：与以往一致，在全部候选副本里随机挑
+ 2. 指定实例：只在其实例的在线副本里挑，不跨实例
+ 3. 身份来源：副本经 admin status 上报 instance_id，网关按需查询并缓存
+ （本地注册的 inline 副本直接用网关自身实例，无需查询）
+ 4. 目标实例不在线：报错而不回退到别的实例（定向即严格）
+ 5. 调用失败重试：限定在同一实例内换副本，绝不跨实例
+ 6. 实例内已无其它副本可换：立即失败，不重复打同一个副本
+ 7. 实例维度缓存键：定向调用的缓存与其它实例隔离
+
+运行: .venv/bin/python test/test_instance_targeting.py
+"""
+import asyncio
+import sys
+from pathlib import Path
+
+_HERE = Path(__file__).resolve().parent # chongming-zenoh/
+_PROJ = _HERE
+for _p in (_PROJ / "src", _PROJ / "utils" / "worker" / "src"):
+ sys.path.insert(0, str(_p))
+
+from chongming.app.core.worker_manager import WorkerManager # noqa: E402
+
+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 _OkPayload:
+ def __init__(self, data: bytes):
+ self.payload = data
+
+
+class _Ok:
+ def __init__(self, data: bytes):
+ self.ok = _OkPayload(data)
+
+
+class _Reply:
+ """一次 get 的回复：ok 与 err 互斥（与 Zenoh reply 的形态一致）。"""
+
+ def __init__(self, data: bytes = b"{}", err: bytes | None = None):
+ if err is None:
+ self.ok = _OkPayload(data)
+ self.err = None
+ else:
+ self.ok = None
+ self.err = _OkPayload(err)
+
+
+class FakeSession:
+ """记录每次 get 的 key，并按 key 决定回复。
+
+ admin 查询（worker///__admin__）由 admin_instances 决定 instance_id；
+ 业务调用（worker///）可按 call_errors 注入错误回复。
+ """
+
+ def __init__(self, admin_instances: dict | None = None,
+ call_errors: set | None = None):
+ self.admin_instances = admin_instances or {}
+ self.call_errors = set(call_errors or ())
+ self.calls: list = []
+
+ def get(self, key, payload=None, **kwargs):
+ self.calls.append(key)
+ if key.endswith("/__admin__"):
+ wid = key.split("/")[2]
+ iid = self.admin_instances.get(wid)
+ body = (
+ f'{{"worker_id":"{wid}","instance_id":"{iid}","status":"running"}}'.encode()
+ if iid else f'{{"worker_id":"{wid}","status":"running"}}'.encode()
+ )
+ return iter([_Reply(body)])
+ if key in self.call_errors:
+ return iter([_Reply(err=b'{"error":"boom"}')])
+ return iter([_Reply(b'{"ok":true}')])
+
+ def liveliness(self):
+ raise RuntimeError("本测试不触及 liveliness")
+
+ def close(self):
+ pass
+
+
+def new_manager(session, gateway_id="gw-local"):
+ m = WorkerManager(session=session)
+ m._gateway_id = gateway_id
+ return m
+
+
+async def main():
+ # ── 1. 不定向：与以往一致，全候选集随机 ─────────────────
+ sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02"})
+ m = new_manager(sess)
+ m._workers_cache["demo"] = ["worker/demo/w-a", "worker/demo/w-b"]
+ picked = {await m.get_worker("demo") for _ in range(60)}
+ check("1.1 不定向时可命中全部副本", picked == set(m._workers_cache["demo"]),
+ f"picked={sorted(picked)}")
+ check("1.2 不定向不触发实例身份查询（零额外 RPC）",
+ not [c for c in sess.calls if c.endswith("__admin__")], f"calls={sess.calls}")
+ check("1.3 不定向调用走随机副本",
+ await m.call_raw("demo", "ping", b"{}") == b'{"ok":true}')
+
+ # ── 2. 定向：只在目标实例内挑副本 ────────────────────────
+ sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02",
+ "w-c": "edge-01"})
+ m = new_manager(sess)
+ m._workers_cache["demo"] = [
+ "worker/demo/w-a", "worker/demo/w-b", "worker/demo/w-c",
+ ]
+ picked = {await m.get_worker("demo", instance_id="edge-01") for _ in range(60)}
+ check("2.1 定向只命中目标实例的副本",
+ picked == {"worker/demo/w-a", "worker/demo/w-c"}, f"picked={sorted(picked)}")
+
+ # ── 3. 身份查询：按需查询并缓存，实例归属正确 ─────────────
+ sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02",
+ "w-c": "edge-01"})
+ m = new_manager(sess)
+ m._workers_cache["demo"] = [
+ "worker/demo/w-a", "worker/demo/w-b", "worker/demo/w-c",
+ ]
+ index = await m._instance_workers("demo", "edge-01")
+ check("3.1 实例→副本反向索引正确",
+ sorted(index) == ["worker/demo/w-a", "worker/demo/w-c"], f"index={sorted(index)}")
+ admin_calls = [c for c in sess.calls if c.endswith("__admin__")]
+ check("3.2 每个副本只查询一次身份", len(admin_calls) == 3, f"n={len(admin_calls)}")
+ await m._refresh_instance_map("demo")
+ check("3.3 重复刷新复用缓存不重复查询",
+ len([c for c in sess.calls if c.endswith("__admin__")]) == len(admin_calls))
+
+ # ── 3b. inline 本地注册副本：直接用网关实例，零查询 ──────
+ sess = FakeSession()
+ m = new_manager(sess, gateway_id="gw-1")
+ m.register_worker("demo", "worker-0-1-999")
+ index = await m._instance_workers("demo", "gw-1")
+ check("3.4 inline 副本归入网关实例", index == ["worker/demo/worker-0-1-999"],
+ f"index={index}")
+ check("3.5 inline 副本身份查询零 RPC",
+ not [c for c in sess.calls if c.endswith("__admin__")], f"calls={sess.calls}")
+
+ # ── 4. 定向实例不在线：严格报错，不回退 ──────────────────
+ sess = FakeSession(admin_instances={"w-a": "edge-01"})
+ m = new_manager(sess)
+ m._workers_cache["demo"] = ["worker/demo/w-a"]
+ check("4.1 目标实例无副本时 get_worker 返回 None",
+ await m.get_worker("demo", instance_id="edge-99") is None)
+ try:
+ await m.call_raw("demo", "ping", b"{}", instance_id="edge-99")
+ check("4.2 定向到离线实例应报错", False, "未抛错")
+ except RuntimeError as e:
+ msg = str(e)
+ check("4.2 定向到离线实例报错且说明不跨实例回退",
+ "edge-99" in msg and "不" in msg, f"msg={msg[:90]}")
+
+ # ── 5. 重试锚定同实例：绝不跨实例 ────────────────────────
+ sess = FakeSession(
+ admin_instances={"w-a": "edge-01", "w-b": "edge-02", "w-c": "edge-01"},
+ call_errors={"worker/demo/w-b/ping"}, # 只让别实例的副本出错，验证不会被选中
+ )
+ m = new_manager(sess)
+ m._workers_cache["demo"] = [
+ "worker/demo/w-a", "worker/demo/w-b", "worker/demo/w-c",
+ ]
+ # 目标实例 edge-01 内的 w-a 失败 -> 应换到同实例的 w-c，不碰 edge-02 的 w-b
+ sess.call_errors = {"worker/demo/w-a/ping"}
+ out = await m.call_raw("demo", "ping", b"{}", instance_id="edge-01", retries=2)
+ business = [c for c in sess.calls if not c.endswith("__admin__")]
+ check("5.1 重试不跨实例（未打其它实例的副本）",
+ all("/w-b/" not in c for c in business), f"calls={business}")
+ check("5.2 同实例内换副本后成功", out == b'{"ok":true}', f"out={out!r}")
+
+ # ── 6. 实例内无其它副本可换：立即失败 ────────────────────
+ sess = FakeSession(admin_instances={"w-a": "edge-01"},
+ call_errors={"worker/demo/w-a/ping"})
+ m = new_manager(sess)
+ m._workers_cache["demo"] = ["worker/demo/w-a", "worker/demo/w-b"]
+ sess.admin_instances["w-b"] = "edge-02"
+ raised = None
+ try:
+ await m.call_raw("demo", "ping", b"{}", instance_id="edge-01", retries=2)
+ except RuntimeError as e:
+ raised = e
+ check("6.1 实例内无副本可换应报错", raised is not None, f"raised={raised!r}")
+ n = len([c for c in sess.calls if not c.endswith("__admin__")])
+ check("6.2 不重复打同一个副本（只调用一次）", n == 1, f"业务调用次数={n}")
+ check("6.3 错误信息点名实例与原因",
+ raised is not None and "edge-01" in str(raised), f"msg={str(raised)[:90]}")
+
+ # ── 7. 定向调用的缓存按实例隔离 ──────────────────────────
+ sess = FakeSession(admin_instances={"w-a": "edge-01", "w-b": "edge-02"})
+ m = new_manager(sess)
+ m._workers_cache["demo"] = ["worker/demo/w-a", "worker/demo/w-b"]
+ r1 = await m.call_raw("demo", "state", b"{}", cacheable=True, instance_id="edge-01")
+ r2 = await m.call_raw("demo", "state", b"{}", cacheable=True, instance_id="edge-02")
+ keys = [f"{iid}@demo:state:" for iid in ("edge-01", "edge-02")]
+ check("7.1 不同实例的定向缓存键互不影响",
+ all(any(k.startswith(p) for k in m._rpc_mem_cache._d) for p in keys),
+ f"cache_keys={list(m._rpc_mem_cache._d)}")
+ check("7.2 定向调用结果正常返回", r1 == b'{"ok":true}' and r2 == b'{"ok":true}',
+ f"r1={r1!r}, r2={r2!r}")
+
+ print()
+ if failures:
+ print(f"FAILED: {len(failures)} 项 -> {failures}")
+ return 1
+ print("全部通过")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(asyncio.run(main()))
diff --git a/test/test_instance_targeting_http.py b/test/test_instance_targeting_http.py
new file mode 100644
index 0000000..d0eb8c9
--- /dev/null
+++ b/test/test_instance_targeting_http.py
@@ -0,0 +1,207 @@
+"""
+HTTP 实例定向头（X-Chongming-Instance）透传测试
+
+依赖隔离：桩掉 chongming 包链、auth 与 chongming_cache，直接加载真实
+route_generator.py 与 worker_manager.py，用 FastAPI TestClient 验证:
+ 1. 携带 X-Chongming-Instance 的请求，该实例 id 透传到 call_raw / call_stream
+ 2. 未携带该头时 instance_id 为 None（行为与改动前完全一致，兼容旧调用方）
+ 3. 覆盖全部端点形态：GET query / POST body / validate=false 原样透传 / SSE 流式
+ 4. 头值为空白时视为未指定
+ 5. 定向到离线实例（RuntimeError）-> HTTP 503，错误信息透出
+
+运行: .venv/bin/python test/test_instance_targeting_http.py
+"""
+import importlib.util
+import logging
+import sys
+import types
+from pathlib import Path
+
+import orjson
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+TEST_DIR = Path(__file__).resolve().parent
+REPO_ROOT = TEST_DIR.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
+_cache_stub = types.ModuleType("chongming_cache")
+_cache_stub.Cache = object
+_cache_stub.DistributedLock = object # worker_manager 顶层导入（本测试不触达）
+for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
+ sys.modules[mod.__name__] = mod
+
+_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]
+
+_cs_path = REPO_ROOT / "src" / "chongming" / "app" / "core" / "call_stats.py"
+_cs_spec = importlib.util.spec_from_file_location(
+ "chongming.app.core.call_stats", _cs_path
+)
+_cs_mod = importlib.util.module_from_spec(_cs_spec)
+sys.modules["chongming.app.core.call_stats"] = _cs_mod
+_cs_spec.loader.exec_module(_cs_mod) # type: ignore[union-attr]
+
+_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
+INSTANCE_HEADER = _rg_mod.INSTANCE_HEADER
+
+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:
+ """记录每次调用收到的 instance_id；可指定定点失败以验证 503 映射。"""
+
+ def __init__(self, fail_instance=None):
+ self.fail_instance = fail_instance # 指定实例报错（模拟实例离线）
+ self.raw_instances = [] # call_raw 收到的 instance_id
+ self.stream_instances = [] # call_stream 收到的 instance_id
+
+ async def call_raw(self, worker_type, subject, payload, timeout=3.0,
+ retries=1, cacheable=False, instance_id=None):
+ self.raw_instances.append(instance_id)
+ if instance_id and instance_id == self.fail_instance:
+ raise RuntimeError(
+ f"类型 '{worker_type}' 的实例 '{instance_id}' 没有在线副本"
+ "（定向调用不跨实例回退）。"
+ )
+ return orjson.dumps({"op": subject, "instance": instance_id})
+
+ async def call_stream(self, worker_type, subject, payload, timeout=300.0,
+ idle_timeout=10.0, instance_id=None):
+ self.stream_instances.append(instance_id)
+ if instance_id and instance_id == self.fail_instance:
+ raise RuntimeError(f"实例 '{instance_id}' 没有在线副本")
+
+ async def _gen():
+ yield orjson.dumps({"instance": instance_id})
+
+ return _gen()
+
+
+ROUTES = [
+ {"subject": "t.get", "path": "/calc/add", "method": "GET",
+ "params": ["a: float", "b: float"], "summary": "add",
+ "timeout": 3.0, "auth_required": False},
+ {"subject": "t.post", "path": "/calc/echo", "method": "POST",
+ "params": ["msg: str"], "summary": "echo",
+ "timeout": 3.0, "auth_required": False},
+ {"subject": "t.raw", "path": "/calc/raw", "method": "POST",
+ "params": ["payload: any"], "summary": "原样透传",
+ "validate": False, "timeout": 3.0, "auth_required": False},
+ {"subject": "t.stream", "path": "/stream/chat", "method": "POST",
+ "params": ["prompt: str"], "summary": "sse",
+ "stream": True, "timeout": 60.0, "stream_idle_timeout": 1.0,
+ "auth_required": False},
+]
+
+
+def build_client(fwm):
+ app = FastAPI()
+ register_route_group(app, fwm, "demotype", ROUTES, prefix="/api/v1")
+ return TestClient(app)
+
+
+def main():
+ # ── 1. GET query 端点 ────────────────────────────────────
+ fwm = FakeWorkerManager()
+ c = build_client(fwm)
+ r = c.get("/api/v1/calc/add?a=1&b=2", headers={INSTANCE_HEADER: "edge-01"})
+ check("1.1 GET 携带头 -> 200", r.status_code == 200, f"status={r.status_code}")
+ check("1.2 GET 头透传到 call_raw", fwm.raw_instances == ["edge-01"],
+ f"instances={fwm.raw_instances}")
+ check("1.3 响应体回带实例（证明打到目标实例）",
+ r.json().get("instance") == "edge-01", f"body={r.json()}")
+
+ # ── 2. 未携带头：兼容旧调用方 ────────────────────────────
+ fwm = FakeWorkerManager()
+ c = build_client(fwm)
+ r = c.get("/api/v1/calc/add?a=1&b=2")
+ check("2.1 未携带头 -> 200（默认随机均衡路径不变）", r.status_code == 200)
+ check("2.2 未携带头时 instance_id=None", fwm.raw_instances == [None],
+ f"instances={fwm.raw_instances}")
+ check("2.3 响应体不含实例", r.json().get("instance") is None, f"body={r.json()}")
+
+ # ── 3. POST body 端点 ────────────────────────────────────
+ fwm = FakeWorkerManager()
+ c = build_client(fwm)
+ r = c.post("/api/v1/calc/echo", json={"msg": "hi"},
+ headers={INSTANCE_HEADER: "edge-02"})
+ check("3.1 POST body 头透传", fwm.raw_instances == ["edge-02"],
+ f"instances={fwm.raw_instances}")
+
+ # ── 4. validate=false 原样透传端点 ───────────────────────
+ fwm = FakeWorkerManager()
+ c = build_client(fwm)
+ r = c.post("/api/v1/calc/raw", content=b'{"anything": [1, 2]}',
+ headers={INSTANCE_HEADER: "edge-03",
+ "content-type": "application/json"})
+ check("4.1 原样透传端点头透传", fwm.raw_instances == ["edge-03"],
+ f"instances={fwm.raw_instances}")
+
+ # ── 5. SSE 流式端点 ──────────────────────────────────────
+ fwm = FakeWorkerManager()
+ c = build_client(fwm)
+ with c.stream("POST", "/api/v1/stream/chat", json={"prompt": "x"},
+ headers={INSTANCE_HEADER: "edge-04"}) as r:
+ body = b"".join(r.iter_bytes())
+ check("5.1 流式端点头透传", fwm.stream_instances == ["edge-04"],
+ f"instances={fwm.stream_instances}")
+ check("5.2 流式响应含目标实例", b"edge-04" in body, f"body={body[:80]!r}")
+
+ # ── 6. 空白头值视为未指定 ────────────────────────────────
+ fwm = FakeWorkerManager()
+ c = build_client(fwm)
+ r = c.get("/api/v1/calc/add?a=1&b=2", headers={INSTANCE_HEADER: " "})
+ check("6.1 空白头值 -> instance_id=None", fwm.raw_instances == [None],
+ f"instances={fwm.raw_instances}")
+
+ # ── 7. 定向实例离线 -> 503 ───────────────────────────────
+ fwm = FakeWorkerManager(fail_instance="edge-gone")
+ c = build_client(fwm)
+ r = c.get("/api/v1/calc/add?a=1&b=2", headers={INSTANCE_HEADER: "edge-gone"})
+ check("7.1 定向实例离线 -> 503", r.status_code == 503, f"status={r.status_code}")
+ check("7.2 503 详情点名实例", "edge-gone" in str(r.json().get("detail", "")),
+ f"detail={r.json().get('detail')}")
+
+ print()
+ if failures:
+ print(f"FAILED: {len(failures)} 项 -> {failures}")
+ return 1
+ print("全部通过")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/test/test_mqtt_custom_topics.py b/test/test_mqtt_custom_topics.py
index 7c589d6..79336f1 100644
--- a/test/test_mqtt_custom_topics.py
+++ b/test/test_mqtt_custom_topics.py
@@ -33,7 +33,8 @@ class FakeWM:
self.calls = []
self.result = {"result": "ok"}
- async def call(self, worker_type, subject, payload, timeout=30, cacheable=False):
+ async def call(self, worker_type, subject, payload, timeout=30, cacheable=False,
+ instance_id=None):
self.calls.append(
{
"worker_type": worker_type,
@@ -41,6 +42,7 @@ class FakeWM:
"payload": payload,
"timeout": timeout,
"cacheable": cacheable,
+ "instance_id": instance_id,
}
)
return self.result
@@ -153,6 +155,41 @@ async def test_flat_payload_routed():
print("✅ 非信封（扁平）载荷按原始报文路由通过")
+async def test_target_instance():
+ """实例定向：载荷顶层 target_instance 作为控制字段摘除并透传给 worker_manager.call"""
+ wm = FakeWM()
+ bridge, published = make_bridge(
+ wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
+ )
+ # 1. 带 target_instance：透传实例名，且控制字段不进业务 payload
+ await bridge._handle(make_msg(
+ "devices/d1/telemetry",
+ {"meta": {}, "data": {"v": 1}, "target_instance": "edge-01"},
+ ))
+ assert wm.calls[-1]["instance_id"] == "edge-01", wm.calls[-1]
+ assert "target_instance" not in wm.calls[-1]["payload"], wm.calls[-1]["payload"]
+ assert wm.calls[-1]["payload"]["data"] == {"v": 1}
+
+ # 2. 不带该字段：instance_id 为 None（跨实例随机均衡，与改动前一致）
+ await bridge._handle(make_msg("devices/d1/telemetry", {"meta": {}, "data": {"v": 1}}))
+ assert wm.calls[-1]["instance_id"] is None, wm.calls[-1]
+
+ # 3. 空白值视为未指定
+ await bridge._handle(make_msg(
+ "devices/d1/telemetry",
+ {"meta": {}, "data": {"v": 1}, "target_instance": " "},
+ ))
+ assert wm.calls[-1]["instance_id"] is None, wm.calls[-1]
+
+ # 4. 内置 chongming/... 主题同样支持
+ await bridge._handle(make_msg(
+ f"{MQTT_TOPIC_PREFIX}/example/add",
+ {"meta": {}, "data": {"a": 1}, "target_instance": "edge-02"},
+ ))
+ assert wm.calls[-1]["instance_id"] == "edge-02", wm.calls[-1]
+ print("✅ 实例定向（target_instance 控制字段）透传通过")
+
+
async def test_non_json_dropped():
wm = FakeWM()
bridge, published = make_bridge(
@@ -219,6 +256,7 @@ async def main():
await test_custom_routing()
await test_envelope_meta_fields()
await test_flat_payload_routed()
+ await test_target_instance()
await test_non_json_dropped()
await test_board_ping_topic()
test_plugin_config_driven_collection()
diff --git a/test/test_passthrough_optimization.py b/test/test_passthrough_optimization.py
index b67b0ef..d5441dc 100644
--- a/test/test_passthrough_optimization.py
+++ b/test/test_passthrough_optimization.py
@@ -16,12 +16,13 @@
D. WorkerManager 单测（FakeSession）:
9. call_raw 零解析返回原始字节；call() 仍解析为 dict（MQTT 语义保留）
10. cacheable: 缓存原始 bytes、二次命中不再发查询、业务错误不缓存
- 11. reply.err -> 重试 -> RuntimeError；空回复兜底 b"{}"
+ 11. reply.err -> 换副本重试/无副本可换时立即 RuntimeError；空回复兜底 b"{}"
运行: uv run python test_passthrough_optimization.py
"""
import asyncio
import importlib.util
+import random
import logging
import sys
import types
@@ -32,7 +33,7 @@ import orjson
from fastapi import FastAPI
from fastapi.testclient import TestClient
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "src"))
sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
@@ -49,6 +50,7 @@ _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
+_cache_stub.DistributedLock = object # worker_manager 顶层导入（本测试不触达）
for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
sys.modules[mod.__name__] = mod
@@ -101,7 +103,8 @@ class FakeWorkerManager:
self.error = error
self.call_raw_payloads = [] # 捕获的预序列化请求字节
- async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+ async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+ instance_id=None):
self.call_raw_payloads.append(payload)
if self.error == "no_worker":
raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
@@ -282,16 +285,27 @@ def test_openapi_request_body_injected():
# ---- D. WorkerManager 单测（FakeSession）----
class FakeSession:
- """模拟 zenoh Session：get() 返回单回复迭代器"""
+ """模拟 zenoh Session：get() 返回单回复迭代器
- def __init__(self, payload=b'{"result": 5}', err_payload=None):
+ err_payload 非空时整体报错；err_target 给定时只对该 key 前缀报错（其余正常回复），
+ 用于验证「失败换副本重试」。
+ """
+
+ def __init__(self, payload=b'{"result": 5}', err_payload=None, err_target=None):
self.payload = payload
self.err_payload = err_payload
+ self.err_target = err_target
self.get_calls = 0
+ self.call_keys = []
def get(self, *args, **kwargs):
self.get_calls += 1
- if self.err_payload is not None:
+ key = args[0] if args else kwargs.get("key_expr", "")
+ self.call_keys.append(str(key))
+ failing = self.err_payload is not None and (
+ self.err_target is None or str(key).startswith(self.err_target)
+ )
+ if failing:
reply = SimpleNamespace(ok=None, err=SimpleNamespace(payload=self.err_payload))
else:
reply = SimpleNamespace(ok=SimpleNamespace(payload=self.payload), err=None)
@@ -341,14 +355,49 @@ async def test_wm_cache_bytes():
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))
+ check("reply.err 抛 RuntimeError 且保留具体错误", "boom" in str(e), str(e))
+ check("单副本无其它可换时不空转重试（1 次查询）",
+ session.get_calls == 1, str(session.get_calls))
+
+
+async def test_wm_reply_err_switches_worker():
+ """多副本：先打的副本报错则换另一个副本重试，且不重复打已失败的副本。
+
+ 首个副本由 random.choice 决定，故对多个 seed 试跑：必有一个覆盖「先失败后成功」
+ 路径（另一个覆盖「首个即成功」），两条路径都不允许重复打同一个失败的副本。
+ """
+ ok, detail = False, "无 seed 覆盖到失败换副本路径"
+ for seed in (0, 1):
+ random.seed(seed)
+ session = FakeSession(
+ payload=b'{"result": 5}', err_payload=b'{"error": "boom"}',
+ err_target="worker/demo/w-dead",
+ )
+ wm = make_wm(session)
+ wm._workers_cache = {"demo": ["worker/demo/w-dead", "worker/demo/w-alive"]}
+ try:
+ result = await wm.call_raw("demo", "calc.add", b'{}', retries=2)
+ except RuntimeError as e: # 未换到正常副本（不该发生）
+ detail = f"seed={seed} 意外失败: {e}"
+ continue
+ assert result == b'{"result": 5}', result
+ dead_calls = session.call_keys.count("worker/demo/w-dead/calc.add")
+ if dead_calls > 1:
+ detail = f"seed={seed} 重复打失败副本: {session.call_keys}"
+ continue
+ if session.call_keys == [
+ "worker/demo/w-dead/calc.add",
+ "worker/demo/w-alive/calc.add",
+ ]:
+ ok, detail = True, f"seed={seed} keys={session.call_keys}"
+ check("失败副本 -> 换副本重试成功（且不重复打失败副本）", ok, detail)
async def test_wm_empty_reply_fallback():
@@ -375,6 +424,7 @@ def run():
asyncio.run(test_wm_call_dict_semantics())
asyncio.run(test_wm_cache_bytes())
asyncio.run(test_wm_reply_err_retry())
+ asyncio.run(test_wm_reply_err_switches_worker())
asyncio.run(test_wm_empty_reply_fallback())
print(f"\n结果: {len(failures)} 个失败")
diff --git a/test/test_stream_e2e_zenoh.py b/test/test_stream_e2e_zenoh.py
index f8014c4..9a6d226 100644
--- a/test/test_stream_e2e_zenoh.py
+++ b/test/test_stream_e2e_zenoh.py
@@ -24,7 +24,7 @@ from pathlib import Path
import orjson
import zenoh
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
sys.path.insert(0, str(REPO_ROOT / "src"))
@@ -39,6 +39,7 @@ _logger_stub = types.ModuleType("chongming.app.core.logger")
_logger_stub.get_logger = lambda name: logging.getLogger(name)
_cache_stub = types.ModuleType("chongming_cache")
_cache_stub.Cache = object
+_cache_stub.DistributedLock = object # worker_manager 顶层导入（本测试不触达）
for mod in (_pkg, _app, _core, _logger_stub, _cache_stub):
sys.modules[mod.__name__] = mod
@@ -67,7 +68,8 @@ def check(name, cond, detail=""):
failures.append(name)
-LISTEN_PORT = 7447
+import os
+LISTEN_PORT = int(os.environ.get("CM_STREAM_TEST_PORT", "7447"))
LISTEN = f"tcp/127.0.0.1:{LISTEN_PORT}"
diff --git a/test/test_stream_route.py b/test/test_stream_route.py
index 436d06e..d3f37e5 100644
--- a/test/test_stream_route.py
+++ b/test/test_stream_route.py
@@ -21,7 +21,7 @@ import orjson
from fastapi import FastAPI
from fastapi.testclient import TestClient
-REPO_ROOT = Path(__file__).resolve().parent
+REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "src"))
sys.path.insert(0, str(REPO_ROOT / "utils" / "worker" / "src"))
@@ -38,6 +38,7 @@ _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
+_cache_stub.DistributedLock = object # worker_manager 顶层导入（本测试不触达）
for mod in (_pkg, _app, _core, _logger_stub, _auth_stub, _cache_stub):
sys.modules[mod.__name__] = mod
@@ -90,18 +91,21 @@ class FakeWorkerManager:
self.call_payloads = []
self.stream_payloads = []
- async def call(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False):
+ async def call(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+ instance_id=None):
self.call_payloads.append((worker_type, subject, payload))
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):
+ async def call_raw(self, worker_type, subject, payload, timeout=3.0, retries=1, cacheable=False,
+ instance_id=None):
# 网关透传路径：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):
+ async def call_stream(self, worker_type, subject, payload, timeout=300.0, idle_timeout=10.0,
+ instance_id=None):
if self.stream_error == "no_worker":
raise RuntimeError(f"类型 '{worker_type}' 没有可用 Worker。")
self.stream_payloads.append((worker_type, subject, payload))