#!/usr/bin/env python3 # -*- coding: utf-8 -*- """m11b_selftest —— M11b 端到端自测（响应 QC #3：逐项打印 PASS/FAIL，可采信）。 三层： A. 离线单元：partitions / broadcast / latency / tx_write 守卫 / m11b_api 签名 / rtx_db 事务原语探测 + **契约三处同步核对**（实现 / __init__ 导出 / init 注册） B. 内存事务引擎端到端：把 rtx_db 的 transaction/q_all/dbname 换成**可回滚的** FakeEngine（staged writes + begin/commit/rollback + 故障注入）， 跑真实的 tx_write.apply_runtime_event 主链路，覆盖 5 类核心承诺： 正常写入 / 幂等重放 / 乐观锁冲突回滚且无脏数据 / 广播严格在 commit 之后 （无幽灵更新）/ poll 缓冲未命中或有缺口 → 退回查库 C. 真实库端到端（--live 且能连库时跑）：原子性回滚、幂等、poll、SLA 实测。 连不上时打印 SKIP 并显式声明「环境受限未验证」，绝不以 A/B 冒充 C。 用法：python3 scripts/m11b_selftest.py [--live] 退出码：0 = 无 FAIL；1 = 有 FAIL。 """ import asyncio import os import re import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) if ROOT not in sys.path: sys.path.insert(0, ROOT) from pbl_runtime_ext import broadcast as bcast # noqa: E402 from pbl_runtime_ext import latency as lat # noqa: E402 from pbl_runtime_ext import m11b_api as mapi # noqa: E402 from pbl_runtime_ext import partitions as part # noqa: E402 from pbl_runtime_ext import rtx_db # noqa: E402 from pbl_runtime_ext import tx_write as wtx # noqa: E402 RESULTS = [] def check(name, cond, detail=""): ok = bool(cond) RESULTS.append((name, ok, detail)) print("%-56s %s %s" % (name[:56], "PASS" if ok else "FAIL", detail)) return ok # ================================================================ A 离线单元 def suite_offline(): print("\n--- A. 离线单元 ---") for mod, label in ((part, "partitions"), (bcast, "broadcast"), (lat, "latency"), (wtx, "tx_write"), (mapi, "m11b_api"), (rtx_db, "rtx_db")): ok, msgs = mod.self_check() for m in msgs: check("%s | %s" % (label, m[:46]), "FAIL" not in m, m) check("%s.self_check 整体" % label, ok) pkg = os.path.join(ROOT, "pbl_runtime_ext") init_src = open(os.path.join(pkg, "__init__.py"), encoding="utf-8").read() reg_src = open(os.path.join(pkg, "init.py"), encoding="utf-8").read() for name in ("apply_runtime_event", "poll_events", "read_states", "assert_append_only", "pbl_runtime_broadcast_pull", "pbl_runtime_broadcast_stats"): impl = hasattr(mapi, name) or hasattr(wtx, name) exported = name in init_src registered = name in reg_src check("A[三处同步] %s" % name, impl and exported and registered, "impl=%s export=%s register=%s" % (impl, exported, registered)) import importlib pkg_init = importlib.import_module("pbl_runtime_ext") for name in ("pbl_runtime_broadcast_pull", "pbl_runtime_broadcast_stats", "apply_runtime_event", "poll_events", "read_states", "assert_append_only"): check("A[包出口可导入] %s" % name, callable(getattr(pkg_init, name))) _init = __import__("pbl_runtime_ext.init", fromlist=["init"]) ok_b, miss_b, names_b = _init.check_registrations(("m11b",)) check("A[M11b 注册清单完整（fail-closed 门禁）]", ok_b, "missing=%s total=%d" % (miss_b, len(names_b))) ok_a, miss_a, names_a = _init.check_registrations(("m11a",)) try: import pbl_common.api # noqa: F401 have_common = True except Exception: have_common = False if ok_a: check("A[M11a 注册清单完整]", True, "total=%d" % len(names_a)) elif not have_common: check("A[M11a 注册·环境受限未验证(pbl_common 不可导入)]", True, "missing=%s（真实宿主有 pbl_common，load 时 fail-closed 核对）" % miss_a) else: check("A[M11a 注册清单完整]", False, "missing=%s" % miss_a) # ================================================================ B 内存事务引擎 def _split_where(clause, params): """'tenant_id=%s AND seq_no > %s' + params → [(col, op, value), ...]""" out, i = [], 0 for piece in re.split(r"\s+AND\s+", clause.strip()): m = re.match(r"^`?(\w+)`?\s*(>=|<=|>|<|=)\s*(.+)$", piece.strip()) if not m: continue col, op, raw = m.group(1), m.group(2), m.group(3).strip() if raw.startswith("%s"): raw = params[i] if i < len(params) else None i += 1 else: try: raw = int(raw) except (TypeError, ValueError): raw = raw.strip("'\"") out.append((col, op, raw)) return out, i def _hit(row, conds): for col, op, val in conds: rv = row.get(col) if op == "=" and str(rv) != str(val): return False if op == ">": try: if float(rv) <= float(val): return False except (TypeError, ValueError): return False if op == ">=": try: if float(rv) < float(val): return False except (TypeError, ValueError): return False return True class FakeEngine(object): """可回滚内存 sqlor 替身：staged writes，commit 才落库，支持故障注入。""" def __init__(self): self.tables = {} self.seq = 0 self.pending = [] # ('insert', tbl, row) self.log = [] self.hub = None self.fail_on = None # (table, 'insert') → C() 抛错 # -- 事务原语（rtx_db 探测的 mode="sor"）------------------------------ def begin(self): self.log.append(("begin", self.hub.latest_cursor() if self.hub else 0)) def commit(self): for kind, tbl, row in self.pending: if kind == "insert": self.tables.setdefault(tbl, []).append(row) self.log.append(("commit", self.hub.latest_cursor() if self.hub else 0)) self.pending = [] def rollback(self): self.log.append(("rollback", len(self.pending))) self.pending = [] # -- sqlor 规范 API ---------------------------------------------------- def C(self, tbl, row): if self.fail_on == (tbl, "insert"): raise RuntimeError("注入故障：%s INSERT 失败" % tbl) self.seq += 1 row = dict(row) row["id"] = self.seq self.pending.append(("insert", tbl, row)) return self.seq def U(self, tbl, values, where=""): conds, _ = _split_where(where, ()) n = 0 for r in self.committed_rows(tbl): if _hit(r, conds): r.update(dict(values)) n += 1 return n def R(self, tbl, where="", fields="*", order="", limit=100): conds, _ = _split_where(where, ()) return [dict(r) for r in self.committed_rows(tbl) if _hit(r, conds)][:limit] def sqlExe(self, sql, params=None): return self._exec(str(sql), tuple(params or ())) def committed_rows(self, tbl): return self.tables.get(tbl, []) # -- 极简 SQL 执行（只覆盖 tx_write 实际发出的语句形态）---------------- def _exec(self, s, params): s = " ".join(s.split()) m = re.match(r"^SELECT COALESCE\(MAX\(seq_no\),\s*0\) AS max_seq FROM (\w+) " r"WHERE (.+)$", s, re.I) if m: tbl, clause = m.group(1), m.group(2) conds, _ = _split_where(clause, params) rows = [r for r in self.committed_rows(tbl) if _hit(r, conds)] return [{"max_seq": max([int(r.get("seq_no") or 0) for r in rows], default=0)}] m = re.match(r"^SELECT (.+?) FROM (\w+) WHERE (.+?)" r"(?:\s+ORDER BY\s+(.+?))?(?:\s+LIMIT\s+(\d+))?\s*$", s, re.I) if m: fields, tbl, clause, order, limit = m.groups() conds, _ = _split_where(clause, params) rows = [dict(r) for r in self.committed_rows(tbl) if _hit(r, conds)] if order and "seq_no" in order: rows.sort(key=lambda r: int(r.get("seq_no") or 0), reverse="DESC" in order.upper()) elif order and "state_key" in order: rows.sort(key=lambda r: str(r.get("state_key") or "")) if limit: rows = rows[:int(limit)] want = [f.strip().strip("`") for f in fields.split(",")] if want != ["*"]: rows = [dict((k, r.get(k)) for k in want) for r in rows] return rows m = re.match(r"^UPDATE (\w+) SET (.+?) WHERE (.+)$", s, re.I) if m: tbl, set_part, clause = m.groups() cols = [c.strip().strip("`") for c in set_part.split(",")] vals = list(params[:len(cols)]) if self.fail_on == (tbl, "update"): raise RuntimeError("注入故障：%s UPDATE 失败" % tbl) conds, _ = _split_where(clause, tuple(params[len(cols):])) n = 0 for r in self.committed_rows(tbl): if _hit(r, conds): r.update(dict(zip(cols, vals))) n += 1 return n raise AssertionError("FakeEngine 不支持的 SQL：%s" % s) class TxAdapter(object): """把 FakeEngine 适配成 rtx_db.Transaction 接口 → tx_write 走真实代码路径。""" def __init__(self, engine): self.e = engine self.committed = False self.rolled_back = False self.stmt_count = 0 async def __aenter__(self): self.e.begin() return self async def __aexit__(self, exc_type, exc, tb): if exc_type is not None or not self.committed: await self.rollback() return False async def execute(self, sql, params=None): return self.e.sqlExe(sql, params or ()) async def insert(self, table, row): self.stmt_count += 1 return self.e.C(table, row) async def update(self, table, values, where): self.stmt_count += 1 return self.e.U(table, values, where) async def select(self, table, where="", fields="*", order="", limit=200): return self.e.R(table, where, fields, order, limit) async def commit(self): if self.committed or self.rolled_back: return self.e.commit() self.committed = True async def rollback(self): if self.committed or self.rolled_back: return self.e.rollback() self.rolled_back = True def install_fake(engine): rtx_db.dbname = lambda env=None: "fake_pbl_rtx" rtx_db.transaction = lambda db=None: TxAdapter(engine) async def _q_all(sql, params=None, db=None): return rtx_db._norm_rows(engine.sqlExe(sql, params or ())) rtx_db.q_all = _q_all wtx.rtx_db = rtx_db def suite_memory(): print("\n--- B. 内存事务引擎端到端（真实 apply_runtime_event 主链路）---") engine = FakeEngine() bcast.reset_hub() hub = bcast.get_hub() engine.hub = hub install_fake(engine) async def run(): # 1) 正常写入 r1 = await wtx.apply_runtime_event( tenant_id=7, world_id=1, session_id=11, event_type="move", payload={"x": 1, "y": 2}, state_updates=[{"state_key": "pos", "state_value": {"x": 1, "y": 2}}], snapshot={"entities": 1}, client_key="cli-A1") check("B1[正常写入] ok 且非幂等命中", r1.get("ok") and r1.get("dedup") is False, "seq_no=%s event_id=%s" % (r1.get("seq_no"), r1.get("event_id"))) check("B1[正常写入] 事件表 1 行", len(engine.committed_rows("pbl_runtime_event")) == 1) check("B1[正常写入] 状态表 1 行 v1", engine.committed_rows("pbl_entity_state")[0]["state_version"] == 1) check("B1[正常写入] 快照表 1 行", len(engine.committed_rows("pbl_world_state_snapshot")) == 1) check("B1[正常写入] 单事务多语句", (r1.get("stmt_count") or 0) >= 2, "stmt_count=%s" % r1.get("stmt_count")) hub_evts, _ = hub.pull("pbl.world.7.1", None, 10) check("B1[正常写入] 提交后广播进缓冲", len(hub_evts) == 1, "cursor=%s" % hub.latest_cursor()) # 2) 广播严格在 commit 之后（无幽灵更新） ops = [x[0] for x in engine.log] check("B2[无幽灵更新] 事务日志 begin→commit", ops[:2] == ["begin", "commit"], "log=%s" % ops) cur_at_commit = [x[1] for x in engine.log if x[0] == "commit"][0] check("B2[无幽灵更新] commit 时 hub 游标=0，广播后=1", cur_at_commit == 0 and hub.latest_cursor() == 1, "commit_cursor=%s now=%s" % (cur_at_commit, hub.latest_cursor())) # 3) 幂等重放 before = hub.latest_cursor() r2 = await wtx.apply_runtime_event( tenant_id=7, world_id=1, session_id=11, event_type="move", payload={"x": 1, "y": 2}, state_updates=[{"state_key": "pos", "state_value": {"x": 1, "y": 2}}], client_key="cli-A1") check("B3[幂等重放] dedup=True", r2.get("dedup") is True, "event_id=%s" % r2.get("event_id")) check("B3[幂等重放] 事件仍 1 行", len(engine.committed_rows("pbl_runtime_event")) == 1) check("B3[幂等重放] 未重复广播", hub.latest_cursor() == before, "cursor %s→%s" % (before, hub.latest_cursor())) # 4) 乐观锁冲突 → 整体回滚且无脏数据 n_ev = len(engine.committed_rows("pbl_runtime_event")) try: await wtx.apply_runtime_event( tenant_id=7, world_id=1, session_id=11, event_type="move", payload={"x": 9}, state_updates=[{"state_key": "pos", "state_value": {"x": 9}}], base_version=99, client_key="cli-CONFLICT") check("B4[乐观锁冲突] 必须抛 RtxError", False, "未抛异常") except rtx_db.RtxError as exc: check("B4[乐观锁冲突] PBL-STATE-CONFLICT", exc.code == "PBL-STATE-CONFLICT", exc.msg[:56]) check("B4[乐观锁冲突] 事件行数未增加（无脏数据）", len(engine.committed_rows("pbl_runtime_event")) == n_ev, "%d→%d" % (n_ev, len(engine.committed_rows("pbl_runtime_event")))) check("B4[乐观锁冲突] 冲突键未落库", not [r for r in engine.committed_rows("pbl_runtime_event") if r.get("idem_key") == "cli-CONFLICT"]) check("B4[乐观锁冲突] 状态版本未污染（仍 v1）", engine.committed_rows("pbl_entity_state")[0]["state_version"] == 1) check("B4[乐观锁冲突] 未广播", hub.latest_cursor() == before) # 5) DML 中途失败（状态 INSERT 故障注入）→ 事件行不存在（真回滚） cur = hub.latest_cursor() engine.fail_on = ("pbl_entity_state", "insert") try: await wtx.apply_runtime_event( tenant_id=7, world_id=1, session_id=12, event_type="spawn", payload={"kind": "npc"}, state_updates=[{"state_key": "hp", "state_value": 100}], client_key="cli-FAIL-DML") check("B5[DML失败] 异常上抛（不静默）", False) except RuntimeError as exc: check("B5[DML失败] 异常上抛（不静默）", True, str(exc)[:40]) engine.fail_on = None check("B5[DML失败] 事件行不存在（回滚证据，非标记 rolled_back）", not [r for r in engine.committed_rows("pbl_runtime_event") if str(r.get("session_id")) == "12"], "events=%d" % len(engine.committed_rows("pbl_runtime_event"))) check("B5[DML失败] 无状态脏行", not [r for r in engine.committed_rows("pbl_entity_state") if str(r.get("session_id")) == "12"]) check("B5[DML失败] 未广播（无幽灵更新）", hub.latest_cursor() == cur) check("B6[回滚证据] 日志含 rollback", "rollback" in [x[0] for x in engine.log], "log=%s" % [x[0] for x in engine.log]) # 7) seq 单调 + 状态版本递增 await wtx.apply_runtime_event( tenant_id=7, world_id=1, session_id=11, event_type="move", payload={"x": 5}, state_updates=[{"state_key": "pos", "state_value": {"x": 5}}], client_key="cli-A2") evs = engine.committed_rows("pbl_runtime_event") check("B7[顺序与版本] seq_no 单调 1→2", [e["seq_no"] for e in evs] == [1, 2], str([e["seq_no"] for e in evs])) check("B7[顺序与版本] state_version 递增到 2", engine.committed_rows("pbl_entity_state")[0]["state_version"] == 2) # 8) poll：hub 命中 p1 = await wtx.poll_events(tenant_id=7, world_id=1, session_id=11, after_cursor=0) check("B8[poll命中hub] source=hub", p1.get("source") == "hub", "events=%d" % len(p1.get("events") or [])) # 9) poll：hub 未命中（缓冲清空）→ 退回查库 bcast.reset_hub() p2 = await wtx.poll_events(tenant_id=7, world_id=1, session_id=11, after_seq=0) check("B9[poll退回查库] source=db", p2.get("source") == "db", "events=%d" % len(p2.get("events") or [])) check("B9[poll退回查库] 取回 2 条已提交事件", len(p2.get("events") or []) == 2, str([e.get("seq_no") for e in p2.get("events") or []])) check("B9[poll退回查库] payload 已反序列化为对象", bool(p2["events"]) and isinstance(p2["events"][0].get("payload"), dict)) # 10) poll：游标早于缓冲最旧记录（缺口）→ 退回查库 hub3 = bcast.reset_hub() for i in range(1, 8): hub3.publish("pbl.world.7.1", {"event_id": i, "seq_no": i}) p3 = await wtx.poll_events(tenant_id=7, world_id=1, session_id=11, after_cursor=0, limit=4) check("B10[缺口退查库] source=db", p3.get("source") == "db", "truncated=%s" % p3.get("truncated")) # 11) read_states st = await wtx.read_states(tenant_id=7, world_id=1, session_id=11) check("B11[read_states] 返回状态且 state_value 为对象", bool(st) and isinstance(st[0].get("state_value"), dict), "keys=%s" % [x.get("state_key") for x in st]) # 12) 缺租户 fail-closed try: await wtx.apply_runtime_event(tenant_id=None, world_id=1, session_id=11, event_type="move") check("B12[缺租户] 必须拒绝", False) except rtx_db.RtxError as exc: check("B12[缺租户] PBL-TENANT-0001", exc.code == "PBL-TENANT-0001") # 13) dspy 端点包装层 pull = await mapi.pbl_runtime_broadcast_pull(tenant_id=7, world_id=1, session_id=11, after_seq=0) check("B13[broadcast_pull] ok", pull.get("ok") is True, "source=%s count=%s" % (pull.get("source"), pull.get("count"))) stats = await mapi.pbl_runtime_broadcast_stats(tenant_id=7, world_id=1) check("B13[broadcast_stats] 含 sla/contracts", stats.get("ok") and "sla" in stats and "contracts" in stats, "published=%s" % stats.get("published")) bad = await mapi.pbl_runtime_broadcast_pull(channel="other.channel", tenant_id=None) check("B13[broadcast_pull] 无租户段拒绝跨租户拉取", bad.get("ok") is False, str(bad.get("error"))[:40]) # 14) SLA 打点链路 s = lat.summary() check("B14[SLA打点] 三阶段均有样本", s["adjudicate"]["samples"] > 0 and s["broadcast"]["samples"] > 0 and s["end_to_end"]["samples"] > 0, "adj=%s bcast=%s e2e=%s" % (s["adjudicate"]["avg_ms"], s["broadcast"]["avg_ms"], s["end_to_end"]["avg_ms"])) check("B14[SLA打点] 广播实测 ≤300ms", s["broadcast"]["max_ms"] is None or s["broadcast"]["max_ms"] <= 300, "max=%sms" % s["broadcast"]["max_ms"]) # 15) 分区预建（内存引擎无分区能力 → 必须如实 degraded，不谎报） res = await part.ensure_forward_partitions() check("B15[分区] 失败时如实降级（不谎报已建）", res.get("degraded") is True or res.get("ok") is True, "ok=%s degraded=%s err=%s" % (res.get("ok"), res.get("degraded"), str(res.get("error"))[:40])) asyncio.new_event_loop().run_until_complete(run()) # ================================================================ C 真实库 def suite_live(force=False): print("\n--- C. 真实库端到端（原子性/回滚/广播时序的权威证据）---") try: from apppublic import DBPools # noqa: F401 from ahserver.serverenv import ServerEnv # noqa: F401 except Exception as exc: # noqa: BLE001 print("SKIP C：环境受限未验证 —— 本工作空间无 ahserver/apppublic 运行时（%r）" % type(exc).__name__) print(" B 层用可回滚内存事务引擎跑通同一份 tx_write 主链路代码替代；") print(" 连库跑法：在 pbls 应用环境执行 `python3 scripts/m11b_selftest.py --live`") RESULTS.append(("C[真实库] 环境受限未验证（已声明）", True, "SKIP")) return if not force: print("SKIP C：未指定 --live（不主动写生产库）") RESULTS.append(("C[真实库] 未指定 --live，跳过", True, "SKIP")) return async def run(): bcast.reset_hub() key = "live-%d" % int(time.time() * 1000) r1 = await wtx.apply_runtime_event( tenant_id=7, world_id=9001, session_id=90011, event_type="live_move", payload={"x": 1}, state_updates=[{"state_key": "live_pos", "state_value": {"x": 1}}], client_key=key) check("C1[真实库] 单事务写入成功", bool(r1.get("ok") and r1.get("event_id")), "latency=%sms" % r1.get("latency_ms")) rows = await rtx_db.q_all("SELECT id FROM pbl_runtime_event WHERE tenant_id=%s " "AND idem_key=%s", (7, key)) check("C1[真实库] 事件已持久化", len(rows) == 1) r2 = await wtx.apply_runtime_event( tenant_id=7, world_id=9001, session_id=90011, event_type="live_move", payload={"x": 1}, client_key=key) rows2 = await rtx_db.q_all("SELECT id FROM pbl_runtime_event WHERE tenant_id=%s " "AND idem_key=%s", (7, key)) check("C2[真实库] 幂等重放不新增行", r2.get("dedup") is True and len(rows2) == 1) try: await wtx.apply_runtime_event( tenant_id=7, world_id=9001, session_id=90011, event_type="live_move", payload={"x": 2}, state_updates=[{"state_key": "live_pos", "state_value": {"x": 2}}], base_version=999999, client_key=key + "-conflict") check("C3[真实库] 乐观锁必须冲突", False) except rtx_db.RtxError as exc: check("C3[真实库] 乐观锁冲突", exc.code == "PBL-STATE-CONFLICT") bad = await rtx_db.q_all("SELECT id FROM pbl_runtime_event WHERE tenant_id=%s " "AND idem_key=%s", (7, key + "-conflict")) check("C3[真实库] 回滚后事件行不存在（原子性证据）", len(bad) == 0) p = await wtx.poll_events(tenant_id=7, world_id=9001, session_id=90011, after_seq=0) check("C4[真实库] poll 取回已提交事件", p.get("ok") and len(p.get("events") or []) >= 1, "source=%s n=%d" % (p.get("source"), len(p.get("events") or []))) s = await mapi.pbl_runtime_broadcast_stats(tenant_id=7, world_id=9001) adj = (s.get("sla") or {}).get("adjudicate") or {} bct = (s.get("sla") or {}).get("broadcast") or {} check("C5[真实库] SLA 实测 p95 在预算内", (adj.get("p95_ms") or 0) <= 200 and (bct.get("p95_ms") or 0) <= 300, "adj_p95=%s bcast_p95=%s" % (adj.get("p95_ms"), bct.get("p95_ms"))) asyncio.new_event_loop().run_until_complete(run()) def main(): suite_offline() suite_memory() suite_live(force="--live" in sys.argv) fails = [n for n, ok, _ in RESULTS if not ok] print("\n================ 汇总 ================") print("总计 %d 项，PASS %d，FAIL %d" % (len(RESULTS), len(RESULTS) - len(fails), len(fails))) for f in fails: print(" FAIL: %s" % f) print("RESULT: %s" % ("PASS" if not fails else "FAIL")) return 0 if not fails else 1 if __name__ == "__main__": sys.exit(main())