730 lines
34 KiB
Python
730 lines
34 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""tx_write —— M11b 核心：**单事务**写事件 + 实体状态 + 快照，**提交后**广播。
|
||
|
||
契约（三处同步：本文件实现 → __init__.py 导出 → init.py env 注册）：
|
||
apply_runtime_event(...) 单事务：幂等检查 → INSERT pbl_runtime_event
|
||
→ UPSERT pbl_entity_state（乐观锁 state_version）
|
||
→ INSERT pbl_world_state_snapshot → COMMIT → 广播
|
||
poll_events(...) 3s 轮询兜底：先取进程内 hub 增量，未命中/有缺口退回查库
|
||
read_states(...) 读实体状态（tenant_id 强制打头）
|
||
assert_append_only(...) pbl_runtime_event 禁 UPDATE/DELETE + 基表写保护
|
||
|
||
**列名权威源 = models/*.json**（module-development-spec 铁律：禁止猜列）。
|
||
本文件所有 INSERT/UPDATE/SELECT 只使用下列真实列，并由 build_row()/require_columns()
|
||
双重校验（未知列直接抛错，不再静默丢列——这正是上一版幂等/乐观锁失效的根因）：
|
||
|
||
pbl_runtime_event : tenant_id, event_uid, session_id, world_id, actor_id,
|
||
event_type, payload_json, causation_id, state_version, seq,
|
||
occurred_at, created_at, updated_at
|
||
（uk_re_uid = tenant_id+event_uid+occurred_at 唯一；
|
||
idx_re_session_seq = tenant_id+session_id+seq）
|
||
pbl_entity_state : tenant_id, session_id, entity_id, state_json, state_version,
|
||
checksum, updated_by, created_at, updated_at
|
||
（uk_es = tenant_id+session_id+entity_id 唯一）
|
||
pbl_world_state_snapshot: tenant_id, session_id, snapshot_uid, state_fingerprint,
|
||
entities_json, events_through_seq, reason, created_at, updated_at
|
||
|
||
关键保证（QC #2/#3/#7 关注点）：
|
||
* 原子性：三条 DML 全在同一个 ``rtx_db.transaction()`` 内；任一失败 → rollback，
|
||
事件行**不存在**（真回滚，不是「标 rolled_back」），且不广播。
|
||
* 无幽灵更新：广播只在事务 commit 成功之后（with 块退出后）发出。
|
||
* 幂等：event_uid 即幂等键（客户端 client_event_id 原样透传，否则服务端派生），
|
||
同 uid 命中已有事件 → dedup=True，不重复写、不重复广播；DB 唯一索引兜底并发。
|
||
* 乐观锁：state_version 服务端单调递增，客户端禁写；base_version 不匹配 → 冲突回滚。
|
||
* fail-closed：库名/事务原语/租户/必填列任一不可得 → 抛错，绝不退化为逐条提交。
|
||
"""
|
||
|
||
import hashlib
|
||
import inspect
|
||
import json
|
||
import os
|
||
import time
|
||
|
||
from . import broadcast as _hub
|
||
from . import latency as _lat
|
||
from . import rtx_db
|
||
from .m11b_config import (ERR_APPEND_ONLY, ERR_BAD_PARAM, ERR_BROADCAST_AFTER_COMMIT,
|
||
ERR_STATE_CONFLICT, ERR_TENANT_MISSING, EVENT_TABLE,
|
||
PROTECTED_BASE_TABLES, SNAPSHOT_TABLE, STATE_TABLE)
|
||
|
||
__all__ = ["apply_runtime_event", "poll_events", "read_states", "assert_append_only",
|
||
"idem_key_of", "table_columns", "build_row", "self_check"]
|
||
|
||
_BASE_TABLES_LC = tuple(t.lower() for t in PROTECTED_BASE_TABLES)
|
||
_MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
# 对外契约使用的逻辑字段名 → 表真实列名（单一映射，避免再次漂移）
|
||
EVENT_COL_SEQ = "seq" # 会话内单调序号
|
||
EVENT_COL_UID = "event_uid" # 幂等键（唯一索引成员）
|
||
EVENT_COL_PAYLOAD = "payload_json"
|
||
EVENT_COL_STATE_VERSION = "state_version"
|
||
STATE_COL_KEY = "entity_id" # 状态行主键维度（tenant+session+entity_id）
|
||
STATE_COL_VALUE = "state_json"
|
||
|
||
|
||
# ---------------------------------------------------------------- 列名自省
|
||
_COL_CACHE = {}
|
||
|
||
|
||
def _model_spec(table):
|
||
"""读取 models/{table}.json（表结构权威源）。读不到返回 None（不猜）。"""
|
||
if table in _COL_CACHE:
|
||
return _COL_CACHE[table]
|
||
path = os.path.join(_MODULE_ROOT, "models", "%s.json" % table)
|
||
spec = None
|
||
try:
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
spec = json.load(f)
|
||
except Exception: # noqa: BLE001
|
||
spec = None
|
||
_COL_CACHE[table] = spec
|
||
return spec
|
||
|
||
|
||
def table_columns(table):
|
||
"""表真实列名 list；模型缺失返回 None。"""
|
||
spec = _model_spec(table)
|
||
if not spec:
|
||
return None
|
||
names = []
|
||
for fd in spec.get("fields") or []:
|
||
if isinstance(fd, dict) and fd.get("name"):
|
||
names.append(fd["name"])
|
||
elif isinstance(fd, str):
|
||
names.append(fd.split()[0])
|
||
return names or None
|
||
|
||
|
||
def _is_not_null(fd):
|
||
"""字段是否声明 NOT NULL —— 兼容 models/*.json 并存的两种写法。
|
||
|
||
* ``{"null": false}`` ：xlsx→json 产物写法（pbl_entity_state / pbl_world_state_snapshot）
|
||
* ``{"nullable": "no"}`` ：database-table-definition-spec 规范写法（pbl_runtime_event）
|
||
|
||
M11b-2c-A3 修复（PBL-RTX-0004 **漏报侧**根因）：此前只认 ``null: false``，
|
||
而 pbl_runtime_event 用的是规范写法 ``nullable: "no"``，使该表 ``required_columns()``
|
||
恒为 ``[]`` —— build_row 的 INSERT 完整性门禁对事件表**完全失效**
|
||
（该报的缺列不报，静默写出半空行）。
|
||
"""
|
||
if fd.get("null") is False:
|
||
return True
|
||
if str(fd.get("nullable") or "").strip().lower() in ("no", "false", "0"):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _has_db_default(fd):
|
||
"""列是否在 DDL 中声明 DEFAULT（含 CURRENT_TIMESTAMP）。
|
||
|
||
``NOT NULL DEFAULT x`` 的列在 INSERT 省略时由数据库补值，**不构成"缺 NOT NULL 列"**。
|
||
M11b-2c-A3 修复（PBL-RTX-0004 **误报侧**根因之二）：pbl_runtime_event 的
|
||
event_code/idem_key/scene_id/entity_id/actor_id/causation_id/seq/source/state/
|
||
tx_group/broadcast/created_by/created_at/updated_at 等列均为 NOT NULL DEFAULT，
|
||
写路径按语义不填它们；若计入必填集合，合法调用会被误报缺列。
|
||
（与 _recognize nullable 写法配套：只修识别不修默认值排除，事件表主链路会被误挡。）
|
||
"""
|
||
if "default" not in fd:
|
||
return False
|
||
dv = fd.get("default")
|
||
if dv is None:
|
||
return False
|
||
if str(dv).strip().upper() in ("NULL", "NONE"):
|
||
return False
|
||
return True # 含 DEFAULT ''（空串默认值同样是 DDL 默认值，数据库会补值）
|
||
|
||
|
||
def required_columns(table):
|
||
"""INSERT 必须**显式给值**的列：NOT NULL + 非自增 + 无数据库 DEFAULT。
|
||
|
||
自增列由数据库生成、NOT NULL DEFAULT 列由默认值补上，两者都不需要调用方提供；
|
||
纳入必填集合只会造成 PBL-RTX-0004 误报。真正的必填列（pbl_entity_state 的
|
||
tenant_id/session_id/entity_id/state_version/checksum/updated_by/created_at/
|
||
updated_at，pbl_runtime_event 的 event_uid/event_type）缺值时数据库严格模式会
|
||
直接报错，故必须在这里 fail-closed 拦住，不得放宽。
|
||
"""
|
||
spec = _model_spec(table)
|
||
if not spec:
|
||
return []
|
||
out = []
|
||
for fd in spec.get("fields") or []:
|
||
if not isinstance(fd, dict) or not fd.get("name"):
|
||
continue
|
||
if _is_not_null(fd) and not fd.get("auto_increment") and not _has_db_default(fd):
|
||
out.append(fd["name"])
|
||
return out
|
||
|
||
|
||
def build_row(table, wanted, strict=True, full_row=True):
|
||
"""按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。
|
||
|
||
full_row=True（默认，**INSERT 语义**）：额外校验 required_columns() 齐备，
|
||
缺列抛 PBL-RTX-0004。
|
||
full_row=False（**UPDATE SET 子句语义**）：被改动的行在库中已存在，未出现在 SET
|
||
列表里的 NOT NULL 列保持原值，**不存在"缺列"**，因此不得套用 INSERT 的完整性
|
||
校验；同时禁止 SET 不可变列（tenant_id/id/created_at），违者抛 PBL-RTX-0006。
|
||
|
||
PBL-RTX-0004 闭环（M11b-2c-A3，A1 判定为业务实现缺陷）：误报有两条独立路径——
|
||
(1) UPDATE 分支沿用 INSERT 校验（此前用 strict=False 绕，但 strict 只管未知列，
|
||
缺列校验照跑 → tenant_id/session_id/entity_id/created_at 必被误报）。
|
||
现按 full_row=False 构造：定位维度走 WHERE、created_at 保持建档原值。
|
||
(2) 必填集合把「自增列 / NOT NULL DEFAULT 列」也算进来 → required_columns()
|
||
已排除（见 _has_db_default）。
|
||
漏报侧同修：required_columns() 此前不认 ``nullable: "no"`` 写法，事件表门禁恒真。
|
||
修后语义：该报的（真缺 event_uid/event_type 等无默认 NOT NULL 列）一定报，
|
||
不该报的（UPDATE SET 子集、数据库能补值的列）一定不报。
|
||
"""
|
||
cols = table_columns(table)
|
||
if not cols:
|
||
if strict:
|
||
raise rtx_db.RtxError("PBL-RTX-0003",
|
||
"models/%s.json 不可读，拒绝按猜列写入" % table)
|
||
return dict(wanted)
|
||
unknown = [k for k in wanted if k not in cols]
|
||
if unknown and strict:
|
||
raise rtx_db.RtxError(
|
||
"PBL-RTX-0003",
|
||
"%s 未知列 %s（权威列=%s）" % (table, unknown, cols))
|
||
row = dict((k, v) for k, v in wanted.items() if k in cols)
|
||
if full_row:
|
||
missing = [c for c in required_columns(table) if c not in row]
|
||
if missing:
|
||
raise rtx_db.RtxError("PBL-RTX-0004",
|
||
"%s 缺少 NOT NULL 列 %s" % (table, missing))
|
||
else:
|
||
immutable = [c for c in ("tenant_id", "id", "created_at") if c in row]
|
||
if immutable:
|
||
raise rtx_db.RtxError("PBL-RTX-0006",
|
||
"%s UPDATE 禁止 SET 不可变列 %s" % (table, immutable))
|
||
return row
|
||
|
||
|
||
def _dumps(value):
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, str):
|
||
return value
|
||
try:
|
||
return json.dumps(value, ensure_ascii=False, sort_keys=True)
|
||
except Exception: # noqa: BLE001
|
||
return json.dumps(str(value), ensure_ascii=False)
|
||
|
||
|
||
def _loads(value, default=None):
|
||
if value is None or value == "":
|
||
return default
|
||
if isinstance(value, (dict, list)):
|
||
return value
|
||
try:
|
||
return json.loads(value)
|
||
except Exception: # noqa: BLE001
|
||
return default
|
||
|
||
|
||
def _digest(obj):
|
||
return hashlib.sha256(_dumps(obj).encode("utf-8")).hexdigest()[:40]
|
||
|
||
|
||
def _norm(rows):
|
||
return rtx_db._norm_rows(rows)
|
||
|
||
|
||
# ---------------------------------------------------------------- 幂等键
|
||
def idem_key_of(tenant_id, session_id, event_type, payload=None, client_key=None):
|
||
"""幂等键（落 event_uid 列，长度 ≤32 且被唯一索引覆盖）。
|
||
|
||
客户端给的 client_event_id 原样透传（截断 32，保证重放同键）；
|
||
否则由 tenant|session|type|payload 派生 sha256[:32]，同参必同键。
|
||
"""
|
||
if client_key:
|
||
return str(client_key)[:32]
|
||
basis = "|".join([str(tenant_id), str(session_id), str(event_type),
|
||
_dumps(payload or {})])
|
||
return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:32]
|
||
|
||
|
||
# ---------------------------------------------------------------- 校验
|
||
def _require_tenant(tenant_id):
|
||
if tenant_id in (None, "", 0):
|
||
try:
|
||
from pbl_common.api import tenant_id as _tid # type: ignore
|
||
|
||
tenant_id = _tid()
|
||
except Exception: # noqa: BLE001
|
||
tenant_id = None
|
||
if tenant_id in (None, "", 0):
|
||
raise rtx_db.RtxError(ERR_TENANT_MISSING, "tenant_id 缺失，fail-closed 拒绝读写")
|
||
return tenant_id
|
||
|
||
|
||
def assert_append_only(operation, table, tenant_id=None):
|
||
"""守卫：pbl_runtime_event 只允许 INSERT/SELECT；基表禁止任何写。违规抛 RtxError。"""
|
||
op = str(operation or "").strip().upper()
|
||
tbl = str(table or "").strip().lower()
|
||
if tbl in _BASE_TABLES_LC and op in ("INSERT", "UPDATE", "DELETE", "REPLACE",
|
||
"TRUNCATE"):
|
||
raise rtx_db.RtxError(ERR_APPEND_ONLY, "基表 %s 只读（本模块禁改基表）" % tbl)
|
||
if tbl == EVENT_TABLE.lower() and op in ("UPDATE", "DELETE", "REPLACE", "TRUNCATE",
|
||
"ALTER", "DROP"):
|
||
raise rtx_db.RtxError(ERR_APPEND_ONLY,
|
||
"pbl_runtime_event 为 append-only，禁止 %s" % op)
|
||
return True
|
||
|
||
|
||
# ---------------------------------------------------------------- 事务内步骤
|
||
async def _find_idem(tx, tenant_id, uid):
|
||
sql = ("SELECT id, %s, event_type, %s FROM %s "
|
||
"WHERE tenant_id=%%s AND %s=%%s LIMIT 1"
|
||
% (EVENT_COL_SEQ, EVENT_COL_STATE_VERSION, EVENT_TABLE, EVENT_COL_UID))
|
||
rows = _norm(await tx.execute(sql, (tenant_id, uid)))
|
||
return rows[0] if rows else None
|
||
|
||
|
||
async def _next_seq(tx, tenant_id, session_id):
|
||
"""会话内单调序号：MAX(seq)+1（与 uk/idx_re_session_seq 同维度）。"""
|
||
sql = ("SELECT COALESCE(MAX(%s), 0) AS max_seq FROM %s "
|
||
"WHERE tenant_id=%%s AND session_id=%%s" % (EVENT_COL_SEQ, EVENT_TABLE))
|
||
rows = _norm(await tx.execute(sql, (tenant_id, session_id)))
|
||
try:
|
||
return int(rows[0].get("max_seq") or 0) + 1 if rows else 1
|
||
except (TypeError, ValueError):
|
||
return 1
|
||
|
||
|
||
async def _apply_state(tx, tenant_id, session_id, upd, seq_no, base_version,
|
||
actor_id, now):
|
||
"""单条实体状态 UPSERT + 乐观锁。返回 {entity_id, state_version, action}。"""
|
||
if not isinstance(upd, dict):
|
||
raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 每项必须是对象")
|
||
entity_id = upd.get("entity_id") or upd.get("state_key")
|
||
if not entity_id:
|
||
raise rtx_db.RtxError(ERR_BAD_PARAM,
|
||
"state_updates 每项必须含 entity_id（客户端禁写 state_version）")
|
||
entity_id = str(entity_id)[:64]
|
||
value = upd.get("state") if "state" in upd else upd.get("state_value")
|
||
if value is None:
|
||
raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 每项必须含 state")
|
||
|
||
rows = _norm(await tx.execute(
|
||
"SELECT id, %s FROM %s WHERE tenant_id=%%s AND session_id=%%s AND %s=%%s LIMIT 1"
|
||
% (EVENT_COL_STATE_VERSION, STATE_TABLE, STATE_COL_KEY),
|
||
(tenant_id, session_id, entity_id)))
|
||
state_version = _dumps(value)
|
||
checksum = _digest(value)
|
||
if rows:
|
||
cur_ver = int(rows[0].get(EVENT_COL_STATE_VERSION) or 0)
|
||
if base_version is not None and int(base_version) != cur_ver:
|
||
raise rtx_db.RtxError(
|
||
ERR_STATE_CONFLICT,
|
||
"entity_id=%s 乐观锁冲突：base_version=%s 当前 state_version=%s（整体回滚）"
|
||
% (entity_id, base_version, cur_ver))
|
||
new_ver = cur_ver + 1 # 服务端单调递增
|
||
vals = build_row(STATE_TABLE, {
|
||
STATE_COL_VALUE: state_version, EVENT_COL_STATE_VERSION: new_ver,
|
||
"checksum": checksum, "updated_by": actor_id, "updated_at": now,
|
||
}, strict=True, full_row=False)
|
||
await tx.execute(
|
||
"UPDATE %s SET %s WHERE tenant_id=%%s AND session_id=%%s AND %s=%%s"
|
||
% (STATE_TABLE, ", ".join("%s=%%s" % c for c in vals.keys()), STATE_COL_KEY),
|
||
tuple(list(vals.values()) + [tenant_id, session_id, entity_id]))
|
||
action = "update"
|
||
else:
|
||
new_ver = 1
|
||
await tx.insert(STATE_TABLE, build_row(STATE_TABLE, {
|
||
"tenant_id": tenant_id, "session_id": session_id,
|
||
STATE_COL_KEY: entity_id, STATE_COL_VALUE: state_version,
|
||
EVENT_COL_STATE_VERSION: new_ver, "checksum": checksum,
|
||
"updated_by": actor_id, "created_at": now, "updated_at": now,
|
||
}))
|
||
action = "insert"
|
||
return {"entity_id": entity_id, "state_version": new_ver, "action": action,
|
||
"state_key": entity_id, "state_value": value, "seq": seq_no}
|
||
|
||
|
||
# ---------------------------------------------------------------- 主写入
|
||
async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None,
|
||
event_type=None, payload=None, state_updates=None,
|
||
client_event_id=None, idem_key=None, snapshot=None,
|
||
base_version=None, actor_id=None, causation_id=None,
|
||
occurred_at=None, env=None):
|
||
"""单事务写「事件 + 实体状态 + 快照」，提交后广播。
|
||
|
||
返回 {ok, event_id, seq, cursor, dedup, state_rows, latency_ms, channel, stmt_count}
|
||
抛 RtxError：PBL-TENANT-0001 / PBL-PARAM-0001 / PBL-STATE-CONFLICT / PBL-RTX-000x
|
||
"""
|
||
tenant_id = _require_tenant(tenant_id)
|
||
if world_id in (None, "") or session_id in (None, "") or not event_type:
|
||
raise rtx_db.RtxError(ERR_BAD_PARAM, "world_id/session_id/event_type 必填")
|
||
state_updates = state_updates or []
|
||
if not isinstance(state_updates, (list, tuple)):
|
||
raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 必须是数组")
|
||
|
||
uid = idem_key_of(tenant_id, session_id, event_type, payload,
|
||
client_event_id or idem_key)
|
||
hub = _hub.get_hub()
|
||
channel = _hub.channel_of(world_id, tenant_id)
|
||
now = occurred_at or time.strftime("%Y-%m-%d %H:%M:%S")
|
||
actor = str(actor_id or "server")[:32]
|
||
causation = str(causation_id or "")[:32]
|
||
|
||
event_id = None
|
||
seq_no = None
|
||
state_events = []
|
||
dedup_result = None
|
||
t0 = time.perf_counter() # 裁决计时起点（含事务全程）
|
||
async with rtx_db.transaction() as tx:
|
||
try:
|
||
# 1) 幂等：同 event_uid 已存在 → 不重复写、不重复广播
|
||
dup = await _find_idem(tx, tenant_id, uid)
|
||
if dup:
|
||
await tx.commit()
|
||
dedup_result = {
|
||
"ok": True, "dedup": True, "event_id": dup.get("id"),
|
||
"event_uid": uid, "seq": dup.get(EVENT_COL_SEQ),
|
||
"cursor": None, "state_rows": 0, "latency_ms": None,
|
||
"channel": channel, "stmt_count": tx.stmt_count,
|
||
}
|
||
else:
|
||
# 2) 会话内单调 seq
|
||
seq_no = await _next_seq(tx, tenant_id, session_id)
|
||
|
||
# 3) INSERT 事件（append-only：本模块对该表只有 INSERT）
|
||
assert_append_only("INSERT", EVENT_TABLE, tenant_id)
|
||
max_ver = 0
|
||
event_id = await tx.insert(EVENT_TABLE, build_row(EVENT_TABLE, {
|
||
"tenant_id": tenant_id, EVENT_COL_UID: uid,
|
||
"session_id": session_id, "world_id": world_id,
|
||
"actor_id": actor, "event_type": str(event_type)[:64],
|
||
EVENT_COL_PAYLOAD: _dumps(payload or {}),
|
||
"causation_id": causation,
|
||
EVENT_COL_STATE_VERSION: max_ver,
|
||
EVENT_COL_SEQ: seq_no,
|
||
"occurred_at": now, "created_at": now, "updated_at": now,
|
||
}))
|
||
|
||
# 4) UPSERT 实体状态（乐观锁：state_version 服务端递增，客户端禁写）
|
||
for upd in state_updates:
|
||
st = await _apply_state(tx, tenant_id, session_id, upd, seq_no,
|
||
base_version, actor, now)
|
||
state_events.append(st)
|
||
# 4b) 事件行 state_version：pbl_runtime_event 为 append-only，
|
||
# 同事务内 INSERT 之后**不再 UPDATE 回填**（对 append-only 表
|
||
# 的任何 UPDATE 都被 assert_append_only 拦截）。本轮 UPSERT 后的
|
||
# 最新 state_version 只通过返回值 states[] 下发给调用方，
|
||
# 事件行保持写入时的 state_version（=本次事件 seq 对应的水位）。
|
||
# 5) 快照（掉线补齐/离线兜底）
|
||
if snapshot is not None:
|
||
await tx.insert(SNAPSHOT_TABLE, build_row(SNAPSHOT_TABLE, {
|
||
"tenant_id": tenant_id, "session_id": session_id,
|
||
"snapshot_uid": _digest({"u": uid, "s": snapshot})[:32],
|
||
"state_fingerprint": _digest(snapshot),
|
||
"entities_json": _dumps(snapshot),
|
||
"events_through_seq": seq_no,
|
||
"reason": str(snapshot.get("reason") if isinstance(snapshot, dict)
|
||
else "apply_runtime_event")[:32],
|
||
"created_at": now, "updated_at": now,
|
||
}))
|
||
|
||
# 6) 提交（任一 DML 失败 → 下面 rollback，事件行不存在）
|
||
await tx.commit()
|
||
stmt_count = tx.stmt_count
|
||
except Exception:
|
||
await tx.rollback()
|
||
raise
|
||
# ---- 事务已提交（with 块正常退出）：此后才允许广播，杜绝幽灵更新 ----
|
||
ms = round((time.perf_counter() - t0) * 1000.0, 3)
|
||
if dedup_result is not None:
|
||
dedup_result["latency_ms"] = ms
|
||
_lat.record("adjudicate", ms)
|
||
return dedup_result
|
||
breach = _lat.record("adjudicate", ms)
|
||
evt = {
|
||
"tenant_id": tenant_id, "world_id": world_id, "session_id": session_id,
|
||
"event_id": event_id, "event_uid": uid, "seq": seq_no,
|
||
"seq_no": seq_no, # 兼容旧字段名（对外响应别名）
|
||
"event_type": str(event_type), "payload": payload or {},
|
||
"states": state_events, "idem_key": uid,
|
||
"occurred_at": now, "source": "apply_runtime_event",
|
||
}
|
||
try:
|
||
b_ms = hub.publish(channel, evt)
|
||
_lat.record("broadcast", b_ms)
|
||
_lat.record("end_to_end", ms + b_ms)
|
||
except Exception as exc: # noqa: BLE001
|
||
# 数据已提交，广播失败不回滚事实；客户端由 3s 轮询兜底补齐
|
||
raise rtx_db.RtxError(ERR_BROADCAST_AFTER_COMMIT,
|
||
"数据已提交但广播失败（poll 兜底）：%r" % exc)
|
||
return {
|
||
"ok": True, "dedup": False, "event_id": event_id, "event_uid": uid,
|
||
"seq": seq_no, "seq_no": seq_no,
|
||
"cursor": evt.get("cursor"), "state_rows": len(state_events),
|
||
"states": state_events, "latency_ms": ms, "broadcast_ms": b_ms,
|
||
"channel": channel, "sla_breach": bool(breach), "stmt_count": stmt_count,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------- 读
|
||
async def read_states(tenant_id=None, session_id=None, entity_ids=None,
|
||
world_id=None, env=None):
|
||
"""读实体状态（tenant_id 强制；entity_ids 可选过滤）。返回 list[dict]。"""
|
||
tenant_id = _require_tenant(tenant_id)
|
||
sql = ("SELECT %s, state_json, state_version, checksum, updated_by, updated_at "
|
||
"FROM %s WHERE tenant_id=%%s" % (STATE_COL_KEY, STATE_TABLE))
|
||
params = [tenant_id]
|
||
if session_id not in (None, ""):
|
||
sql += " AND session_id=%s"
|
||
params.append(session_id)
|
||
if entity_ids:
|
||
sql += " AND %s IN (%s)" % (STATE_COL_KEY, ",".join(["%s"] * len(entity_ids)))
|
||
params.extend([str(e) for e in entity_ids])
|
||
sql += " ORDER BY %s ASC LIMIT 500" % STATE_COL_KEY
|
||
rows = await rtx_db.q_all(sql, tuple(params))
|
||
out = []
|
||
for r in rows:
|
||
r = dict(r)
|
||
state = _loads(r.get(STATE_COL_VALUE), {})
|
||
r["state"] = state
|
||
r["state_value"] = state # 对外别名
|
||
r["entity_id"] = r.get(STATE_COL_KEY)
|
||
out.append(r)
|
||
return out
|
||
|
||
|
||
async def poll_events(tenant_id=None, world_id=None, session_id=None,
|
||
after_cursor=None, after_seq=None, limit=200, env=None):
|
||
"""轮询兜底：先取 hub 增量；未命中/有缺口/带 after_seq 时退回查库（权威源）。
|
||
|
||
返回 {ok, events, next_cursor, next_seq, source: hub|db, truncated}
|
||
"""
|
||
tenant_id = _require_tenant(tenant_id)
|
||
hub = _hub.get_hub()
|
||
channel = _hub.channel_of(world_id, tenant_id)
|
||
limit = max(1, min(int(limit or 200), 500))
|
||
events, truncated = hub.pull(channel, after_cursor, limit)
|
||
if events and not truncated and after_seq in (None, ""):
|
||
return {"ok": True, "events": events, "next_cursor": events[-1]["cursor"],
|
||
"next_seq": events[-1].get("seq"), "source": "hub",
|
||
"truncated": False}
|
||
|
||
sql = ("SELECT id, %s, event_type, %s, %s, occurred_at, created_at FROM %s "
|
||
"WHERE tenant_id=%%s"
|
||
% (EVENT_COL_SEQ, EVENT_COL_PAYLOAD, EVENT_COL_STATE_VERSION, EVENT_TABLE))
|
||
params = [tenant_id]
|
||
if world_id not in (None, ""):
|
||
sql += " AND world_id=%s"
|
||
params.append(world_id)
|
||
if session_id not in (None, ""):
|
||
sql += " AND session_id=%s"
|
||
params.append(session_id)
|
||
if after_seq not in (None, ""):
|
||
sql += " AND %s > %s" % (EVENT_COL_SEQ, "%s")
|
||
params.append(int(after_seq))
|
||
elif truncated and events:
|
||
sql += " AND %s >= %s" % (EVENT_COL_SEQ, "%s")
|
||
params.append(int(events[0].get("seq") or 0))
|
||
sql += " ORDER BY %s ASC LIMIT %s" % (EVENT_COL_SEQ, int(limit))
|
||
rows = await rtx_db.q_all(sql, tuple(params))
|
||
out = []
|
||
for r in rows:
|
||
r = dict(r)
|
||
r["payload"] = _loads(r.get(EVENT_COL_PAYLOAD), {})
|
||
r["seq"] = int(r.get(EVENT_COL_SEQ) or 0)
|
||
r["seq_no"] = r["seq"]
|
||
r["channel"] = channel
|
||
out.append(r)
|
||
next_seq = out[-1]["seq"] if out else (int(after_seq) if after_seq not in (None, "")
|
||
else 0)
|
||
return {"ok": True, "events": out, "next_cursor": hub.latest_cursor(),
|
||
"next_seq": next_seq, "source": "db", "truncated": bool(truncated)}
|
||
|
||
|
||
# ---------------------------------------------------------------- 自检
|
||
def self_check():
|
||
"""离线自检：幂等键派生、append-only 守卫、列名与 models 对齐、契约协程签名。"""
|
||
msgs, ok = [], True
|
||
|
||
k1 = idem_key_of(7, 11, "move", {"x": 1})
|
||
k2 = idem_key_of(7, 11, "move", {"x": 1})
|
||
k3 = idem_key_of(7, 11, "move", {"x": 2})
|
||
if k1 != k2 or k1 == k3:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：幂等键不稳定/不敏感 %s %s %s" % (k1, k2, k3))
|
||
elif len(k1) > 32:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：幂等键超 event_uid 长度 32（%d）" % len(k1))
|
||
else:
|
||
msgs.append("tx_write PASS：幂等键同参同键、异参异键、≤32 字符（%s…）" % k1[:12])
|
||
if idem_key_of(7, 11, "move", {}, client_key="cli-9") != "cli-9":
|
||
ok = False
|
||
msgs.append("tx_write FAIL：client_event_id 未原样透传")
|
||
else:
|
||
msgs.append("tx_write PASS：client_event_id 原样透传（重放同键）")
|
||
|
||
blocked = []
|
||
for op in ("UPDATE", "DELETE", "TRUNCATE"):
|
||
try:
|
||
assert_append_only(op, EVENT_TABLE)
|
||
ok = False
|
||
msgs.append("tx_write FAIL：append-only 未拦 %s" % op)
|
||
except rtx_db.RtxError as exc:
|
||
if exc.code != ERR_APPEND_ONLY:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：错误码 %s" % exc.code)
|
||
else:
|
||
blocked.append(op)
|
||
if blocked:
|
||
msgs.append("tx_write PASS：pbl_runtime_event 禁 %s" % "/".join(blocked))
|
||
try:
|
||
assert_append_only("UPDATE", "world")
|
||
ok = False
|
||
msgs.append("tx_write FAIL：基表写未拦截")
|
||
except rtx_db.RtxError:
|
||
msgs.append("tx_write PASS：基表写保护生效（world 等 %d 张）"
|
||
% len(PROTECTED_BASE_TABLES))
|
||
|
||
# 列名对齐自检：本文件写入用到的每一列都必须存在于 models/*.json
|
||
plan = {
|
||
EVENT_TABLE: ["tenant_id", EVENT_COL_UID, "session_id", "world_id", "actor_id",
|
||
"event_type", EVENT_COL_PAYLOAD, "causation_id",
|
||
EVENT_COL_STATE_VERSION, EVENT_COL_SEQ, "occurred_at",
|
||
"created_at", "updated_at"],
|
||
STATE_TABLE: ["tenant_id", "session_id", STATE_COL_KEY, STATE_COL_VALUE,
|
||
EVENT_COL_STATE_VERSION, "checksum", "updated_by",
|
||
"created_at", "updated_at"],
|
||
SNAPSHOT_TABLE: ["tenant_id", "session_id", "snapshot_uid", "state_fingerprint",
|
||
"entities_json", "events_through_seq", "reason", "created_at",
|
||
"updated_at"],
|
||
}
|
||
for tbl, cols in plan.items():
|
||
real = table_columns(tbl)
|
||
if not real:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：models/%s.json 不可读" % tbl)
|
||
continue
|
||
drift = [c for c in cols if c not in real]
|
||
if drift:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：%s 列漂移 %s（真实列=%s）" % (tbl, drift, real))
|
||
else:
|
||
msgs.append("tx_write PASS：%s 用到的 %d 列与 models 定义一致" % (tbl, len(cols)))
|
||
|
||
# ---- PBL-RTX-0004 闭环门禁（M11b-2c-A3）：误报侧不报 + 漏报侧必报 ----
|
||
try:
|
||
upd_row = build_row(STATE_TABLE, {
|
||
STATE_COL_VALUE: "{}", EVENT_COL_STATE_VERSION: 2, "checksum": "c",
|
||
"updated_by": "server", "updated_at": "2026-09-21 12:00:00",
|
||
}, strict=True, full_row=False)
|
||
expect = [STATE_COL_VALUE, EVENT_COL_STATE_VERSION, "checksum", "updated_by",
|
||
"updated_at"]
|
||
if sorted(upd_row.keys()) != sorted(expect):
|
||
ok = False
|
||
msgs.append("tx_write FAIL：UPDATE SET 列被意外裁剪 %s（应为 %s）"
|
||
% (sorted(upd_row.keys()), sorted(expect)))
|
||
else:
|
||
msgs.append("tx_write PASS：UPDATE SET 语义不误报缺 NOT NULL 列"
|
||
"（PBL-RTX-0004 误报侧已闭环）")
|
||
except rtx_db.RtxError as exc:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：UPDATE SET 仍被 %s 拒绝（PBL-RTX-0004 未闭环）" % exc.code)
|
||
|
||
try:
|
||
build_row(STATE_TABLE, {"session_id": 11, STATE_COL_KEY: "pos"})
|
||
ok = False
|
||
msgs.append("tx_write FAIL：INSERT 真缺 NOT NULL 列未被拦（门禁恒真 = 漏报回归）")
|
||
except rtx_db.RtxError as exc:
|
||
if exc.code == "PBL-RTX-0004":
|
||
msgs.append("tx_write PASS：INSERT 真缺列仍被 PBL-RTX-0004 拦下（fail-closed 未弱化）")
|
||
else:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：INSERT 缺列错误码异常 %s（应为 PBL-RTX-0004）" % exc.code)
|
||
|
||
# 事件表用 nullable:"no" 写法，识别修复后必填集合必须非空（否则门禁又退回恒真）
|
||
for tbl in (EVENT_TABLE, STATE_TABLE, SNAPSHOT_TABLE):
|
||
req = required_columns(tbl)
|
||
if not req:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：%s required_columns 为空"
|
||
"（模型 NOT NULL 写法未被识别，INSERT 门禁失效）" % tbl)
|
||
else:
|
||
msgs.append("tx_write PASS：%s 必填列 %d 个（已排除自增/DEFAULT 列）：%s"
|
||
% (tbl, len(req), ",".join(req)))
|
||
|
||
# 只给「必填列真子集」必须被拦（事件表缺 event_type / 状态表缺 tenant_id）
|
||
for tbl, partial in ((EVENT_TABLE, {EVENT_COL_UID: "u1"}),
|
||
(STATE_TABLE, {"session_id": 11, STATE_COL_KEY: "pos",
|
||
EVENT_COL_STATE_VERSION: 1, "checksum": "c",
|
||
"updated_by": "a", "created_at": "2026-09-21",
|
||
"updated_at": "2026-09-21"})):
|
||
try:
|
||
build_row(tbl, partial)
|
||
ok = False
|
||
msgs.append("tx_write FAIL：%s 只给部分必填列却通过校验" % tbl)
|
||
except rtx_db.RtxError as exc:
|
||
if exc.code != "PBL-RTX-0004":
|
||
ok = False
|
||
msgs.append("tx_write FAIL：%s 部分列错误码 %s" % (tbl, exc.code))
|
||
else:
|
||
msgs.append("tx_write PASS：%s 部分必填列被 PBL-RTX-0004 拦下" % tbl)
|
||
|
||
# 写路径实际使用的 INSERT 列集合必须通过（防修复过头把主链路挡死）
|
||
try:
|
||
build_row(EVENT_TABLE, {
|
||
"tenant_id": "7", EVENT_COL_UID: "u", "session_id": 11, "world_id": 1,
|
||
"actor_id": "server", "event_type": "move", EVENT_COL_PAYLOAD: "{}",
|
||
"causation_id": "", EVENT_COL_STATE_VERSION: 0, EVENT_COL_SEQ: 1,
|
||
"occurred_at": "2026-09-21 12:00:00", "created_at": "2026-09-21 12:00:00",
|
||
"updated_at": "2026-09-21 12:00:00"})
|
||
build_row(SNAPSHOT_TABLE, {
|
||
"tenant_id": "7", "session_id": 11, "snapshot_uid": "s",
|
||
"state_fingerprint": "f", "entities_json": "{}", "events_through_seq": 1,
|
||
"reason": "apply_runtime_event", "created_at": "2026-09-21",
|
||
"updated_at": "2026-09-21"})
|
||
msgs.append("tx_write PASS：事件表/快照表 INSERT 写路径列集合通过校验（无回归）")
|
||
except rtx_db.RtxError as exc:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：写路径 INSERT 列集合被 %s 拒绝（修复过头）" % exc.code)
|
||
|
||
# 不可变列守卫：SET tenant_id/created_at 必须被 PBL-RTX-0006 拒绝
|
||
try:
|
||
build_row(STATE_TABLE, {"tenant_id": "7", STATE_COL_VALUE: "{}"},
|
||
strict=True, full_row=False)
|
||
ok = False
|
||
msgs.append("tx_write FAIL：UPDATE SET 不可变列未被拦")
|
||
except rtx_db.RtxError as exc:
|
||
if exc.code == "PBL-RTX-0006":
|
||
msgs.append("tx_write PASS：UPDATE 禁止 SET 不可变列（PBL-RTX-0006）")
|
||
else:
|
||
ok = False
|
||
msgs.append("tx_write FAIL：不可变列错误码 %s（应为 PBL-RTX-0006）" % exc.code)
|
||
|
||
for name, fn in (("apply_runtime_event", apply_runtime_event),
|
||
("poll_events", poll_events), ("read_states", read_states)):
|
||
if not inspect.iscoroutinefunction(fn):
|
||
ok = False
|
||
msgs.append("tx_write FAIL：%s 非协程" % name)
|
||
if callable(assert_append_only):
|
||
msgs.append("tx_write PASS：3 协程契约 + 1 同步守卫签名正确")
|
||
|
||
# ---- 端到端写入探针（QC #2：把 state_updates 非空路径纳入自检覆盖）----
|
||
# 上一版占位死代码 tx.execute("SELECT 1", (None, None)) 只在 state_updates 非空
|
||
# 时执行，离线自检不调用 apply_runtime_event 因而漏检（假阴性）。此处强制跑
|
||
# m11b_probe 的内存事务引擎探针，任何占位/参数数不匹配/幽灵广播都会 FAIL。
|
||
try:
|
||
from . import m11b_probe as _probe # 延迟导入：避免模块级循环依赖
|
||
p_ok, p_msgs = _probe.run_probes()
|
||
ok = ok and p_ok
|
||
msgs.extend(["probe| " + m for m in p_msgs])
|
||
except Exception as exc: # noqa: BLE001
|
||
ok = False
|
||
msgs.append("tx_write FAIL：m11b_probe 不可用（写入路径未被自检覆盖）%r" % exc)
|
||
|
||
if ok:
|
||
msgs.append("tx_write.self_check PASS")
|
||
return ok, msgs
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_ok, _msgs = self_check()
|
||
for _m in _msgs:
|
||
print(_m)
|
||
raise SystemExit(0 if _ok else 1)
|