92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
|
"""
|
|||
|
|
worker 超时冷静期（派发负反馈）测试
|
|||
|
|
|
|||
|
|
背景：半死副本（被 SIGKILL 的实例留下的孤儿 worker 等）只要还在服务发现里，
|
|||
|
|
每次被随机选中都会白吃一整轮超时（默认 2s）；worker_manager 只做"本次请求换副本
|
|||
|
|
重试"，没有跨请求记忆，于是 2 秒惩罚反复发生（实测一轮 31 条超时日志）。
|
|||
|
|
|
|||
|
|
修复：WorkerManager.mark_worker_suspect 给超时/无回复的副本记一段冷静期
|
|||
|
|
（_suspect_until，TTL 30s），三条选副本路径（get_worker / _pick_from_instance /
|
|||
|
|
_pick_other_worker）都优先跳过它，全在冷静期时退回全集（宁慢不缺）。
|
|||
|
|
|
|||
|
|
验证：
|
|||
|
|
1. 标记后的副本跨请求不再被选中
|
|||
|
|
2. 全部候选都处于冷静期时退回全集（不饿死）
|
|||
|
|
3. TTL 到期自动恢复
|
|||
|
|
4. 过期项会被清理（防长期运行膨胀）
|
|||
|
|
5. 换副本路径同样避开冷静期
|
|||
|
|
6. 源码兜底：超时/无回复分支确实调用了标记
|
|||
|
|
|
|||
|
|
运行: uv run python test/test_worker_suspect.py
|
|||
|
|
"""
|
|||
|
|
import asyncio
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
ROOT = Path(__file__).parent.parent
|
|||
|
|
sys.path.insert(0, str(ROOT / "gateway" / "src"))
|
|||
|
|
|
|||
|
|
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)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# 跳过 __init__（依赖 zenoh 会话/缓存），只装配选副本所需的字段
|
|||
|
|
m = WorkerManager.__new__(WorkerManager)
|
|||
|
|
m._suspect_until = {}
|
|||
|
|
m._workers_cache = {"demo": ["w1", "w2", "w3"]}
|
|||
|
|
m._local_workers = {}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def sample(fn, n=40):
|
|||
|
|
return {asyncio.run(fn()) for _ in range(n)}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ========== 1. 标记后跨请求跳过 ==========
|
|||
|
|
m.mark_worker_suspect("w1")
|
|||
|
|
m.mark_worker_suspect("w2")
|
|||
|
|
picked = sample(lambda: m.get_worker("demo"))
|
|||
|
|
check("冷静期内的副本不再被选中", picked == {"w3"}, str(picked))
|
|||
|
|
|
|||
|
|
# ========== 2. 全在冷静期 → 退回全集 ==========
|
|||
|
|
m.mark_worker_suspect("w3")
|
|||
|
|
picked = sample(lambda: m.get_worker("demo"))
|
|||
|
|
check("全部处于冷静期时退回全集（不饿死）", picked == {"w1", "w2", "w3"}, str(picked))
|
|||
|
|
|
|||
|
|
# ========== 3. TTL 到期恢复 ==========
|
|||
|
|
m._suspect_until = {"w1": time.monotonic() - 0.1}
|
|||
|
|
check("冷静期过期自动恢复", m._filter_suspects(["w1"]) == ["w1"])
|
|||
|
|
|
|||
|
|
# ========== 4. 过期项清理 ==========
|
|||
|
|
m._suspect_until = {}
|
|||
|
|
for i in range(40):
|
|||
|
|
m.mark_worker_suspect(f"x{i}", ttl=-1.0) # 立即过期
|
|||
|
|
check("过期项会被清理（防膨胀）", len(m._suspect_until) <= 32, str(len(m._suspect_until)))
|
|||
|
|
|
|||
|
|
# ========== 5. 换副本路径同样避开 ==========
|
|||
|
|
m._suspect_until = {}
|
|||
|
|
m.mark_worker_suspect("w1")
|
|||
|
|
picked = sample(lambda: m._pick_other_worker("demo", []))
|
|||
|
|
check("换副本路径同样避开冷静期", picked == {"w2", "w3"}, str(picked))
|
|||
|
|
|
|||
|
|
# ========== 6. 源码兜底：失败分支确实标记 ==========
|
|||
|
|
src = (ROOT / "gateway" / "src" / "chongming" / "app" / "core" / "worker_manager.py").read_text(
|
|||
|
|
encoding="utf-8"
|
|||
|
|
)
|
|||
|
|
n = src.count("self.mark_worker_suspect(worker_expr)")
|
|||
|
|
check("超时/无回复分支都调用了标记", n >= 2, f"{n} 处")
|
|||
|
|
|
|||
|
|
print("\n" + ("=" * 40))
|
|||
|
|
if failures:
|
|||
|
|
print(f"❌ {len(failures)} 项未通过: {failures}")
|
|||
|
|
sys.exit(1)
|
|||
|
|
print("✅ 全部通过")
|