15.3 修复流程与回滚预案

识别出事、归类出事之后,最后一节回答出事那一刻该做什么。先把处置拆成发现、止血、定位、修复、补偿、复盘六步,再把最难的决策——回滚——切成三条边界:哪些能回滚、哪些只能补偿、哪些只能接受。配一段可跑的仿真,把「发多了资源」的一次事故在三种处置下算成货币通胀与玩家损失,并给出等权代价最小的推荐。

15.3 修复流程与回滚预案

前两节做完了「发现」和「归类」。但事故真正发生的那一刻,没有人会按案例库的顺序出牌——补偿脚本还在跑、资源还在发、玩家群已经刷屏。这一节给的是一套固定的动作,让慌乱中的人有章可循。它承接经济系统运营 checklist 里「出事要有预案」这条要求,把它落成一个能算数字的模型:一次「发多了资源」的事故,在三种处置下经济会变成什么样、玩家会损失多少。

六步处置流程

事故处置的顺序比内容重要。乱序处置会把一次小事故放大成一次大事故:

  1. 发现:从 15.1 的检测器、8.3 的告警看板或玩家反馈里收到信号。这一步只要确认信号存在,不需要确认它是不是真的。
  2. 止血:第一步永远是止血,不是查根因。能关的开关先关(关掉产出补算、关掉领奖接口、暂停交易),能冻结的先冻结。让「还在变多」停下来,比知道「为什么变多」紧急得多。
  3. 定位:止血之后才查根因。这一节的前提是 9.3 的账本模型——如果每一笔资源变动都记了账,定位就是把异常批次在账本里捞出来,而不是猜。
  4. 修复:改代码、改配置、重算数据。修复必须可回放:同一个输入重跑,结果一致。
  5. 补偿:给受影响的人一个交代。补偿要口径统一、可对账、可幂等——15.2 里「补偿发多」本身就是一类事故,别用一次事故的处置制造下一次事故。
  6. 复盘:把「症状/根因/影响面/处置」填进案例库(15.2)。复盘的目的不是追责,是让下一次同类事故的定位时间从半天缩短到十分钟。

回滚的三条边界

六步里最难的是第 4 步的一个子问题:要不要回滚。回滚是「把数据改回事故发生前的状态」,听起来干脆,实际上只有三种资源符合不同的边界:

边界适用对象为什么
能回滚尚未被消费的批量数据(未领取的邮件、未结算的积分)数据还在,改回去不影响别人
只能补偿已经流入玩家存量、但影响可估的(多发但没花掉的资源)硬扣会制造负余额与投诉,补偿更划算
只能接受已经进入流通、无法逐笔追溯的(花掉的资源、成交过的交易)追溯成本高于损失本身,只能靠后续回收活动慢慢吸收

判断的依据只有一个:这份数据「动过没有」。 没动过就能回滚,动过就只能补偿或接受。把这条边界想清楚,事故现场就不会陷入「到底能不能回滚」的拉扯——答案在数据本身。

把事故参数写成配置

下面这段是 D 层配置:一次「补偿邮件重复发放、多发一批资源」的事故参数,加三种处置策略。金额单位是通用货币「金铢」,全部是整数。

{
  "version": 1,
  "game": "王朝纪元",
  "note": "一次「发多了资源」事故的处置预案唯一真源。全部为整数,金额单位是通用货币「金铢」。三种处置策略用同一组事故参数算经济影响与玩家损失。",
  "incident": {
    "id": "comp-overissue",
    "name": "补偿邮件重复发放,多发一批资源",
    "affectedPlayers": 2000,
    "extraPerPlayer": 100000,
    "serverActivePlayers": 50000,
    "totalGoldSupply": 800000000,
    "spentFractionPct": 45,
    "detectDelayHours": 6
  },
  "strategies": [
    {
      "id": "full_clawback",
      "name": "全额回收",
      "recoverPct": 100,
      "allowDebt": true,
      "compensationPerUnaffected": 0,
      "note": "把多发资源全部追回,已花掉的从当前余额里扣,允许余额为负"
    },
    {
      "id": "proportional",
      "name": "按比例回收",
      "recoverPct": 70,
      "allowDebt": false,
      "compensationPerUnaffected": 0,
      "note": "只回收未花掉部分的一部分,不允许余额为负,超出部分放弃"
    },
    {
      "id": "compensate_only",
      "name": "不回滚只补偿",
      "recoverPct": 0,
      "allowDebt": false,
      "compensationPerUnaffected": 2000,
      "note": "多发资源留在经济里,改为给未受影响玩家发等值补偿券"
    }
  ]
}

三种处置,算成一笔账

下面这段是 S 层仿真脚本。它读上面这份配置,对三种策略分别算实际追回、残留在经济里的超发、补偿注入、通胀率与玩家损失。核心模型只有两条:能追回的不能超过玩家手里还剩的(否则就变成负债),给未受影响玩家的补偿本身也是一次货币注入。

