diff --git a/src/xskill/canary.py b/src/xskill/canary.py index 04f63840..76254263 100644 --- a/src/xskill/canary.py +++ b/src/xskill/canary.py @@ -24,6 +24,7 @@ import hashlib import json import logging +import random import shutil import threading from dataclasses import dataclass @@ -294,6 +295,105 @@ def pick_side_scoped(traj_id: str, skill_name: str, probability: float, return pick_side(traj_id, skill_name, probability) +# ═══════════════════════════════════════════════════════════════════ +# 有状态分流:CanaryRouter —— 保证总量的均衡随机分配 +# ═══════════════════════════════════════════════════════════════════ +# pick_side 是无状态哈希:每次调用独立做一次伯努利试验。当分流 key 的 +# 基数很小(典型:team-CS 只有 3 个 worker),3 次 0.5 概率的独立试验完全 +# 可能全落到 main,staging 流量为 0,灰度永远拿不到样本。 +# +# CanaryRouter 解决这个:按 (skill) 维护"已分配的 client→side"账本,新 +# client 进来时选让"当前 main/staging 比例最接近 probability"的那一侧(平 +# 局随机),从而把 staging 的份额"锁"进总量,而不是听天由命。同一 client +# 在同一个 staging 版本内 side 钉死(轨迹一致性);staging sha 或 +# probability 变了则重置账本、重新均衡。 +# +# 只用在 team-CS manifest 路径(client 基数小)。高基数路径(traj_id / +# window_id)继续用 pick_side——大数定律下无状态哈希天然均衡。 + + +class CanaryRouter: + """有状态灰度分流器:按 client 随机分配并保证 staging 总量份额。 + + 与无状态 :func:`pick_side` 的区别:pick_side 每次调用独立哈希,client + 很少时 staging 可能被饿死到 0。本路由器按 skill 记账,新 client 落到 + "让运行比例最贴近 probability"的那一侧,staging 的份额被写进总量、不会 + 因为运气差而缺席。 + + - 同一 (client, skill, staging_sha, probability) → 同一 side(钉死,保证 + 轨迹一致性)。 + - staging sha 变化(新候选 / 重建)或 probability 变化 → 该 skill 账本 + 清空、所有 client 重新均衡。 + - 线程安全(team server 并发 sync)。 + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + # skill_name -> {"staging_sha", "probability", "sides": {client_id: side}} + self._skills: dict[str, dict] = {} + + def assign(self, *, client_id: str, skill_name: str, + probability: float, staging_sha: str) -> str: + """返回 ``client_id`` 对 ``skill_name`` 应走的 side。 + + 首次见到该 (skill, staging_sha, probability) 组合时按均衡随机落点并 + 记账;之后同 client 直接回查。staging_sha / probability 变了则重置 + 该 skill 的账本后重新分配。 + """ + with self._lock: + st = self._skills.get(skill_name) + if (st is None + or st["staging_sha"] != staging_sha + or st["probability"] != probability): + st = { + "staging_sha": staging_sha, + "probability": probability, + "sides": {}, + } + self._skills[skill_name] = st + sides = st["sides"] + cached = sides.get(client_id) + if cached is not None: + return cached + n_main = sum(1 for v in sides.values() if v == "main") + n_staging = sum(1 for v in sides.values() if v == "staging") + side = self._balanced_side(n_main, n_staging, probability) + sides[client_id] = side + return side + + @staticmethod + def _balanced_side(n_main: int, n_staging: int, + probability: float) -> str: + """选让"加入本 client 后 main/staging 比例最接近 probability"的 side。 + + - probability≤0 → main;≥1 → staging(与 pick_side 边界一致)。 + - 首个 client(无历史)→ 纯随机按 probability 取,给后续均衡留种子。 + - 否则比较"加 staging"与"加 main"后的比例误差,取误差小的;误差相等 + (典型 probability=0.5 的奇数位)随机破平局。 + """ + if probability <= 0: + return "main" + if probability >= 1: + return "staging" + total = n_main + n_staging + if total == 0: + return "staging" if random.random() < probability else "main" + ratio_if_staging = (n_staging + 1) / (total + 1) + ratio_if_main = n_staging / (total + 1) + err_staging = abs(ratio_if_staging - probability) + err_main = abs(ratio_if_main - probability) + if err_staging < err_main: + return "staging" + if err_main < err_staging: + return "main" + return "staging" if random.random() < 0.5 else "main" + + def reset(self) -> None: + """清空全部账本(测试 / 生命周期重置用)。""" + with self._lock: + self._skills.clear() + + def read_skill_on_branch(skill_dir: Path, branch: str) -> str | None: """读取指定分支上的 SKILL.md 文本。不切分支,用 git show。""" code, out, _ = run_git(["show", f"{branch}:SKILL.md"], cwd=str(skill_dir)) diff --git a/src/xskill/team/server/client_registry.py b/src/xskill/team/server/client_registry.py index f23221fe..6e49c0d4 100644 --- a/src/xskill/team/server/client_registry.py +++ b/src/xskill/team/server/client_registry.py @@ -4,7 +4,7 @@ ux_score 明细。这个文件是第一样。 client_id 是 server 生成的 uuid——它同时是 ① canary 分桶 key(喂 -pick_side)② 上传轨迹的落盘分桶(clients//sessions/)③ +CanaryRouter.assign)② 上传轨迹的落盘分桶(clients//sessions/)③ 手改分支命名(user-staging/)。 """ from __future__ import annotations diff --git a/src/xskill/team/server/skill_manifest.py b/src/xskill/team/server/skill_manifest.py index 9b148cf8..ad4943f0 100644 --- a/src/xskill/team/server/skill_manifest.py +++ b/src/xskill/team/server/skill_manifest.py @@ -1,16 +1,19 @@ """skill_manifest.py — 给一个 client 现算它该持有的 ≤100 个 skill slot(SP1) -server 端**不存"账本表"**。manifest = ``pick_side`` 纯函数 + skill git -状态(has_staging / main_sha / staging_sha)的实时投影,每次 sync 现算。 +server 端的 skill 槽位投影不存表:ranked/recommended 排序 + skill git 状态 +(has_staging / main_sha / staging_sha)每次 sync 现算。**唯一例外是灰度 side +决策**——它由 ``CanaryRouter`` 有状态记账(见下),因为无状态哈希在 client 基数 +很小时会把 staging 饿死到 0。 slot 结构 = 80 ranked + 20 recommended: - ranked —— 按 ux_score(main 侧近 30 天均分)滑窗取高分。 - recommended —— SP3 = 用户画像质心推荐位:基于该 client 用过的 skill 的质心, - 从候选里取 cosine 最近邻(``profile_reco.py``)。无画像 - (冷启动)或非 team server 调用 → 退回 ux 排序往下取。 + 从候选里取 cosine 最近邻(``profile_reco.py``)。无画像 + (冷启动)或非 team server 调用 → 退回 ux 排序往下取。 -灰度归因:某 skill 有 staging 分支 → side = pick_side(client_id, name, p), -确定性伪随机,同 client 同 skill 在整轮灰度内 side 钉死。无 staging → main。 +灰度归因:某 skill 有 staging 分支 → side = CanaryRouter.assign(client_id, +name, p),按 client 随机分配并保证 staging 总量份额(同 client 同 staging +版本内 side 钉死)。无 staging → main。 """ from __future__ import annotations @@ -18,13 +21,18 @@ import time from pathlib import Path -from xskill.canary import has_staging, main_sha, pick_side, staging_sha +from xskill.canary import CanaryRouter, has_staging, main_sha, staging_sha from xskill.skill.skill import Skill from xskill.skill.repo import SkillRepo from xskill.team.shared.protocol import SkillSlot, SyncResponse _logger = logging.getLogger("xskill.team.manifest") +# team-CS manifest 路径的有状态灰度分流器:按 client 随机分配并保证 staging +# 总量份额(见 CanaryRouter)。module 级单例——team server 单进程,跨 sync 持续 +# 记账;server 重启后账本清空、client 在下次 sync 重新均衡(可接受)。 +_ROUTER = CanaryRouter() + def _rank_key(skill: Skill) -> tuple[float, int]: """排序键:(main 侧近 30 天 ux 均分, use_count),都缺则 (0.0, 0)。""" @@ -35,8 +43,12 @@ def _rank_key(skill: Skill) -> tuple[float, int]: def _resolve_slot(skill: Skill, client_id: str, probability: float, bucket: str) -> SkillSlot: """对一个 skill 现算它对该 client 的 side + sha。""" if has_staging(skill.path): - side = pick_side(client_id, skill.name, probability) - sha = staging_sha(skill.path) if side == "staging" else main_sha(skill.path) + s_sha = staging_sha(skill.path) + side = _ROUTER.assign( + client_id=client_id, skill_name=skill.name, + probability=probability, staging_sha=s_sha, + ) + sha = s_sha if side == "staging" else main_sha(skill.path) else: side = "main" sha = main_sha(skill.path) diff --git a/tests/test_canary.py b/tests/test_canary.py index 076e7f4a..dd503204 100644 --- a/tests/test_canary.py +++ b/tests/test_canary.py @@ -331,3 +331,84 @@ def test_jam_threshold_default_is_50(): def test_jam_threshold_read_from_dict(): assert canary.CanaryConfig.from_dict({"jam_threshold": 30}).jam_threshold == 30 + + +# ────────────────────────────────────────────────────── +# CanaryRouter —— 有状态均衡随机分流 +# ────────────────────────────────────────────────────── + +def test_canary_router_locks_client_side(): + """同 client 同 (staging_sha, probability) → 永远同一 side。""" + r = canary.CanaryRouter() + sides = {r.assign(client_id="c1", skill_name="s", probability=0.5, + staging_sha="sha-X") for _ in range(10)} + assert len(sides) == 1 # 10 次调用 side 钉死 + + +def test_canary_router_probability_zero_all_main(): + r = canary.CanaryRouter() + for i in range(10): + assert r.assign(client_id=f"c{i}", skill_name="s", probability=0.0, + staging_sha="sha") == "main" + + +def test_canary_router_probability_one_all_staging(): + r = canary.CanaryRouter() + for i in range(10): + assert r.assign(client_id=f"c{i}", skill_name="s", probability=1.0, + staging_sha="sha") == "staging" + + +def test_canary_router_guarantees_staging_share_p05(): + """p=0.5 + 3 client:staging 必拿到 1~2 个,永远不会被饿死到 0。 + + 这是修复的核心:无状态 pick_side 在 3 client 下有 1/8 概率全 main。 + """ + for _ in range(200): # 200 次独立试验(首个 client 随机) + r = canary.CanaryRouter() + sides = [r.assign(client_id=f"c{i}", skill_name="s", probability=0.5, + staging_sha="sha") for i in range(3)] + n_staging = sides.count("staging") + assert 1 <= n_staging <= 2 # 份额被锁进总量,不会是 0 或 3 + + +def test_canary_router_guarantees_staging_share_p02(): + """p=0.2 + 5 client:恰好 1 个 staging(≈ 20% 份额,确定性收敛)。""" + for _ in range(200): + r = canary.CanaryRouter() + sides = [r.assign(client_id=f"c{i}", skill_name="s", probability=0.2, + staging_sha="sha") for i in range(5)] + assert sides.count("staging") == 1 + + +def test_canary_router_ratio_tracks_probability_many_clients(): + """100 client 下运行比例贴近 probability(误差 ≤ 1 个)。""" + import random as _r + for p in (0.2, 0.3, 0.5, 0.7): + _r.seed(0) + r = canary.CanaryRouter() + sides = [r.assign(client_id=f"c{i}", skill_name="s", probability=p, + staging_sha="sha") for i in range(100)] + ratio = sides.count("staging") / 100 + assert abs(ratio - p) <= 0.02 # 误差 ≤ 2 个 client + + +def test_canary_router_resets_on_staging_sha_change(): + """staging sha 变了 → 账本重置,client 可被重新分配到对面。""" + r = canary.CanaryRouter() + first = r.assign(client_id="c1", skill_name="s", probability=1.0, + staging_sha="sha-OLD") + assert first == "staging" + # 新 staging 版本 → 重置 → c1 重新分配;p=0 强制 main 验证确实重算了 + second = r.assign(client_id="c1", skill_name="s", probability=0.0, + staging_sha="sha-NEW") + assert second == "main" + + +def test_canary_router_resets_on_probability_change(): + """probability 变了 → 重算;p=1.0 即使该 client 之前被分到 main 也变 staging。""" + r = canary.CanaryRouter() + r.assign(client_id="c1", skill_name="s", probability=0.0, staging_sha="sha") + forced = r.assign(client_id="c1", skill_name="s", probability=1.0, + staging_sha="sha") + assert forced == "staging" diff --git a/tests/test_team_skill_manifest.py b/tests/test_team_skill_manifest.py index 83dfda2d..c34c960e 100644 --- a/tests/test_team_skill_manifest.py +++ b/tests/test_team_skill_manifest.py @@ -8,7 +8,7 @@ from xskill.pipeline.atom import AtomTask, AtomTaskStore from xskill.team.server.profile_reco import ClientProfileRecommender -from xskill.team.server.skill_manifest import build_manifest +from xskill.team.server.skill_manifest import build_manifest, _ROUTER def _git(args, cwd): @@ -110,6 +110,26 @@ def test_manifest_staging_side_is_deterministic_per_client(tmp_path): assert forced.side == "staging" +def test_manifest_guarantees_staging_traffic_with_few_clients(tmp_path): + """3 个 worker + 唯一 staging skill:至少 1 个 worker 拿到 staging。 + + 这是 board6 灰度饿死 bug 的回归测试——旧的无状态 pick_side 会把 3 个 + client 全哈希到 main,staging 流量为 0。CanaryRouter 把 staging 份额锁进 + 总量,保证 ≥1。 + """ + _ROUTER.reset() + skill_dir = tmp_path / "skill" + skill_dir.mkdir() + _make_skill(skill_dir, "only-skill", with_staging=True) + sides = [ + build_manifest(client_id=f"worker-{i}", skill_dir=skill_dir, + probability=0.5, ranked_slots=80, total_slots=100).slots[0].side + for i in range(3) + ] + assert sides.count("staging") >= 1 + _ROUTER.reset() + + # ═══════════════════════════════════════════════════════════════════ # SP3 — 画像推荐(profile_reco) # ═══════════════════════════════════════════════════════════════════