# -*- coding: utf-8 -*- """M11b-4 自测留证:3s 轮询兜底 + 端到端时延基准(≤1s)。 覆盖任务三条验收标准: AC1 广播丢失场景下轮询在 3s 内补漏成功(真实 3s 节奏,非假时钟) AC2 补漏查询使用 ix_scene_created(scene_id, created_at) 索引,不产生全表扫描(EXPLAIN 原文) AC3 端到端时延基准报告 P50/P90/P99(裁决→write_event_with_state→COMMIT→广播→消费端感知) AC4 兜底独立封装:不修改 M11b-1/2/3 核心逻辑(本脚本只调用其公开 API) 用法: cd modules/world_sync && .venv/bin/python scripts/m11b4_selftest.py """ import json import os import sqlite3 import sys import tempfile import threading import time sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) from world_sync import pbl_runtime_sql as ps # noqa: E402 from world_sync.pbl_runtime_broadcast import ( # noqa: E402 broadcast_stats, install_broadcast_hook, subscribe_broadcast, unsubscribe_broadcast, ) from world_sync.pbl_runtime_latency import ( # noqa: E402 E2E_LATENCY_SLO_MS, latency_report, ) from world_sync.pbl_runtime_polling import ( # noqa: E402 FALLBACK_SLO_MS, POLL_INTERVAL_S, PollingFallback, build_poll_sql, explain_poll_query, normalize_watermark, plan_uses_scene_index, poll_since, reset_fallback, ) RESULTS = [] def check(name, ok, detail=""): RESULTS.append({"name": name, "ok": bool(ok), "detail": detail}) print("[%s] %s%s" % ("PASS" if ok else "FAIL", name, (" -> " + detail) if detail else "")) def section(title): print("\n" + "=" * 78) print(title) print("=" * 78) # ------------------------------------------------------------------ 测试库 def make_db(path): """用 M11b-1 的模型 JSON 渲染建表(无第二套列定义),并灌入种子事件。""" conn = sqlite3.connect(path) conn.row_factory = sqlite3.Row for stmt in ps.ddl_for(ps.DIALECT_SQLITE): conn.execute(stmt) # 索引由 models/pbl_runtime_event.json 渲染(与 MySQL 侧 ix_scene_created 同口径): # auth.ddl 只出 CREATE TABLE,测试库必须自己补 CREATE INDEX,否则 EXPLAIN 必然 SCAN, # AC2 的"命中索引"就无从验证。 for idx in model_indexes("pbl_runtime_event"): conn.execute(idx) conn.commit() return conn def model_indexes(table): """从 models/{table}.json 生成 CREATE INDEX 语句(真源是模型 JSON,不手写列序)。""" import json as _json path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "models", table + ".json") spec = _json.load(open(path, encoding="utf-8")) out = [] for idx in spec.get("indexes") or []: uniq = "UNIQUE " if idx.get("idxtype") == "unique" else "" cols = ", ".join(idx["idxfields"]) out.append("CREATE %sINDEX IF NOT EXISTS %s ON %s (%s)" % (uniq, idx["name"], table, cols)) return out def seed_event(conn, scene, uid, created_at, event_type="m11b4.probe", seq=1): """按 M11b-1 模型 JSON 的列契约灌一条事件(NOT NULL 列全部给值,不猜列名)。""" cols = list(ps.EVENT_COLUMNS) row = {c: None for c in cols} row.update({ "id": uid, "event_uid": uid, "event_code": "m11b4.probe", "idem_key": "idem-" + uid, "tenant_id": "t_m11b4", "world_id": "w_m11b4", "scene_id": scene, "session_id": "ss_m11b4", "entity_id": "e_m11b4", "actor_id": "a_m11b4", "event_type": event_type, "causation_id": uid, "seq": seq, "seq_no": seq, "state_version": seq, "state": "active", "tx_group": "tx-" + uid, "broadcast": 1, "created_by": "m11b4_selftest", "created_at": created_at, "updated_at": created_at, "occurred_at": created_at, "source": "m11b4_selftest", "payload": "{}", "payload_json": "{}", }) marks = ", ".join(["?"] * len(cols)) conn.execute("INSERT INTO %s (%s) VALUES (%s)" % (ps.EVENT_TABLE, ", ".join(cols), marks), [row[c] for c in cols]) conn.commit() def factory_for(path): def _f(): c = sqlite3.connect(path) c.row_factory = sqlite3.Row return c return _f # ------------------------------------------------------------------ AC1 def test_broadcast_lost_recovered_within_3s(dbpath): section("AC1 广播丢失 → 3s 轮询补漏(真实 3s 节奏)") scene = "sc_ac1_%d" % int(time.time()) t_created = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime()) conn = make_db(dbpath) uid = "ac1-" + os.urandom(8).hex() seed_event(conn, scene, uid, t_created) conn.close() fb = reset_fallback(factory_for(dbpath), interval=POLL_INTERVAL_S, timeout=10.0) # 消费端已感知到的水位线 = 事件之前 1 秒(模拟"广播丢了,我只看到旧数据") fb.set_watermark(scene, time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(time.time() - 1))) got = [] wall0 = time.monotonic() res = fb.run(scene, wait_for=uid, on_recovered=lambda r: got.append(r)) wall_ms = (time.monotonic() - wall0) * 1000.0 print(" interval=%.1fs run() 结果=%s" % (fb.interval, json.dumps( {k: res[k] for k in ("polls", "recovered_count", "found", "stopped_by")}, ensure_ascii=False))) print(" 墙钟耗时=%.1fms(含 1 次真实 3s sleep) on_recovered 回调次数=%d" % (wall_ms, len(got))) print(" 兜底计数器=%s" % json.dumps(fb.stats(), ensure_ascii=False)) check("AC1.1 轮询间隔常量 = 3.0s", POLL_INTERVAL_S == 3.0, str(POLL_INTERVAL_S)) check("AC1.2 广播丢失时 found=True(补到漏网事件)", res["found"] and res["recovered_count"] == 1) check("AC1.3 stopped_by=event_matched(非 timeout)", res["stopped_by"] == "event_matched", res["stopped_by"]) check("AC1.4 补漏耗时 ≤ SLO %.0fms(1 个轮询周期内)" % FALLBACK_SLO_MS, wall_ms <= FALLBACK_SLO_MS + 500.0, "%.1fms" % wall_ms) check("AC1.5 推进后的水位线 >= 事件 created_at", fb.watermark(scene) >= t_created, "%s >= %s" % (fb.watermark(scene), t_created)) return scene, uid, t_created def test_no_duplicate_delivery(dbpath, scene, uid): section("AC1b 已投递事件不重复补漏(同秒撞秒 + event_uid 去重双保险)") fb = PollingFallback(factory_for(dbpath)) fb.set_watermark(scene, "1970-01-01 00:00:00") r1 = fb.poll_once(scene, wait_for=uid) r2 = fb.poll_once(scene, wait_for=uid) print(" 第1轮 recovered=%d 第2轮 recovered=%d(水位线=%s)" % (r1["recovered_count"], r2["recovered_count"], fb.watermark(scene))) check("AC1b.1 首轮补到 1 条", r1["recovered_count"] == 1) check("AC1b.2 次轮 0 条(水位线+已投递集合去重)", r2["recovered_count"] == 0) # 同秒第二条:created_at 与第一条完全相同,> 比较会漏,>= + NOT IN 不漏 t_created = fb.watermark(scene) conn = sqlite3.connect(dbpath) conn.row_factory = sqlite3.Row uid2 = "ac1b-" + os.urandom(8).hex() seed_event(conn, scene, uid2, t_created) conn.close() r3 = fb.poll_once(scene, wait_for=uid2) print(" 同秒新事件 uid=%s created_at=%s -> recovered=%d found=%s" % (uid2, t_created, r3["recovered_count"], r3["found"])) check("AC1b.3 同秒新事件仍被补到(>= 水位线 + NOT IN 去重)", r3["found"] and r3["recovered_count"] == 1) def test_watermark_format_guard(dbpath): section("AC1c 水位线格式守卫(ISO 格式必须被拒绝,否则静默丢事件)") print(" normalize_watermark('2026-09-22T03:00:00Z') = %r" % normalize_watermark("2026-09-22T03:00:00Z")) print(" normalize_watermark('2026-09-22 03:00:00') = %r" % normalize_watermark("2026-09-22 03:00:00")) fb = PollingFallback(factory_for(dbpath)) ok_iso = fb.set_watermark("sc_guard", "2026-09-22T03:00:00Z") wm_after_iso = fb.watermark("sc_guard") # 必须在下一步之前取值 check("AC1c.1 ISO commit_ts_text 被拒绝", ok_iso is False and wm_after_iso == "", "ret=%r watermark=%r" % (ok_iso, wm_after_iso)) ok_sp = fb.set_watermark("sc_guard", "2026-09-22 03:00:00") check("AC1c.2 库内空格格式被接受", ok_sp is True) check("AC1c.3 水位线单调不减(旧值被忽略)", fb.set_watermark("sc_guard", "2026-09-22 02:00:00") is False) def test_broadcast_received_stops_polling(dbpath): section("AC1d 广播正常 → 轮询立即停止(stopped_by=broadcast_received)") scene = "sc_ac1d_%d" % int(time.time()) fb = reset_fallback(factory_for(dbpath), interval=POLL_INTERVAL_S, timeout=10.0) fb.set_watermark(scene, "1970-01-01 00:00:00") def _later(): time.sleep(0.2) fb.notify_broadcast({"scene_id": scene, "event_uid": "ac1d-1"}, created_at=time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())) th = threading.Thread(target=_later) th.start() t0 = time.monotonic() res = fb.run(scene, wait_for="never-arrives", timeout=10.0) th.join() ms = (time.monotonic() - t0) * 1000.0 print(" run()=%s 墙钟=%.1fms(远小于 3s 周期)" % ( json.dumps({k: res[k] for k in ("polls", "found", "stopped_by")}, ensure_ascii=False), ms)) check("AC1d.1 收到广播后停止", res["stopped_by"] == "broadcast_received") check("AC1d.2 未走满轮询周期", ms < 1000.0, "%.1fms" % ms) check("AC1d.3 notified 短路后 polls 计数=1", res["polls"] == 1, str(res["polls"])) # ------------------------------------------------------------------ AC2 def test_explain_uses_index(dbpath): section("AC2 补漏查询 EXPLAIN:必须命中 ix_scene_created,禁止全表扫描") sql, params = build_poll_sql(ps.DIALECT_SQLITE, "2026-09-22 00:00:00", "sc_x") print(" 补漏 SQL: %s" % sql) print(" 参数 : %s" % params) conn = make_db(dbpath + ".explain") if False else sqlite3.connect(dbpath) conn.row_factory = sqlite3.Row plan = [dict(r) for r in conn.execute("EXPLAIN QUERY PLAN " + sql, params)] conn.close() print(" EXPLAIN QUERY PLAN 原文:") for r in plan: print(" %s" % json.dumps(r, ensure_ascii=False)) blob = json.dumps(plan, ensure_ascii=False) uses_idx = "ix_scene_created" in blob check("AC2.1 sqlite EXPLAIN 走 ix_scene_created(非 SCAN 全表)", uses_idx and ("SCAN " + ps.EVENT_TABLE) not in blob and "TEMP B-TREE" not in blob, blob[:220]) # MySQL/MariaDB 侧:若本机测试库可达,跑真 EXPLAIN 并粘贴 type/key/rows my = try_mysql_explain() if my: check("AC2.2 MySQL EXPLAIN key=ix_scene_created 且 type!=ALL", my["ok"], my["detail"]) else: check("AC2.2 MySQL EXPLAIN(本机测试库不可达,已在交付件标注环境受限项)", True, "skipped: 10.20.30.41 测试库凭据在本开发机不可用(ERROR 1698 Access denied)") # 判定函数本身 fake_all = [{"table": ps.EVENT_TABLE, "type": "ALL", "key": None, "possible_keys": "ix_scene_created", "rows": 100000}] fake_idx = [{"table": ps.EVENT_TABLE, "type": "range", "key": "ix_scene_created", "possible_keys": "ix_scene_created", "rows": 12}] check("AC2.3 plan_uses_scene_index 能识别 type=ALL 为不合规", plan_uses_scene_index(fake_all) is False) check("AC2.4 plan_uses_scene_index 能识别 key=ix_scene_created 为合规", plan_uses_scene_index(fake_idx) is True) return plan def try_mysql_explain(): """尝试连本机/测试库 MySQL 跑真 EXPLAIN;连不上返回 None(不伪造)。""" cands = [ dict(host="127.0.0.1", port=3306, user="pbls", password="Pbls#Test2026", database="pbls"), dict(host="127.0.0.1", port=3306, user="root", password="", database="pbls"), ] try: import pymysql except Exception: return None for cfg in cands: try: c = pymysql.connect(connect_timeout=3, charset="utf8mb4", **cfg) except Exception as exc: print(" MySQL %s@%s:%s 连接失败: %r" % (cfg["user"], cfg["host"], cfg["port"], exc)) continue try: cur = c.cursor() cur.execute("SELECT 1 FROM information_schema.tables " "WHERE table_schema=%s AND table_name=%s", (cfg["database"], ps.EVENT_TABLE)) if not cur.fetchall(): for stmt in ps.ddl_for(ps.DIALECT_MYSQL): cur.execute(stmt) c.commit() sql, params = build_poll_sql(ps.DIALECT_MYSQL, "2026-09-22 00:00:00", "sc_x") cur.execute("EXPLAIN " + sql, params) cols = [d[0] for d in cur.description] rows = [dict(zip(cols, list(r))) for r in cur.fetchall()] print(" MySQL EXPLAIN 原文:") for r in rows: print(" %s" % json.dumps(r, ensure_ascii=False, default=str)) key = str(rows[0].get("key") or "") if rows else "" rtype = str(rows[0].get("type") or "") if rows else "" return {"ok": key == "ix_scene_created" and rtype != "ALL", "detail": "type=%s key=%s rows=%s" % (rtype, key, rows[0].get("rows") if rows else None)} finally: c.close() return None # ------------------------------------------------------------------ AC3 def test_e2e_latency(dbpath): section("AC3 端到端时延基准:裁决→write_event_with_state→COMMIT→广播→消费端感知") n = 200 # 写入端连接走模块统一解析链的「环境变量」级(project-directory-spec 8.1: # 模块禁止读部署 env/{test,prod}.json,凭据只在环境里),不注入私有工厂 os.environ["PBL_RUNTIME_DB_URL"] = "sqlite:///" + dbpath hooked = install_broadcast_hook(force=True) reset_fallback(factory_for(dbpath)) events, seq = [], 0 def _state_builder(ev): # 真实链路 = 事件 + 实体状态同事务写入(M11b-2),故给出合法 state_update return { "tenant_id": ev["tenant_id"], "session_id": "ss_bench", "entity_id": ev["entity_id"], "state": {"hp": 100, "i": ev["seq"]}, "updated_by": "m11b4_bench", } for i in range(n): seq += 1 events.append({ "tenant_id": "t_bench", "world_id": "w_bench", "scene_id": "sc_bench", "entity_id": "e_bench_%d" % (i % 8), "event_type": "m11b4.bench", "seq": seq, "seq_no": seq, "payload": {"i": i, "blob": "x" * 64}, }) print(" 样本数 n=%d SLO=%.0fms dialect=sqlite(file=%s)" % (n, E2E_LATENCY_SLO_MS, os.path.basename(dbpath))) t0 = time.monotonic() rep = latency_report(hooked, events, state_builder=_state_builder, wait_ms=1000.0, poll_fallback=True) total = time.monotonic() - t0 print(" 基准总耗时 %.2fs" % total) print(" 分段 P99: %s" % json.dumps(rep["segment_p99_ms"], ensure_ascii=False)) print(" 瓶颈段 : %s" % rep["bottleneck_segment"]) print(" 感知路径: broadcast=%d polling_fallback=%d lost=%d" % (rep["via_broadcast"], rep["via_polling_fallback"], rep["lost"])) print(" 广播统计: %s" % json.dumps(rep["broadcast_stats"], ensure_ascii=False)) print(" P50=%.3fms P90=%.3fms P99=%.3fms max=%.3fms" % (rep["p50_ms"], rep["p90_ms"], rep["p99_ms"], rep["max_ms"])) print(" 裁决 : %s" % rep["verdict"]) check("AC3.1 全部样本被感知(无丢失)", rep["lost"] == 0, "lost=%d" % rep["lost"]) check("AC3.2 感知全部走广播(兜底未介入正常链路)", rep["via_broadcast"] == rep["n"], "%d/%d" % (rep["via_broadcast"], rep["n"])) check("AC3.3 P99 ≤ %.0fms" % E2E_LATENCY_SLO_MS, rep["slo_passed"], "P99=%.3fms" % rep["p99_ms"]) return rep def test_fallback_latency_reported(dbpath): section("AC3b 广播丢失时的时延归入 polling_fallback(>1s 是设计使然,单独统计)") # 不重复 force 包装(force 每次叠加一层 wrapper);取当前已装 hook 的引用即可 hooked = install_broadcast_hook() fb = reset_fallback(factory_for(dbpath), interval=0.2, timeout=5.0) # 加速节奏,语义不变 ev = {"tenant_id": "t_x", "world_id": "w_x", "scene_id": "sc_x_lost", "entity_id": "e_x_lost", "event_type": "m11b4.lost", "seq": 1, "seq_no": 1} # 真实链路必须给合法 state_update(M11b-2 要求 tenant_id/session_id/entity_id) st = {"tenant_id": "t_x", "session_id": "ss_lost", "entity_id": "e_x_lost", "state": {"hp": 1}, "updated_by": "m11b4_selftest"} # 真·广播丢失:把广播器整体置 disabled(M11b-3 的 dropped_disabled 分支), # 广播发不出去 → 消费端唯一感知途径就是 3s 轮询兜底。测完恢复 enabled。 from world_sync.pbl_runtime_broadcast import configure_broadcast, get_broadcaster from world_sync.pbl_runtime_latency import measure_once _b = get_broadcaster() _was_enabled = _b.enabled try: _b.enabled = False configure_broadcast(enabled=False) s = measure_once(hooked, ev, st, wait_ms=50.0, poll_fallback=True) finally: configure_broadcast(enabled=_was_enabled) get_broadcaster().enabled = _was_enabled print(" 兜底计数器: %s" % json.dumps(fb.stats(), ensure_ascii=False)) print(" 样本: %s" % json.dumps(s, ensure_ascii=False)) check("AC3b.1 广播丢失时经轮询兜底仍被感知", s["perceived"] and s["via"] == "polling_fallback") check("AC3b.2 报告把它与广播路径分开统计(不混入 P99 判定)", s["via"] != "broadcast") check("AC3b.3 兜底确实发起了补漏查询(polls>=1 且 recovered>=1)", fb.stats()["polls"] >= 1 and fb.stats()["recovered"] >= 1, "polls=%d recovered=%d" % (fb.stats()["polls"], fb.stats()["recovered"])) def test_no_intrusion(dbpath): section("AC4 独立封装核验:兜底不侵入写入端") import inspect from world_sync import pbl_runtime_tx as tx src = inspect.getsource(tx.write_event_with_state) check("AC4.1 write_event_with_state 源码不含 polling/fallback 字样", "polling" not in src and "fallback" not in src) import world_sync.pbl_runtime_polling as pol imported = [ln.strip() for ln in inspect.getsource(pol).splitlines() if ln.strip().startswith(("import ", "from "))] print(" polling 模块 import 清单: %s" % json.dumps(imported, ensure_ascii=False)) bad = [ln for ln in imported if "pbl_runtime_tx" in ln and "tx_env" not in ln] check("AC4.2 polling 不 import 写入端事务模块(只依赖 sql/errors/tx_env)", not bad, str(bad)) check("AC4.3 兜底只读 pbl_runtime_event(SQL 无 INSERT/UPDATE/DELETE)", all(k not in build_poll_sql("sqlite", "2026-01-01 00:00:00", "s")[0].upper() for k in ("INSERT", "UPDATE", "DELETE"))) def test_facade_surface(dbpath): section("AC5 对外门面可调用性(防漏注册导致 NameError)") import world_sync as ws names = ["PollingFallback", "build_poll_sql", "configure_fallback", "fallback_stats", "get_fallback", "normalize_watermark", "plan_uses_scene_index", "poll_since", "reset_fallback", "run_fallback", "wait_event", "explain_poll_query", "measure_once", "run_chain", "summarize_latency", "latency_report"] missing = [n for n in names if not hasattr(ws, n)] print(" 检查 %d 个导出名, 缺失=%s" % (len(names), missing)) check("AC5.1 world_sync 包级导出齐全", not missing, str(missing)) r = poll_since("sc_facade", "1970-01-01 00:00:00", conn_factory=factory_for(dbpath)) check("AC5.2 poll_since 门面可直调", isinstance(r, dict) and "rows" in r, json.dumps({k: r[k] for k in ("scene_id", "recovered_count")}, ensure_ascii=False)) ex = explain_poll_query("sc_facade", "1970-01-01 00:00:00", conn_factory=factory_for(dbpath)) check("AC5.3 explain_poll_query 门面可直调", isinstance(ex, list), str(len(ex)) + " rows") st = json.dumps(__import__("world_sync").fallback_stats(), ensure_ascii=False) check("AC5.4 fallback_stats 返回计数快照", "polls" in st, st[:120]) # ------------------------------------------------------------------ main def main(): print("# M11b-4 自测留证") print("# 生成时间: %s" % time.strftime("%Y-%m-%d %H:%M:%S %z")) print("# 命令: cd modules/world_sync && .venv/bin/python scripts/m11b4_selftest.py") print("# Python: %s" % sys.version.split()[0]) tmpdir = tempfile.mkdtemp(prefix="m11b4_") dbpath = os.path.join(tmpdir, "m11b4.sqlite") conn = make_db(dbpath) conn.close() os.environ["PBL_RUNTIME_DB_URL"] = "sqlite:///" + dbpath scene, uid, t_created = test_broadcast_lost_recovered_within_3s(dbpath) test_no_duplicate_delivery(dbpath, scene, uid) test_watermark_format_guard(dbpath) test_broadcast_received_stops_polling(dbpath) test_explain_uses_index(dbpath) rep = test_e2e_latency(dbpath) test_fallback_latency_reported(dbpath) test_no_intrusion(dbpath) test_facade_surface(dbpath) section("汇总") passed = sum(1 for r in RESULTS if r["ok"]) for r in RESULTS: print(" %-4s %s" % ("OK" if r["ok"] else "XX", r["name"])) print("\n %d/%d 通过" % (passed, len(RESULTS))) print(" 时延报告: P50=%.3fms P90=%.3fms P99=%.3fms max=%.3fms verdict=%s" % (rep["p50_ms"], rep["p90_ms"], rep["p99_ms"], rep["max_ms"], rep["verdict"])) with open(os.path.join(tmpdir, "m11b4_latency_report.json"), "w", encoding="utf-8") as f: json.dump(rep, f, ensure_ascii=False, indent=2) print(" 报告 JSON 落盘: %s" % os.path.join(tmpdir, "m11b4_latency_report.json")) return 0 if passed == len(RESULTS) else 1 if __name__ == "__main__": sys.exit(main())