# file: sim/ch15_rollback.py
# tier: S
# verify: python3 sim/ch15_rollback.py --config config/
"""发多了资源:三种处置策略的经济影响与玩家损失。

读 config/ch15_rollback.json 的一组事故参数,对「全额回收 / 按比例回收 / 不回滚只补偿」
三种策略分别算:实际追回、残留在经济里的超发、补偿注入、货币供给抬升与通胀率、
被扣成负余额的玩家损失。产出 out/ch15_rollback.json 与 out/ch15_rollback.csv。
纯标准库,零依赖。
"""

from __future__ import annotations

import argparse
import csv
import json
import os


def evaluate(inc: dict, st: dict) -> dict:
    affected = inc["affectedPlayers"]
    extra_per = inc["extraPerPlayer"]
    active = inc["serverActivePlayers"]
    supply = inc["totalGoldSupply"]
    spent_pct = inc["spentFractionPct"]

    extra_total = affected * extra_per
    holdings_total = extra_total * (100 - spent_pct) // 100          # 未花掉、还握在玩家手里的部分
    attempted = extra_total * st["recoverPct"] // 100

    if st["allowDebt"]:
        recovered = min(attempted, extra_total)
        negative_total = max(0, recovered - holdings_total)          # 超出持有、只能记为负债
    else:
        recovered = min(attempted, holdings_total)                    # 不允许负余额,超出部分放弃
        negative_total = 0

    extra_remaining = extra_total - recovered
    compensation_total = st["compensationPerUnaffected"] * (active - affected)
    injection = extra_remaining + compensation_total
    negative_players = affected if negative_total > 0 else 0
    return {
        "id": st["id"], "name": st["name"], "recoverPct": st["recoverPct"],
        "extraTotal": extra_total, "recovered": recovered,
        "recoveredPct": recovered * 100 // extra_total if extra_total else 0,
        "extraRemaining": extra_remaining, "compensationTotal": compensation_total,
        "injection": injection, "supplyAfter": supply + injection,
        "inflationPct": round(injection * 100 / supply, 3) if supply else 0.0,
        "negativeBalanceTotal": negative_total, "negativePlayers": negative_players,
        "avgDebtPerPlayer": negative_total // negative_players if negative_players else 0,
        "playerHarmPct": round(negative_total * 100 / extra_total, 2) if extra_total else 0.0,
    }


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    args = ap.parse_args()

    with open(os.path.join(args.config, "ch15_rollback.json"), encoding="utf-8") as f:
        cfg = json.load(f)

    inc = cfg["incident"]
    rows = [evaluate(inc, st) for st in cfg["strategies"]]
    report = {"version": cfg["version"], "game": cfg["game"], "incident": inc, "strategies": rows}

    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "ch15_rollback.json"), "w", encoding="utf-8") as f:
        json.dump(report, f, ensure_ascii=False, indent=2)
    with open(os.path.join(args.out, "ch15_rollback.csv"), "w", encoding="utf-8", newline="") as f:
        w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
        w.writeheader()
        w.writerows(rows)
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

本机实跑(python3 sim/ch15_rollback.py --config config/):

affected=2000 extraPerPlayer=100000 supply=800000000 spentPct=45
strategy          recovered  extraLeft  injection   infl%   negTotal  harm%
full_clawback     200000000          0          0     0.0   90000000   45.0
proportional      110000000   90000000   90000000   11.25          0    0.0
compensate_only           0  200000000  296000000    37.0          0    0.0

(同机相对量级,绝对数值不可移植。)

三种处置的账很清楚:全额回收把多发资源全清了(通胀 0%),代价是 2000 个玩家被扣成负余额、合计 9000 万金铢的负债——那是「只能接受」的部分被硬当成「能回滚」来处置;按比例回收允许玩家留下未花掉部分的 30%,追回 1.1 亿,经济里还剩 9000 万(通胀 11.25%),但零玩家负债;不回滚只补偿让多发资源全留在经济里,再给 4.8 万未受影响玩家各发 2000 金铢补偿,合计注入 2.96 亿,通胀冲到 37%。

用基准给处置排序

三种处置各有各的疼,需要一个统一的尺子。下面这段是 B 层基准脚本,它把「无事故」当作基准,给每个策略算一个等权代价 score = 通胀率 + 玩家损失占比,越小越好。B 层只打软告警、不否决。

# file: sim/ch15_rollback_baseline.py
# tier: B
# verify: python3 sim/ch15_rollback_baseline.py --config config/
"""回滚处置的基准对照(B 层,软告警不硬失败)。

把「无事故」当作基准(货币供给不变、玩家零损失),逐策略算它相对基准的偏差:
  - 通胀偏差 inflationPct(超发+补偿注入占基准供给的比例)
  - 玩家损失偏差 playerHarmPct(被扣成负余额占超发总额的比例)
  - 综合代价 score = inflationPct + playerHarmPct(等权)
任一策略越过软阈值即打 WARN,但仍返回 0——B 层只提示,不否决。
"""

