#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M11b 离线自检脚本：单事务事件+状态写入、幂等、回滚、提交后广播、3s 轮询兜底。 不连真实数据库（用内存 SQL 替身），验证的是**代码行为**而非环境： P1 单事务写入：事件 1 行 + 实体状态 1 行，提交后可见 P2 幂等：同 idem_key 二次调用不重复写、不重复广播（dedup=True） P3 回滚：状态写入抛错 → 事件行不落库（无脏数据） P4 提交后广播：广播只在 commit 之后发生（顺序断言） P5 广播通道异常不撤销事实，轮询兜底仍可取到事件 P6 缺租户 fail-closed P7 客户端写服务端权威字段被拒 P8 乐观锁 if_state_version 冲突返回 PBL_RT_STATE_CONFLICT P9 时延字段与 SLA 预算齐备（裁决/广播/端到端 + 3s 兜底周期） P10 append-only 守卫 + 按月分区算法 用法：python3 scripts/m11b_selftest.py （退出码 0=PASS） """ import os import re import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from pbl_runtime_ext import rtx_db as R # noqa: E402 from pbl_runtime_ext import tx_write as W # noqa: E402 from pbl_runtime_ext import broadcast as B # noqa: E402 PH = re.compile(r"\$\{(\w+)\}\$") COLS = re.compile(r"INSERT\s+INTO\s+`?(\w+)`?\s*\(([^)]*)\)", re.IGNORECASE) class Store(object): def __init__(self): self.tables = {} self.seq = 0 self.commits = [] self.rollbacks = [] self.order = [] # 事件序列（commit / broadcast / insert） def rows(self, t): return self.tables.setdefault(t, []) def insert(self, table, row): self.seq += 1 row = dict(row) row["id"] = self.seq self.rows(table).append(row) self.order.append("insert:%s" % table) return self.seq class FakeTx(object): """事务替身：exec/query 打到内存 Store，退出时 commit 或 rollback。""" def __init__(self, store, boom_on=None): self.store = store self.boom_on = boom_on self.pending = [] def __enter__(self): return self def __exit__(self, et, ev, tb): if et is None: self.store.tables = getattr(self, "_final", self.store.tables) self.store.commits.append(len(self.store.rows("pbl_runtime_event"))) self.store.order.append("commit") else: for t, rows in getattr(self, "_snapshot", {}).items(): self.store.tables[t] = rows self.store.rollbacks.append(str(ev)[:80]) self.store.order.append("rollback") return False def _begin(self): self._snapshot = {k: list(v) for k, v in self.store.tables.items()} def exec(self, sql, params=None): params = params or {} if self.boom_on and self.boom_on in sql: raise RuntimeError("injected failure on %s" % self.boom_on) m = COLS.search(sql) if m: table = m.group(1) cols = [c.strip().strip("`") for c in m.group(2).split(",")] ph = PH.findall(sql.split("VALUES", 1)[1]) if "VALUES" in sql else [] row = {} for i, c in enumerate(cols): row[c] = params.get(ph[i]) if i < len(ph) else None if table == "pbl_entity_state": key = (row.get("tenant_id"), row.get("session_id"), row.get("entity_id"), row.get("state_key")) for r in self.store.rows(table): if (r.get("tenant_id"), r.get("session_id"), r.get("entity_id"), r.get("state_key")) == key: r.update(row) self.store.order.append("insert:pbl_entity_state") return 2 self.pending.append(key) return self.store.insert(table, row) return 1 def query(self, sql, params=None): params = params or {} self._begin() if "MAX(`seq_no`)" in sql: vals = [int(r.get("seq_no") or 0) for r in self.store.rows("pbl_runtime_event") if str(r.get("tenant_id")) == str(params.get("t")) and str(r.get("session_id")) == str(params.get("s"))] return [{"s": max(vals) if vals else 0}] if "MAX(`state_version`)" in sql: vals = [int(r.get("state_version") or 0) for r in self.store.rows("pbl_entity_state") if str(r.get("tenant_id")) == str(params.get("t")) and str(r.get("session_id")) == str(params.get("s"))] return [{"v": max(vals) if vals else 0}] if "LAST_INSERT_ID" in sql: return [{"i": self.store.seq}] if "FROM `pbl_entity_state`" in sql: out = [] for r in self.store.rows("pbl_entity_state"): if (str(r.get("tenant_id")) == str(params.get("t")) and str(r.get("session_id")) == str(params.get("s")) and str(r.get("entity_id")) == str(params.get("e")) and str(r.get("state_key")) == str(params.get("k"))): out.append({"state_value": r.get("state_value"), "state_version": r.get("state_version")}) return out[:1] return [] def install_store(store): """把 rtx_db / tx_write 的 DB 出入口全部换成内存替身。""" def fake_transaction(label="runtime_tx"): tx = FakeTx(store, boom_on=install_store.boom) tx._begin() return tx def fake_q_one(sql, params=None): params = params or {} if "FROM `pbl_runtime_event`" in sql and "idem_key" in sql: for r in store.rows("pbl_runtime_event"): if (str(r.get("tenant_id")) == str(params.get("t")) and str(r.get("idem_key")) == str(params.get("i"))): return dict(r) return None def fake_q_all(sql, params=None): params = params or {} if "FROM `pbl_runtime_event`" in sql: rows = [r for r in store.rows("pbl_runtime_event") if str(r.get("tenant_id")) == str(params.get("t")) and str(r.get("session_id")) == str(params.get("s")) and int(r.get("seq_no") or 0) > int(params.get("q") or 0)] return sorted(rows, key=lambda r: int(r.get("seq_no") or 0))[:int(params.get("n") or 100)] if "FROM `pbl_entity_state`" in sql: return [dict(r) for r in store.rows("pbl_entity_state") if str(r.get("tenant_id")) == str(params.get("t")) and str(r.get("session_id")) == str(params.get("s"))] return [] def fake_sql_exec(sql, params=None): return 1 fake_transaction.__call__ = fake_transaction install_store.boom = getattr(install_store, "boom", None) W.transaction = fake_transaction R.q_one = fake_q_one R.q_all = fake_q_all R.sql_exec = fake_sql_exec W.ensure_partition_ready = lambda force=False: {"ok": True, "created": ["pbl_runtime_event_202609"], "skipped": [], "errors": []} B.set_hub(B.BroadcastHub(workers=0)) return B.get_hub() RESULTS = [] def check(name, cond, detail=""): RESULTS.append((bool(cond), name, detail)) print("%s %s %s" % ("PASS" if cond else "FAIL", name, detail)) def main(): hub = None # ---------------- P1 单事务写入 store = Store() hub = install_store(store) install_store.boom = None pushed = [] hub.register_push(lambda sk, ev: pushed.append((sk, ev)) or 1, name="probe") r1 = W.apply_runtime_event(tenant_id="T1", session_id=11, world_id=7, entity_id=101, event_type="move", payload={"x": 1, "y": 2}, state_updates=[{"entity_id": 101, "state_key": "pos", "state_value": {"x": 1, "y": 2}}], snapshot=True, actor_id=9) check("P1 单事务写入 ok", r1.get("ok") and r1.get("transaction") == "single", "seq_no=%s state_version=%s" % (r1.get("seq_no"), r1.get("state_version"))) check("P1 事件 1 行 + 状态 1 行 + 快照 1 行", len(store.rows("pbl_runtime_event")) == 1 and len(store.rows("pbl_entity_state")) == 1 and len(store.rows("pbl_world_state_snapshot")) == 1, "ev=%d st=%d sn=%d" % (len(store.rows("pbl_runtime_event")), len(store.rows("pbl_entity_state")), len(store.rows("pbl_world_state_snapshot")))) check("P1 提交发生", len(store.commits) == 1, "commits=%d" % len(store.commits)) # ---------------- P4 提交后广播（顺序） check("P4 提交后广播", "commit" in store.order and store.order.index("commit") < len(store.order) and pushed and store.order.index("commit") < len([o for o in store.order if o.startswith("insert")]) + 1, "order=%s pushed=%d" % (store.order[-2:], len(pushed))) check("P4 广播帧含 state_version", bool(pushed) and pushed[0][1].get("state_version") == r1.get("state_version")) # ---------------- P2 幂等 r2 = W.apply_runtime_event(tenant_id="T1", session_id=11, world_id=7, entity_id=101, event_type="move", payload={"x": 1, "y": 2}, state_updates=[{"entity_id": 101, "state_key": "pos", "state_value": {"x": 1, "y": 2}}], actor_id=9) check("P2 幂等命中 dedup", r2.get("dedup") is True, "idem=%s" % str(r2.get("idem_key"))[:12]) check("P2 未重复写事件", len(store.rows("pbl_runtime_event")) == 1) check("P2 未重复广播", len(pushed) == 1, "pushed=%d" % len(pushed)) # ---------------- P3 回滚无脏数据 store2 = Store() hub2 = install_store(store2) install_store.boom = "pbl_entity_state" try: W.apply_runtime_event(tenant_id="T2", session_id=21, event_type="move", payload={"a": 1}, state_updates=[{"entity_id": 5, "state_key": "pos", "state_value": {"a": 1}}]) check("P3 状态写失败必须抛错", False, "未抛异常") except R.PblRtError as exc: check("P3 状态写失败抛 TX_FAILED", exc.code == "PBL_RT_TX_FAILED", "code=%s" % exc.code) except Exception as exc: # noqa: BLE001 check("P3 状态写失败抛错", True, "%r" % exc) check("P3 回滚后事件不落库（无脏数据）", len(store2.rows("pbl_runtime_event")) == 0 and len(store2.rollbacks) > 0, "ev=%d rollbacks=%d" % (len(store2.rows("pbl_runtime_event")), len(store2.rollbacks))) install_store.boom = None # ---------------- P5 广播通道异常不撤销事实 + 轮询兜底 store3 = Store() hub3 = install_store(store3) hub3.register_push(lambda sk, ev: (_ for _ in ()).throw(RuntimeError("ws down")), name="broken") r5 = W.apply_runtime_event(tenant_id="T3", session_id=31, event_type="hit", payload={"hp": 5}, state_updates=[{"entity_id": 8, "state_key": "hp", "state_value": {"hp": 5}}]) poll = W.poll_events("T3", 31, since_seq=0, limit=50, try_buffer=True) check("P5 广播失败但事件已提交", r5.get("ok") and len(store3.rows("pbl_runtime_event")) == 1) check("P5 轮询兜底取到事件", poll.get("count") == 1 and poll["events"][0]["seq_no"] == r5["seq_no"], "count=%s source=%s" % (poll.get("count"), poll.get("source"))) check("P5 兜底周期 3s", poll.get("poll_interval_ms") == 3000) # ---------------- P6 缺租户 fail-closed try: W.apply_runtime_event(tenant_id="", session_id=41, event_type="x") check("P6 缺租户必须拒", False, "未抛异常") except R.PblRtError as exc: check("P6 缺租户必须拒", exc.code == "PBL-TENANT-0001", "code=%s" % exc.code) # ---------------- P7 客户端写权威字段被拒 try: W.apply_runtime_event(tenant_id="T7", session_id=51, event_type="x", client_fields={"state_version": 99}) check("P7 客户端写 state_version 被拒", False, "未抛异常") except R.PblRtError as exc: check("P7 客户端写 state_version 被拒", exc.code == "PBL_RT_BAD_PAYLOAD", "code=%s" % exc.code) # ---------------- P8 乐观锁冲突 store8 = Store() install_store(store8) W.apply_runtime_event(tenant_id="T8", session_id=61, event_type="move", payload={"x": 1}, state_updates=[{"entity_id": 3, "state_key": "pos", "state_value": {"x": 1}}]) try: W.apply_runtime_event(tenant_id="T8", session_id=61, event_type="move", payload={"x": 2}, state_updates=[{"entity_id": 3, "state_key": "pos", "state_value": {"x": 2}, "if_state_version": 999}]) check("P8 乐观锁冲突必须拒", False, "未抛异常") except R.PblRtError as exc: check("P8 乐观锁冲突必须拒", exc.code == "PBL_RT_STATE_CONFLICT", "code=%s" % exc.code) check("P8 冲突后状态版本未推进", len(store8.rows("pbl_entity_state")) == 1 and int(store8.rows("pbl_entity_state")[0]["state_version"]) == 1) # ---------------- P9 时延与 SLA lat = r1.get("latency") or {} sla = lat.get("sla") or {} check("P9 时延字段齐备", all(k in lat for k in ("adjudicate_ms", "broadcast_ms", "e2e_ms")), "keys=%s" % sorted(lat.keys())) check("P9 SLA 预算 200/300/1000 + 3s 兜底", sla.get("adjudicate_ms") == 200 and sla.get("broadcast_ms") == 300 and sla.get("e2e_ms") == 1000 and sla.get("poll_fallback_interval_ms") == 3000, "sla=%s" % sla) check("P9 实测在预算内", r1.get("within_sla") is True, "adjudicate=%sms broadcast=%sms e2e=%sms" % (lat.get("adjudicate_ms"), lat.get("broadcast_ms"), lat.get("e2e_ms"))) # ---------------- P10 append-only 守卫 + 分区算法 ok_tx, msgs_tx = W.self_check() check("P10 tx_write 自检（append-only/分区/幂等键）", ok_tx, "; ".join([m for m in msgs_tx if "PASS" in m][:3])) from pbl_runtime_ext import partitions as P ddl = P.ddl_create_partition(__import__("datetime").date(2026, 10, 5), "mariadb") check("P10 月分区 DDL", "pbl_runtime_event_202610" in ddl and "TO_DAYS('2026-11-01')" in ddl, ddl[:70]) mig = P.partition_migration_ddl(2, "mariadb") check("P10 分区改造 DDL 含复合主键+MAXVALUE 兜底", any("ADD PRIMARY KEY (`id`, `created_at`)" in s for s in mig) and any("MAXVALUE" in s for s in mig)) try: P.ddl_drop_partition("pbl_runtime_event_202610") P.ddl_drop_partition("pbl_entity_state") check("P10 拒绝删除非事件月分区", False) except ValueError as exc: check("P10 拒绝删除非事件月分区", True, str(exc)[:40]) fails = [r for r in RESULTS if not r[0]] print("\n==== M11b SELFTEST: %d/%d PASS ====" % (len(RESULTS) - len(fails), len(RESULTS))) for _ok, name, detail in fails: print(" FAIL %s %s" % (name, detail)) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())