7.3 灰度、放量与回滚
内容做完、按期交付，还差最后一步：安全地上线。SLG 是全天候在线品，一次全量发布把线上打挂，损失的是正在进行的赛季和联盟战。产业侧那篇 SLG 经济系统运营 2026 清单 把「发布风控」列为运营动作；本节把它做成可跑的判定器——用分桶决定放量范围，用两级阈值决定继续、暂停还是回滚，并把整个过程输出成 CSV 与 SVG。事故后的修复与回滚流程在 15.3 展开，这里聚焦放量节奏本身。
分桶：把「放多少」变成整数问题
灰度的第一步是分桶：把玩家按稳定哈希分到 totalBuckets 个桶里，coveragePct 决定前多少个桶吃到新版本。三个硬要求：
- 稳定：同一玩家每次哈希到同一个桶，否则同一个人一会儿新一会儿旧。
- 均匀：桶分布要接近均匀，否则 1% 覆盖可能压中某个联盟的全部玩家。
- 可跨语言复现：客户端/服务端/后台脚本用同一份哈希，否则判定不一致。
《王朝纪元》用 FNV-1a 32 位：bucket = fnv1a32(playerId) % totalBuckets，isCanary = bucket < coveragePct。下面这段是分桶函数的实跑样本，TS 镜像 proto/src/ch07_canary.ts 用 Math.imul 实现同一函数，逐位一致——这是本套「一份配置两处消费」的又一次落地。
$ python3 sim/ch07_canary.py --config config/ --plan ch07_canary --sample p1001 --sample p1002
plan=ch07_canary outcome=ROLLBACK reached=100% hours=114
...
bucket(p1001)=17 canary@20%=True
bucket(p1002)=60 canary@20%=False
放量节奏与两级阈值
放量是一串阶段（stage），每阶段给出目标覆盖比例与观察时长。判定发生在每个观察窗口结束时，读四个指标：崩溃率 crashRatePct、错误率 errorRatePct、p99LatencyMs、次日留存 dayRetentionPct。阈值分两级：
- 硬阈值（
hard）：任一指标越界即 ROLLBACK，立即回滚到全量旧版本。 - 警告带（
warn）：越界但未破硬阈值即 HOLD，停在当前覆盖比例，不再放大。 - 都未越界则 GO，进入下一阶段。
下面这段是 D 层配置节选（完整文件 config/ch07_canary.json），给出阈值与两个阶段：
<!-- snippet: config/ch07_canary.json（节选） -->
{
"totalBuckets": 100,
"thresholds": {
"hard": { "maxCrashRatePct": 1.0, "maxErrorRatePct": 2.0, "maxP99LatencyMs": 800, "minDayRetentionPct": 40 },
"warn": { "maxCrashRatePct": 0.9, "maxErrorRatePct": 1.8, "maxP99LatencyMs": 780, "minDayRetentionPct": 41 }
},
"stages": [
{ "id": "S3", "coveragePct": 50, "durationHours": 24,
"telemetry": { "crashRatePct": 0.8, "errorRatePct": 1.4, "p99LatencyMs": 720, "dayRetentionPct": 43 } },
{ "id": "S4", "coveragePct": 100, "durationHours": 48,
"telemetry": { "crashRatePct": 1.6, "errorRatePct": 2.3, "p99LatencyMs": 910, "dayRetentionPct": 38 } }
]
}
telemetry 是监控窗口的观测值：真实环境里它来自监控系统（属 X 层，命令行不可复现），这里直接写进配置，让脚本专注判定策略本身。
判定器
下面这段是 S 层仿真脚本（完整文件 sim/ch07_canary.py），实现分桶、阈值判定与放量曲线，产出 JSON、CSV 与 SVG。
# file: sim/ch07_canary.py
# tier: S
# verify: python3 sim/ch07_canary.py --config config/ --plan ch07_canary
"""灰度放量仿真：分桶 + 指标阈值判定（放量 / 暂停 / 回滚）。
分桶哈希用 FNV-1a 32 位，与 proto/src/ch07_canary.ts 逐位一致（整数运算）。
监控窗口 telemetry 由 config 给定，模拟的是**判定策略**；真实线上指标属 X 层。
"""
from __future__ import annotations
import argparse
import json
import os
import sys
import xml.etree.ElementTree as ET
HERE = os.path.dirname(os.path.abspath(__file__))
OUT = os.path.join(HERE, "..", "out")
def load_config(cfg_dir: str, name: str) -> dict:
with open(os.path.join(cfg_dir, name + ".json"), encoding="utf-8") as f:
return json.load(f)
# ---------- 分桶：FNV-1a 32 位，纯整数，跨语言逐位一致 ----------
def fnv1a32(s: str, offset: int = 2166136261, prime: int = 16777619) -> int:
h = offset & 0xFFFFFFFF
for ch in s.encode("utf-8"):
h = (h ^ ch) & 0xFFFFFFFF
h = (h * prime) & 0xFFFFFFFF
return h
def bucket_of(player_id: str, total_buckets: int, offset: int = 2166136261, prime: int = 16777619) -> int:
return fnv1a32(player_id, offset, prime) % total_buckets
def is_canary(player_id: str, coverage_pct: int, total_buckets: int,
offset: int = 2166136261, prime: int = 16777619) -> bool:
return bucket_of(player_id, total_buckets, offset, prime) < coverage_pct
# ---------- 指标阈值判定 ----------
def evaluate(telemetry: dict, thresholds: dict) -> dict:
"""返回 {decision, breaches}：hard 越界→ROLLBACK，warn 越界→HOLD，否则 GO。"""
hard, warn = thresholds["hard"], thresholds["warn"]
breaches = []
rollback = False
hold = False
if telemetry["crashRatePct"] > hard["maxCrashRatePct"]:
breaches.append("crash>hard"); rollback = True
elif telemetry["crashRatePct"] >= warn["maxCrashRatePct"]:
breaches.append("crash>=warn"); hold = True
if telemetry["errorRatePct"] > hard["maxErrorRatePct"]:
breaches.append("error>hard"); rollback = True
elif telemetry["errorRatePct"] >= warn["maxErrorRatePct"]:
breaches.append("error>=warn"); hold = True
if telemetry["p99LatencyMs"] > hard["maxP99LatencyMs"]:
breaches.append("p99>hard"); rollback = True
elif telemetry["p99LatencyMs"] >= warn["maxP99LatencyMs"]:
breaches.append("p99>=warn"); hold = True
if telemetry["dayRetentionPct"] < hard["minDayRetentionPct"]:
breaches.append("retention<hard"); rollback = True
elif telemetry["dayRetentionPct"] <= warn["minDayRetentionPct"]:
breaches.append("retention<=warn"); hold = True
decision = "ROLLBACK" if rollback else ("HOLD" if hold else "GO")
return {"decision": decision, "breaches": breaches}
def simulate(cfg: dict) -> dict:
rows = []
elapsed = 0
outcome = "COMPLETE"
reached = 0
for st in cfg["stages"]:
ev = evaluate(st["telemetry"], cfg["thresholds"])
rows.append({
"stage": st["id"], "coveragePct": st["coveragePct"], "durationHours": st["durationHours"],
"startHour": elapsed, "endHour": elapsed + st["durationHours"],
"decision": ev["decision"], "breaches": ev["breaches"],
"telemetry": st["telemetry"],
})
if ev["decision"] == "ROLLBACK":
outcome = "ROLLBACK"; reached = st["coveragePct"]; elapsed += st["durationHours"]
break
if ev["decision"] == "HOLD":
outcome = "HOLD"; reached = st["coveragePct"]; elapsed += st["durationHours"]
break
reached = st["coveragePct"]
elapsed += st["durationHours"]
return {"outcome": outcome, "reachedCoveragePct": reached, "totalHours": elapsed, "stages": rows}
def write_csv(sim: dict, path: str) -> None:
with open(path, "w", encoding="utf-8") as f:
f.write("stage,coveragePct,startHour,endHour,crashRatePct,errorRatePct,p99LatencyMs,dayRetentionPct,decision\n")
for r in sim["stages"]:
t = r["telemetry"]
f.write(f"{r['stage']},{r['coveragePct']},{r['startHour']},{r['endHour']},"
f"{t['crashRatePct']},{t['errorRatePct']},{t['p99LatencyMs']},{t['dayRetentionPct']},{r['decision']}\n")
def write_svg(cfg: dict, sim: dict, path: str) -> None:
W, H = 720, 300
PAD_L, PAD_R, PAD_T, PAD_B = 56, 24, 24, 40
stages = sim["stages"]
total_h = max(sim["totalHours"], 1)
inner_w = W - PAD_L - PAD_R
inner_h = H - PAD_T - PAD_B
def x_of(h: int) -> float:
return PAD_L + inner_w * h / total_h
def y_of(cov: int) -> float:
return PAD_T + inner_h * (1 - cov / 100)
svg = ET.Element("svg", {"xmlns": "http://www.w3.org/2000/svg",
"viewBox": f"0 0 {W} {H}", "width": str(W), "height": str(H)})
ET.SubElement(svg, "rect", {"x": "0", "y": "0", "width": str(W), "height": str(H), "fill": "#ffffff"})
ET.SubElement(svg, "line", {"x1": str(PAD_L), "y1": str(PAD_T), "x2": str(PAD_L), "y2": str(H - PAD_B),
"stroke": "#333", "stroke-width": "1"})
ET.SubElement(svg, "line", {"x1": str(PAD_L), "y1": str(H - PAD_B), "x2": str(W - PAD_R),
"y2": str(H - PAD_B), "stroke": "#333", "stroke-width": "1"})
pts = [(x_of(0), y_of(0))]
color = {"GO": "#1f6feb", "HOLD": "#d29922", "ROLLBACK": "#d1242f"}
for r in stages:
pts.append((x_of(r["endHour"]), y_of(r["coveragePct"])))
ET.SubElement(svg, "polyline", {"points": " ".join(f"{x:.1f},{y:.1f}" for x, y in pts),
"fill": "none", "stroke": "#1f6feb", "stroke-width": "2"})
for r in stages:
cx, cy = x_of(r["endHour"]), y_of(r["coveragePct"])
ET.SubElement(svg, "circle", {"cx": f"{cx:.1f}", "cy": f"{cy:.1f}", "r": "4",
"fill": color[r["decision"]]})
t = ET.SubElement(svg, "text", {"x": f"{cx:.1f}", "y": f"{cy - 8:.1f}",
"font-size": "10", "text-anchor": "middle", "fill": "#333"})
t.text = f"{r['coveragePct']}%"
for cov in (0, 50, 100):
y = y_of(cov)
t = ET.SubElement(svg, "text", {"x": str(PAD_L - 8), "y": f"{y + 4:.1f}",
"font-size": "11", "text-anchor": "end", "fill": "#555"})
t.text = f"{cov}%"
ttl = ET.SubElement(svg, "text", {"x": str(PAD_L), "y": "16", "font-size": "13", "fill": "#111"})
ttl.text = f"灰度放量曲线（outcome={sim['outcome']}，reached={sim['reachedCoveragePct']}%）"
os.makedirs(os.path.dirname(path), exist_ok=True)
ET.ElementTree(svg).write(path, encoding="utf-8", xml_declaration=True)
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--config", default=os.path.join(HERE, "..", "config"))
ap.add_argument("--plan", default="ch07_canary", help="配置文件名（不含 .json）")
ap.add_argument("--sample", action="append", default=[], help="打印某玩家分桶")
args = ap.parse_args()
cfg = load_config(args.config, args.plan)
sim = simulate(cfg)
os.makedirs(OUT, exist_ok=True)
with open(os.path.join(OUT, f"{args.plan}_report.json"), "w", encoding="utf-8") as f:
json.dump(sim, f, ensure_ascii=False, indent=2)
write_csv(sim, os.path.join(OUT, f"{args.plan}.csv"))
write_svg(cfg, sim, os.path.join(OUT, f"{args.plan}.svg"))
print(f"plan={args.plan} outcome={sim['outcome']} reached={sim['reachedCoveragePct']}% hours={sim['totalHours']}")
for r in sim["stages"]:
print(f" {r['stage']} cov={r['coveragePct']:>3}% -> {r['decision']:<8} {r['breaches']}")
for pid in args.sample:
print(f" bucket({pid})={bucket_of(pid, cfg['totalBuckets'])} canary@20%={is_canary(pid, 20, cfg['totalBuckets'])}")
return 0
if __name__ == "__main__":
sys.exit(main())
基准方案：50% 直跳 100% 翻车
基准方案是五段式：1% → 5% → 20% → 50% → 100%。前四段稳，最后一段从 50% 直接跳到 100%：
$ python3 sim/ch07_canary.py --config config/ --plan ch07_canary
plan=ch07_canary outcome=ROLLBACK reached=100% hours=114
S0 cov= 1% -> GO []
S1 cov= 5% -> GO []
S2 cov= 20% -> GO []
S3 cov= 50% -> GO []
S4 cov=100% -> ROLLBACK ['crash>hard', 'error>hard', 'p99>hard', 'retention<hard']
S4 四个指标同时越硬线，判定 ROLLBACK，覆盖停在 100% 但随即回退。翻车的根因不是「100% 一定不行」，而是一步跨了 50 个百分点：负载瞬间翻倍，崩溃率、错误率、延迟、留存一起恶化。
修正方案：拆成小步
把 100% 前的跳跃拆成 50% → 70% → 100% 三小步，每步 20~30 个百分点，同样跑一遍（配置 ch07_canary_fixed.json）：
$ python3 sim/ch07_canary.py --config config/ --plan ch07_canary_fixed
plan=ch07_canary_fixed outcome=COMPLETE reached=100% hours=138
S0 cov= 1% -> GO []
S1 cov= 5% -> GO []
S2 cov= 20% -> GO []
S3 cov= 40% -> GO []
S3b cov= 70% -> GO []
S4 cov=100% -> GO []
六段全 GO，outcome=COMPLETE，reached=100%。代价是多花 24 小时（hours 从 114 涨到 138）——用时间换确定性，这正是灰度存在的意义。注意两方案的总阶段数不同（5 vs 6），说明「放量节奏」本身就是可调参数，不是固定套路。
回滚不是失败
三个要点收束本节：
- 回滚是设计出来的能力，不是事故的同义词。判定器越早触发
ROLLBACK，损失越小；把阈值调到「永不回滚」才是真危险。 - HOLD 与 ROLLBACK 要分开：警告带触发
HOLD（停手观察），硬阈值才ROLLBACK（推倒重来）。两者混用会导致要么过于激进、要么永不前进。 - 分桶哈希必须跨语言一致：本节的分桶函数在 Python 与 TS 两侧逐位一致（
p1001→17、p1002→60），任何一侧算错，判定就失去意义。
放量判定用的指标口径、埋点与看板，是第 8 章的主题——指标算错，阈值再准也白搭。
阅读导航：上一节：7.2 内容产能与排期 · 下一节：8.1 北极星指标树 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。