from __future__ import annotations

import argparse
import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from ch15_rollback import evaluate  # noqa: E402

INFLATION_SOFT_PCT = 15
HARM_SOFT_PCT = 10


def run(cfg: dict) -> dict:
    inc = cfg["incident"]
    rows = [evaluate(inc, st) for st in cfg["strategies"]]
    baseline_supply = inc["totalGoldSupply"]

    scored = []
    for r in rows:
        score = round(r["inflationPct"] + r["playerHarmPct"], 3)
        warns = []
        if r["inflationPct"] > INFLATION_SOFT_PCT:
            warns.append(f"通胀 {r['inflationPct']}% 超软阈值 {INFLATION_SOFT_PCT}%")
        if r["playerHarmPct"] > HARM_SOFT_PCT:
            warns.append(f"玩家损失占比 {r['playerHarmPct']}% 超软阈值 {HARM_SOFT_PCT}%")
        scored.append({
            "id": r["id"], "name": r["name"],
            "inflationPct": r["inflationPct"], "playerHarmPct": r["playerHarmPct"],
            "score": score, "warnings": warns,
        })

    best = min(scored, key=lambda s: s["score"])
    return {
        "version": cfg["version"],
        "game": cfg["game"],
        "baseline": {"supply": baseline_supply, "injection": 0, "playerHarmPct": 0.0},
        "softThresholds": {"inflationPct": INFLATION_SOFT_PCT, "playerHarmPct": HARM_SOFT_PCT},
        "strategies": scored,
        "recommended": best["id"],
        "warnCount": sum(len(s["warnings"]) for s in scored),
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default="config/")
    ap.add_argument("--out", default="out/")
    args = ap.parse_args()

    with open(os.path.join(args.config, "ch15_rollback.json"), encoding="utf-8") as f:
        cfg = json.load(f)

    rep = run(cfg)
    os.makedirs(args.out, exist_ok=True)
    with open(os.path.join(args.out, "ch15_rollback_baseline.json"), "w", encoding="utf-8") as f:
        json.dump(rep, f, ensure_ascii=False, indent=2)

    for s in rep["strategies"]:
        print(f"{s['name']:<10} score={s['score']:<7} 通胀={s['inflationPct']}% 损失={s['playerHarmPct']}%")
        for w in s["warnings"]:
            print(f"  WARN {s['name']}: {w}")
    print(f"推荐策略:{rep['recommended']}(score 最小)  WARN 合计 {rep['warnCount']}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

本机实跑(python3 sim/ch15_rollback_baseline.py --config config/):

全额回收       score=45.0    通胀=0.0% 损失=45.0%
  WARN 全额回收: 玩家损失占比 45.0% 超软阈值 10%
按比例回收      score=11.25   通胀=11.25% 损失=0.0%
不回滚只补偿     score=37.0    通胀=37.0% 损失=0.0%
  WARN 不回滚只补偿: 通胀 37.0% 超软阈值 15%
推荐策略:proportional(score 最小)  WARN 合计 2

等权代价把三种处置排了序:proportional(11.25)< compensate_only(37.0)< full_clawback(45.0)。全额回收的代价反而最高,因为它把 9000 万「只能接受」的损失硬扣到玩家头上——这正是三条边界里最容易被违反的一条。这个排序会随权重变化:如果这家公司更怕通胀、更不在乎投诉,把玩家损失的权重调低,结论就会向 full_clawback 移动。模型不替你拍板,它只把拍板的代价算给你看。

本节小结

这是卷一最后一节正文。它给事故处置收了一个口:

  1. 六步流程:发现、止血、定位、修复、补偿、复盘——止血永远排在定位前面,顺序错了会把小事故放大。
  2. 回滚的三条边界:没动过的数据能回滚,动过但可估的只能补偿,进入流通无法追溯的只能接受——判据是「这份数据动过没有」。
  3. 处置可以算成账:三种策略在同一组事故参数下算出通胀率与玩家损失,等权代价把「全额回收」排到了最后——因为它违反了「只能接受」那条边界。

把 15.1 的检测、15.2 的案例库、15.3 的处置流程连起来,就是一条从「发现」到「复盘」的完整闭环。而这条闭环能不能跑得快,取决于一件更早的事:每一笔资源变动有没有记账(9.3)、产出与消耗有没有跑过闭环仿真(12 章)、通胀有没有被提前看见(13 章)。风控不是经济系统之外的一层壳,它是经济系统设计质量的验收器——账记得清楚、闭环算得准,事故就定位得快。

阅读导航:上一节:15.2 经济事故案例库 · 下一节:回到目录 。

继续阅读

探索更多技术文章

浏览归档,发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「slg」更多文章

  1. 12.3 留存与付费的联合分析
  2. 12.2 流失预警与召回
  3. 附录 C:全书公式、参数与数据集索引