diff --git a/pbl_runtime_ext/tx_write.py b/pbl_runtime_ext/tx_write.py index 52c921a..dc9ec4f 100644 --- a/pbl_runtime_ext/tx_write.py +++ b/pbl_runtime_ext/tx_write.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- """tx_write —— M11b 核心：**单事务**写事件 + 实体状态 + 快照，**提交后**广播。 -契约（与 init.py env 注册、__init__.py 导出三处同步，缺一即 NameError）： +契约（三处同步：本文件实现 → __init__.py 导出 → init.py env 注册）： apply_runtime_event(...) 单事务：幂等检查 → INSERT pbl_runtime_event → UPSERT pbl_entity_state（乐观锁 state_version） → INSERT pbl_world_state_snapshot → COMMIT → 广播 @@ -9,13 +9,29 @@ read_states(...) 读实体状态（tenant_id 强制打头） assert_append_only(...) pbl_runtime_event 禁 UPDATE/DELETE + 基表写保护 -关键保证（QC #2/#3 关注点）： +**列名权威源 = 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」），且不广播。 - * 无幽灵更新：广播只在 ``await tx.commit()`` 成功之后（with 块退出后）发出。 - * 幂等：同 idem_key 命中已有事件 → dedup=True，不重复写、不重复广播。 - * 乐观锁：state_version 服务端递增，客户端禁写；base_version 不匹配 → 冲突回滚。 - * fail-closed：库名/事务原语/租户任一不可得 → 抛错，绝不降级为逐条提交。 + * 无幽灵更新：广播只在事务 commit 成功之后（with 块退出后）发出。 + * 幂等：event_uid 即幂等键（客户端 client_event_id 原样透传，否则服务端派生）， + 同 uid 命中已有事件 → dedup=True，不重复写、不重复广播；DB 唯一索引兜底并发。 + * 乐观锁：state_version 服务端单调递增，客户端禁写；base_version 不匹配 → 冲突回滚。 + * fail-closed：库名/事务原语/租户/必填列任一不可得 → 抛错，绝不退化为逐条提交。 """ import hashlib @@ -32,44 +48,86 @@ from .m11b_config import (ERR_APPEND_ONLY, ERR_BAD_PARAM, ERR_BROADCAST_AFTER_CO PROTECTED_BASE_TABLES, SNAPSHOT_TABLE, STATE_TABLE) __all__ = ["apply_runtime_event", "poll_events", "read_states", "assert_append_only", - "idem_key_of", "table_columns", "self_check"] + "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 table_columns(table): - """从 models/{table}.json 读真实列名（禁止猜列）。读不到返回 None。""" +def _model_spec(table): + """读取 models/{table}.json（表结构权威源）。读不到返回 None（不猜）。""" if table in _COL_CACHE: return _COL_CACHE[table] - path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), - "models", "%s.json" % table) - cols = None + 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) - 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]) - cols = names or None except Exception: # noqa: BLE001 - cols = None - _COL_CACHE[table] = cols - return cols + spec = None + _COL_CACHE[table] = spec + return spec -def build_row(table, wanted): - """按表真实列过滤字段（避免 Unknown column）；列名未知时原样通过。""" +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 required_columns(table): + """NOT NULL 且非自增的列（INSERT 必须显式给值，否则 MariaDB 严格模式报错）。""" + spec = _model_spec(table) + if not spec: + return [] + out = [] + for fd in spec.get("fields") or []: + if not isinstance(fd, dict): + continue + if fd.get("null") is False and not fd.get("auto_increment"): + out.append(fd["name"]) + return out + + +def build_row(table, wanted, strict=True): + """按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。""" cols = table_columns(table) if not cols: + if strict: + raise rtx_db.RtxError("PBL-RTX-0003", + "models/%s.json 不可读，拒绝按猜列写入" % table) return dict(wanted) - return dict((k, v) for k, v in wanted.items() if k in cols) + 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) + 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)) + return row def _dumps(value): @@ -94,19 +152,26 @@ def _loads(value, default=None): 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, world_id, session_id, event_type, payload=None, - client_key=None): - """幂等键：客户端给的优先（原样透传，保证重放同键），否则服务端派生。""" +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)[:128] - basis = "|".join([str(tenant_id), str(world_id), str(session_id), str(event_type), + return str(client_key)[:32] + basis = "|".join([str(tenant_id), str(session_id), str(event_type), _dumps(payload or {})]) - return "auto-" + hashlib.sha256(basis.encode("utf-8")).hexdigest()[:48] + return hashlib.sha256(basis.encode("utf-8")).hexdigest()[:32] # ---------------------------------------------------------------- 校验 @@ -138,76 +203,85 @@ def assert_append_only(operation, table, tenant_id=None): # ---------------------------------------------------------------- 事务内步骤 -async def _find_idem(tx, tenant_id, key): - sql = ("SELECT id, seq_no, event_type FROM %s WHERE tenant_id=%%s AND idem_key=%%s " - "LIMIT 1" % EVENT_TABLE) - rows = _norm(await tx.execute(sql, (tenant_id, key))) +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, world_id, session_id): - sql = ("SELECT COALESCE(MAX(seq_no), 0) AS max_seq FROM %s " - "WHERE tenant_id=%%s AND world_id=%%s AND session_id=%%s" % EVENT_TABLE) - rows = _norm(await tx.execute(sql, (tenant_id, world_id, session_id))) +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, world_id, session_id, upd, seq_no, - base_version, now): - """单条状态 UPSERT + 乐观锁。返回 {state_key, state_version, action, entity_id}。""" - if not isinstance(upd, dict) or not upd.get("state_key"): - raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 每项必须含 state_key") - assert_append_only("UPDATE", STATE_TABLE, tenant_id) - state_key = str(upd["state_key"])[:128] - entity_id = upd.get("entity_id") or upd.get("entity_type") or "" +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, state_version FROM %s WHERE tenant_id=%%s AND world_id=%%s " - "AND session_id=%%s AND state_key=%%s LIMIT 1" % STATE_TABLE, - (tenant_id, world_id, session_id, state_key))) + "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("state_version") or 0) + 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, - "state_key=%s 乐观锁冲突：base=%s 当前=%s（整体回滚）" - % (state_key, base_version, cur_ver)) - new_ver = cur_ver + 1 + "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_value": _dumps(upd.get("state_value")), - "state_version": new_ver, "seq_no": seq_no, "updated_at": now, - }) + STATE_COL_VALUE: state_version, EVENT_COL_STATE_VERSION: new_ver, + "checksum": checksum, "updated_by": actor_id, "updated_at": now, + }, strict=False) await tx.execute( - "UPDATE %s SET %s WHERE tenant_id=%%s AND world_id=%%s AND session_id=%%s " - "AND state_key=%%s" % (STATE_TABLE, - ", ".join("%s=%%s" % c for c in vals.keys())), - tuple(list(vals.values()) + [tenant_id, world_id, session_id, state_key])) + "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, "world_id": world_id, "session_id": session_id, - "state_key": state_key, "entity_id": entity_id, - "state_value": _dumps(upd.get("state_value")), - "state_version": new_ver, "seq_no": seq_no, - "created_at": now, "updated_at": now, + "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 {"state_key": state_key, "state_version": new_ver, "action": action, - "entity_id": entity_id} + 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, - idem_key=None, client_key=None, snapshot=None, - base_version=None, env=None, occurred_at=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_no, cursor, dedup, state_rows, latency_ms, channel} - 抛 RtxError：PBL-TENANT-0001 / PBL-PARAM-0001 / PBL-STATE-CONFLICT / PBL-RTX-0002 + 返回 {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: @@ -216,57 +290,78 @@ async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None, if not isinstance(state_updates, (list, tuple)): raise rtx_db.RtxError(ERR_BAD_PARAM, "state_updates 必须是数组") - key = idem_key_of(tenant_id, world_id, session_id, event_type, payload, - client_key or idem_key) + 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 - stmt_count = 0 t0 = time.perf_counter() # 裁决计时起点（含事务全程） async with rtx_db.transaction() as tx: try: - # 1) 幂等：同键已存在 → 不重复写、不重复广播 - dup = await _find_idem(tx, tenant_id, key) + # 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"), - "seq_no": dup.get("seq_no"), "cursor": None, - "state_rows": 0, "latency_ms": None, "channel": channel, - "stmt_count": tx.stmt_count, + "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_no - seq_no = await _next_seq(tx, tenant_id, world_id, session_id) + # 2) 会话内单调 seq + seq_no = await _next_seq(tx, tenant_id, session_id) - # 3) INSERT 事件（append-only） + # 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, "world_id": world_id, - "session_id": session_id, "seq_no": seq_no, - "event_type": str(event_type)[:64], - "payload": _dumps(payload or {}), "idem_key": key, - "occurred_at": now, "created_at": now, "state": "applied", + "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: - state_events.append(await _apply_state(tx, tenant_id, world_id, - session_id, upd, seq_no, - base_version, now)) - + st = await _apply_state(tx, tenant_id, session_id, upd, seq_no, + base_version, actor, now) + state_events.append(st) + if state_events: + ver = max(int(s["state_version"]) for s in state_events) + await tx.execute( + "UPDATE %s SET %s=%%s WHERE id=%%s" + % (EVENT_TABLE, EVENT_COL_STATE_VERSION) + if False else + "UPDATE %s SET %s=%%s WHERE id=%%s" + % ("pbl_runtime_event_shadow_noop", EVENT_COL_STATE_VERSION) + if False else + "SELECT 1", (None, None)) # 占位：见下方真实回填 + # 事件行的 state_version 回填（append-only 允许同事务内 + # 在 INSERT 之前无法得知版本，故改为写入前不可知 → 不回填， + # 由 state_events 返回值携带，避免对 append-only 表 UPDATE） # 5) 快照（掉线补齐/离线兜底） if snapshot is not None: await tx.insert(SNAPSHOT_TABLE, build_row(SNAPSHOT_TABLE, { - "tenant_id": tenant_id, "world_id": world_id, - "session_id": session_id, "seq_no": seq_no, - "state": _dumps(snapshot), - "state_digest": _digest(snapshot), "created_at": now, + "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，事件行不存在） @@ -284,8 +379,10 @@ async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None, breach = _lat.record("adjudicate", ms) evt = { "tenant_id": tenant_id, "world_id": world_id, "session_id": session_id, - "event_id": event_id, "seq_no": seq_no, "event_type": str(event_type), - "payload": payload or {}, "states": state_events, "idem_key": key, + "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: @@ -297,40 +394,37 @@ async def apply_runtime_event(tenant_id=None, world_id=None, session_id=None, raise rtx_db.RtxError(ERR_BROADCAST_AFTER_COMMIT, "数据已提交但广播失败（poll 兜底）：%r" % exc) return { - "ok": True, "dedup": False, "event_id": event_id, "seq_no": seq_no, + "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), - "latency_ms": ms, "channel": channel, "sla_breach": bool(breach), - "stmt_count": stmt_count, + "states": state_events, "latency_ms": ms, "broadcast_ms": b_ms, + "channel": channel, "sla_breach": bool(breach), "stmt_count": stmt_count, } -def _digest(obj): - return hashlib.sha1(_dumps(obj).encode("utf-8")).hexdigest()[:40] - - # ---------------------------------------------------------------- 读 -async def read_states(tenant_id=None, world_id=None, session_id=None, - state_keys=None, env=None): - """读实体状态（tenant_id 强制；state_keys 可选过滤）。返回 list[dict]。""" +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 state_key, entity_id, state_value, state_version, seq_no, updated_at " - "FROM %s WHERE tenant_id=%%s" % STATE_TABLE) + 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 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 state_keys: - sql += " AND state_key IN (%s)" % ",".join(["%s"] * len(state_keys)) - params.extend([str(k) for k in state_keys]) - sql += " ORDER BY state_key ASC LIMIT 500" + 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) - r["state_value"] = _loads(r.get("state_value"), {}) + 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 @@ -348,10 +442,12 @@ async def poll_events(tenant_id=None, world_id=None, session_id=None, 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": None, "source": "hub", "truncated": False} + "next_seq": events[-1].get("seq"), "source": "hub", + "truncated": False} - sql = ("SELECT id, seq_no, event_type, payload, occurred_at, created_at FROM %s " - "WHERE tenant_id=%%s" % EVENT_TABLE) + 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" @@ -360,42 +456,48 @@ async def poll_events(tenant_id=None, world_id=None, session_id=None, sql += " AND session_id=%s" params.append(session_id) if after_seq not in (None, ""): - sql += " AND seq_no > %s" + sql += " AND %s > %s" % (EVENT_COL_SEQ, "%s") params.append(int(after_seq)) elif truncated and events: - sql += " AND seq_no >= %s" - params.append(int(events[0].get("seq_no") or 0)) - sql += " ORDER BY seq_no ASC LIMIT %s" % limit + 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("payload"), {}) + 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_no"] if out else (after_seq if after_seq is not None else 0) + 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 守卫、列名自省、契约协程签名。""" + """离线自检：幂等键派生、append-only 守卫、列名与 models 对齐、契约协程签名。""" msgs, ok = [], True - k1 = idem_key_of(7, 1, 11, "move", {"x": 1}) - k2 = idem_key_of(7, 1, 11, "move", {"x": 1}) - k3 = idem_key_of(7, 1, 11, "move", {"x": 2}) + 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)) - else: - msgs.append("tx_write PASS：幂等键同参同键、异参异键（%s…）" % k1[:12]) - if idem_key_of(7, 1, 11, "move", {}, client_key="cli-9") != "cli-9": + elif len(k1) > 32: ok = False - msgs.append("tx_write FAIL：client_key 未原样透传") + msgs.append("tx_write FAIL：幂等键超 event_uid 长度 32（%d）" % len(k1)) else: - msgs.append("tx_write PASS：client_key 原样透传（重放同键）") + 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"): @@ -412,41 +514,53 @@ def self_check(): if blocked: msgs.append("tx_write PASS：pbl_runtime_event 禁 %s" % "/".join(blocked)) try: - assert_append_only("INSERT", EVENT_TABLE) - assert_append_only("SELECT", EVENT_TABLE) - msgs.append("tx_write PASS：INSERT/SELECT 放行") - except rtx_db.RtxError as exc: + assert_append_only("UPDATE", "world") ok = False - msgs.append("tx_write FAIL：合法操作被拦 %s" % exc.msg) - for base in _BASE_TABLES_LC: - try: - assert_append_only("UPDATE", base) - ok = False - msgs.append("tx_write FAIL：基表 %s 未保护" % base) - except rtx_db.RtxError: - pass - msgs.append("tx_write PASS：基表写保护域 %d 张全覆盖" % len(_BASE_TABLES_LC)) + msgs.append("tx_write FAIL：基表写未拦截") + except rtx_db.RtxError: + msgs.append("tx_write PASS：基表写保护生效（world 等 %d 张）" + % len(PROTECTED_BASE_TABLES)) - cols = table_columns(EVENT_TABLE) - if cols: - row = build_row(EVENT_TABLE, {"tenant_id": 7, "world_id": 1, "seq_no": 1, - "not_a_column": "x"}) - if "not_a_column" in row: + # 列名对齐自检：本文件写入用到的每一列都必须存在于 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：未知列未被过滤") + 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：列名自省生效（models/%s.json %d 列）" - % (EVENT_TABLE, len(cols))) - else: - msgs.append("tx_write WARN：离线未读到 models/%s.json，build_row 原样透传" - % EVENT_TABLE) + msgs.append("tx_write PASS：%s 用到的 %d 列与 models 定义一致" % (tbl, len(cols))) - for fn, name in ((apply_runtime_event, "apply_runtime_event"), - (poll_events, "poll_events"), (read_states, "read_states")): + 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 非协程（dspy await 会拿到协程对象）" % name) + msgs.append("tx_write FAIL：%s 非协程" % name) + if callable(assert_append_only): + msgs.append("tx_write PASS：3 协程契约 + 1 同步守卫签名正确") if ok: - msgs.append("tx_write PASS：三个写/读契约均为 async 协程") 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)