556 lines
25 KiB
Python
556 lines
25 KiB
Python
#!/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())
|