diff --git a/pbl_runtime_ext/api.py b/pbl_runtime_ext/api.py index 0f87d7d..a4096b5 100644 --- a/pbl_runtime_ext/api.py +++ b/pbl_runtime_ext/api.py @@ -239,3 +239,32 @@ async def pbl_session_member_add(**kw): event_type='session.member.joined', payload_json={'user_id': user_id, 'members': cnt + 1}) return {'ok': True, 'members': cnt + 1, 'max_members': MAX_MEMBERS, 'deduped': False} + + +# ================================================================== M11b 覆盖 +# M11b（单事务事件+状态写入、按月分区、提交后广播、3s 轮询兜底）是**权威实现**： +# 同名契约在此覆盖上方 M11a 版本，避免两套逻辑漂移；返回结构向后兼容（只增字段）。 +# 主链路：m11b_api → tx_write.apply_runtime_event（一个事务写 事件+实体状态+快照 +# → 提交后才广播 → 分段计时裁决/广播/端到端并与 SLA 预算比对）。 +from .m11b_api import ( # noqa: E402 契约层覆盖（放最后，导入即生效） + contracts as m11b_contracts, + pbl_entity_state_apply, + pbl_entity_state_get, + pbl_runtime_broadcast_pull, + pbl_runtime_event_append, + pbl_runtime_broadcast_pull, + pbl_runtime_broadcast_stats, + pbl_runtime_event_poll, + pbl_world_broadcast, +) + +CONTRACTS = { + "pbl_runtime_event_append": pbl_runtime_event_append, + "pbl_runtime_event_poll": pbl_runtime_event_poll, + "pbl_runtime_broadcast_pull": pbl_runtime_broadcast_pull, + "pbl_runtime_broadcast_stats": pbl_runtime_broadcast_stats, + "pbl_entity_state_get": pbl_entity_state_get, + "pbl_entity_state_apply": pbl_entity_state_apply, + "pbl_world_broadcast": pbl_world_broadcast, + "pbl_session_member_add": pbl_session_member_add, +} diff --git a/pbl_runtime_ext/broadcast.py b/pbl_runtime_ext/broadcast.py new file mode 100644 index 0000000..b22dd8b --- /dev/null +++ b/pbl_runtime_ext/broadcast.py @@ -0,0 +1,263 @@ +"""M11b 提交后广播（post-commit broadcast）+ 3s 轮询兜底。 + +铁律与语义 +---------- +1. **只在事务提交成功后广播**：广播内容必须是已落库的事实。提交失败绝不广播， + 否则客户端会收到数据库里不存在的状态（幽灵更新，无法收敛）。 +2. **广播是尽力而为（best-effort），不是投递保证**：推送通道（websocket/SSE） + 抖动、进程重启都可能丢事件。丢广播**不撤销已提交事实**，客户端用 + `since_seq` 轮询补齐（兜底周期 3s），保证最终一致、不丢数据。 +3. **环形缓冲**：每个会话在内存里保留最近 N 条已提交事件，供 + `pbl_runtime_broadcast_pull` 做毫秒级增量拉取（比查库轻）， + 缓冲被覆盖时客户端自动退回查库轮询（poll 契约），不会漏。 +4. 广播耗时计入 SLA（≤300ms）：推送在后台线程做，主请求只负责入队 + + 记录 accepted；`pull` 契约用于验证「事件确实可被消费」。 + +线程安全：所有状态变更都在 RLock 下；推送回调在后台 worker 线程执行， +回调异常只计数不影响主链路。 +""" +import collections +import logging +import threading +import time + +from .m11b_config import ( + BROADCAST_QUEUE_SIZE, + BROADCAST_THREADS, + POLL_FALLBACK_INTERVAL_MS, + SLA_BROADCAST_MS, +) + +log = logging.getLogger("pbl_runtime_ext.broadcast") + + +def _now_ms(): + return int(time.time() * 1000) + + +class _Worker(object): + """极简后台推送线程（daemon），避免引入线程池依赖。""" + + def __init__(self, hub, idx): + self.hub = hub + self.idx = idx + self.stop = threading.Event() + self.thread = threading.Thread( + target=self._run, name="pbl-bcast-%d" % idx) + self.thread.daemon = True + + def start(self): + try: + self.thread.start() + except Exception as exc: # noqa: BLE001 + log.warning("广播 worker 启动失败（退回纯轮询）: %s", exc) + + def _run(self): + while not self.stop.is_set(): + item = self.hub._next_job() + if item is None: + self.stop.wait(0.01) + continue + self.hub._deliver(*item) + + +class BroadcastHub(object): + """会话级广播中心：入队 → 后台推送 → 环形缓冲供 pull/轮询补齐。""" + + def __init__(self, capacity=BROADCAST_QUEUE_SIZE, workers=BROADCAST_THREADS): + self._lock = threading.RLock() + self._cond = threading.Condition(self._lock) + self._queue = collections.deque(maxlen=int(capacity)) + self._buffers = {} # session_key -> deque(events) + self._capacity = int(capacity) + self._pushers = [] # [(name, callable)] + self._stats = { + "published": 0, "delivered": 0, "push_failed": 0, + "pulls": 0, "pull_hit": 0, "pull_miss": 0, + "sla_breached": 0, "last_broadcast_ms": 0, "max_broadcast_ms": 0, + } + self._workers = [] + if workers and int(workers) > 0: + for i in range(int(workers)): + w = _Worker(self, i) + w.start() + self._workers.append(w) + + # -------------------------------------------------- 通道注册 + def register_push(self, fn, name="custom"): + """注册推送通道（ws/sse）。fn(session_key, event) -> truthy/awaitable。""" + if not callable(fn): + raise TypeError("push 回调必须可调用") + with self._lock: + self._pushers.append((name, fn)) + return name + + def unregister_push(self, name): + with self._lock: + self._pushers = [(n, f) for (n, f) in self._pushers if n != name] + + # -------------------------------------------------- 发布 + def publish(self, session_key, event): + """提交后调用：先入环形缓冲（保证可被 pull 到），再排入推送队列。 + + 返回 {"accepted", "buffered", "channel", "seq_no", "pull_deadline_ms"}。 + 本方法不阻塞等待推送结果 —— 广播预算 ≤300ms 包含推送，主链路只记账号。 + """ + t0 = _now_ms() + with self._lock: + buf = self._buffers.setdefault( + str(session_key), collections.deque(maxlen=self._capacity)) + buf.append(event) + self._stats["published"] += 1 + channels = [n for n, _f in self._pushers] + self._queue.append((session_key, event, t0)) + try: + self._cond.notify() + except Exception: # noqa: BLE001 + pass + accepted = bool(channels) + # 无后台 worker（降级/自检模式）时同步投递，保证广播语义仍然成立 + if not self._workers and channels: + self._deliver(session_key, event, t0) + cost = _now_ms() - t0 + with self._lock: + self._stats["last_broadcast_ms"] = cost + self._stats["max_broadcast_ms"] = max(self._stats["max_broadcast_ms"], cost) + if cost > SLA_BROADCAST_MS: + self._stats["sla_breached"] += 1 + return {"accepted": accepted, "buffered": True, + "channel": "websocket" if accepted else "poll_fallback", + "channels": channels, + "seq_no": (event or {}).get("seq_no"), + "enqueue_ms": cost, + "budget_ms": SLA_BROADCAST_MS, + "pull_fallback_interval_ms": POLL_FALLBACK_INTERVAL_MS} + + publish_event = publish # 兼容别名 + + # -------------------------------------------------- 消费（pull） + def pull(self, session_key, since_seq=0, limit=200): + """从环形缓冲拉增量。miss（缓冲已覆盖）时返回 need_poll=True， + 客户端据此退回查库轮询契约（pbl_runtime_event_poll）。 + """ + since = int(since_seq or 0) + lim = max(1, min(int(limit or 200), 500)) + with self._lock: + self._stats["pulls"] += 1 + buf = self._buffers.get(str(session_key)) + if not buf: + self._stats["pull_miss"] += 1 + return {"events": [], "count": 0, "last_seq": since, + "buffered": False, "need_poll": True, + "poll_interval_ms": POLL_FALLBACK_INTERVAL_MS} + events = [e for e in buf if int(e.get("seq_no") or 0) > since] + events = events[:lim] + oldest = int(buf[0].get("seq_no") or 0) if buf else 0 + covered = (not events) or (oldest <= since + 1) or (since == 0) + last_seq = int(events[-1].get("seq_no")) if events else since + if events: + self._stats["pull_hit"] += 1 + if not covered: + self._stats["pull_miss"] += 1 + else: + self._stats["pull_hit"] += 0 + return {"events": events, "count": len(events), "last_seq": last_seq, + "buffered": True, "need_poll": not covered, + "poll_interval_ms": POLL_FALLBACK_INTERVAL_MS} + + def buffered_events(self, session_key): + with self._lock: + return list(self._buffers.get(str(session_key), [])) + + def forget(self, session_key): + """会话结束时释放内存缓冲。""" + with self._lock: + self._buffers.pop(str(session_key), None) + + # -------------------------------------------------- 内部推送 + def _next_job(self): + with self._cond: + if not self._queue: + return None + return self._queue.popleft() + + def _deliver(self, session_key, event, enqueued_ms=None): + ok_any = False + for name, fn in list(self._pushers_snapshot()): + try: + res = fn(session_key, event) + if hasattr(res, "__await__"): + res = _run_coro(res) + if res is None or res is False: + raise RuntimeError("push 返回 %r" % (res,)) + ok_any = True + except Exception as exc: # noqa: BLE001 + log.warning("广播通道 %s 投递失败（客户端 3s 轮询补齐）: %s", name, exc) + with self._lock: + self._stats["push_failed"] += 1 + if ok_any: + with self._lock: + self._stats["delivered"] += 1 + if enqueued_ms is not None: + cost = _now_ms() - int(enqueued_ms) + with self._lock: + self._stats["last_broadcast_ms"] = cost + self._stats["max_broadcast_ms"] = max(self._stats["max_broadcast_ms"], cost) + if cost > SLA_BROADCAST_MS: + self._stats["sla_breached"] += 1 + return ok_any + + def _pushers_snapshot(self): + with self._lock: + return list(self._pushers) + + # -------------------------------------------------- 观测 + def stats(self): + with self._lock: + s = dict(self._stats) + s["buffered_sessions"] = len(self._buffers) + s["queued"] = len(self._queue) + s["channels"] = [n for n, _f in self._pushers] + s["budget_ms"] = SLA_BROADCAST_MS + s["poll_fallback_interval_ms"] = POLL_FALLBACK_INTERVAL_MS + s["within_budget"] = s["max_broadcast_ms"] <= SLA_BROADCAST_MS + return s + + def shutdown(self): + for w in self._workers: + w.stop.set() + self._workers = [] + + +def _run_coro(coro): + """在后台线程里安全执行 awaitable（无运行中 loop 时新建）。""" + import asyncio + + try: + loop = asyncio.get_event_loop() + if loop.is_running(): + return asyncio.ensure_future(coro) + except RuntimeError: + pass + return asyncio.new_event_loop().run_until_complete(coro) + + +_HUB = None +_HUB_LOCK = threading.Lock() + + +def get_hub(): + """进程级单例广播中心。""" + global _HUB + if _HUB is None: + with _HUB_LOCK: + if _HUB is None: + _HUB = BroadcastHub() + return _HUB + + +def set_hub(hub): + global _HUB + with _HUB_LOCK: + _HUB = hub + return _HUB diff --git a/pbl_runtime_ext/init.py b/pbl_runtime_ext/init.py index dd9da64..80a4fed 100644 --- a/pbl_runtime_ext/init.py +++ b/pbl_runtime_ext/init.py @@ -14,6 +14,17 @@ load_pbl_runtime_ext(env=None)： from . import tables as _tables from . import tx_event as _tx +# M11b：单事务事件+状态写入与提交后广播（权威实现） +try: + from . import m11b_api as _m11b + from . import tx_write as _wtx +except Exception as _exc: # noqa: BLE001 缺依赖时不阻断 M11a 能力 + _m11b = None + _wtx = None + _M11B_IMPORT_ERROR = str(_exc) +else: + _M11B_IMPORT_ERROR = None + MODULE = "pbl_runtime_ext" EXPECTED_TABLES = 3 @@ -36,9 +47,27 @@ def api(): "EVENT_TABLE": _tx.EVENT_TABLE, "STATE_TABLE": _tx.STATE_TABLE, "SNAPSHOT_TABLE": _tx.SNAPSHOT_TABLE, + # ---- M11b ---- + "apply_runtime_event": getattr(_wtx, "apply_runtime_event", None), + "poll_events": getattr(_wtx, "poll_events", None), + "read_states": getattr(_wtx, "read_states", None), + "assert_append_only": getattr(_wtx, "assert_append_only", None), + "broadcast_hub": getattr(_m11b, "get_hub", lambda: None)() if _m11b else None, + "contracts": dict(getattr(_m11b, "contracts", lambda: {})()) if _m11b else {}, + "ensure_forward_partitions": _partitions_fn(), + "M11B": bool(_m11b and _wtx), } +def _partitions_fn(): + try: + from .partitions import ensure_forward_partitions + + return ensure_forward_partitions + except Exception: # noqa: BLE001 + return None + + def self_check(env=None): """模块自检：表契约 + 写保护 + 单事务/幂等契约。返回 (all_ok, msgs)。""" msgs = [] @@ -57,6 +86,16 @@ def self_check(env=None): all_ok = False msgs.extend(tx_msgs) + # M11b 自检：append-only 守卫 / 月分区算法 / 幂等键派生（离线不连库） + if _M11B_IMPORT_ERROR: + all_ok = False + msgs.append("M11b 模块导入 FAIL：%s" % _M11B_IMPORT_ERROR) + elif _wtx is not None: + m11b_ok, m11b_msgs = _wtx.self_check() + if not m11b_ok: + all_ok = False + msgs.extend(m11b_msgs) + # 幂等探针：同 idem_key 二次投递必须走 dedup 分支（离线用假 sor 验证） probe_ok, probe_msgs = _probe_idempotent() if not probe_ok: diff --git a/pbl_runtime_ext/latency.py b/pbl_runtime_ext/latency.py new file mode 100644 index 0000000..3598318 --- /dev/null +++ b/pbl_runtime_ext/latency.py @@ -0,0 +1,93 @@ +"""M11b 时延计量：裁决/广播/端到端三段预算的计时与 SLA 判定。 + +设计要点： + * 计时用单调时钟 perf_counter，避免系统时钟回拨造成负值； + * 超预算不抛异常（业务已落库，回滚反而破坏一致性），只在结果里 + 打 sla_breached 标记并落 WARN 日志，由监控/测试阶段据此判定缺陷； + * 所有对外响应都带 latency 段，客户端据此决定是否需要 3s 轮询兜底。 +""" +import logging +import time + +from .m11b_config import ( + POLL_FALLBACK_INTERVAL_MS, + SLA_ADJUDICATE_MS, + SLA_BROADCAST_MS, + SLA_E2E_MS, +) + +log = logging.getLogger("pbl_runtime_ext.latency") + +_PHASE_SLA = { + "adjudicate_ms": SLA_ADJUDICATE_MS, + "broadcast_ms": SLA_BROADCAST_MS, + "tx_ms": SLA_ADJUDICATE_MS, + "e2e_ms": SLA_E2E_MS, +} + + +def now_ms(): + """单调毫秒时间戳（仅用于计时，不作为业务时间）。""" + return int(time.perf_counter() * 1000) + + +class Stopwatch: + """分段计时器。start() 记起点，mark(phase) 结算一段耗时。""" + + __slots__ = ("_origin", "_marks", "_last") + + def __init__(self): + self._origin = now_ms() + self._last = self._origin + self._marks = {} + + def mark(self, phase): + cur = now_ms() + cost = max(0, cur - self._last) + self._last = cur + self._marks[phase] = cost + return cost + + def total_ms(self): + return max(0, now_ms() - self._origin) + + def snapshot(self): + return dict(self._marks) + + +def evaluate_sla(sw, extra_ms=None): + """结算 SLA：返回 (latency_dict, breached_list)。 + + extra_ms: 额外手工补记的段（例如跨进程回传耗时）。 + """ + marks = sw.snapshot() + if extra_ms: + marks.update({k: int(v) for k, v in extra_ms.items() if v is not None}) + marks["e2e_ms"] = sw.total_ms() + + breached = [] + for key, limit in _PHASE_SLA.items(): + val = marks.get(key) + if val is None: + continue + if val > limit: + breached.append({"phase": key, "cost_ms": val, "limit_ms": limit}) + + if breached: + log.warning("runtime SLA breached: %s", breached) + + latency = { + "adjudicate_ms": marks.get("adjudicate_ms", 0), + "tx_ms": marks.get("tx_ms", 0), + "broadcast_ms": marks.get("broadcast_ms", 0), + "e2e_ms": marks.get("e2e_ms", 0), + "sla": { + "adjudicate_ms": SLA_ADJUDICATE_MS, + "broadcast_ms": SLA_BROADCAST_MS, + "e2e_ms": SLA_E2E_MS, + "poll_fallback_interval_ms": POLL_FALLBACK_INTERVAL_MS, + }, + "breached": breached, + "within_sla": not breached, + } + return latency, breached diff --git a/pbl_runtime_ext/m11b_api.py b/pbl_runtime_ext/m11b_api.py new file mode 100644 index 0000000..8a11647 --- /dev/null +++ b/pbl_runtime_ext/m11b_api.py @@ -0,0 +1,299 @@ +"""M11b 契约实现层：把 .dspy 端点参数转成单事务写入/广播/轮询调用。 + +与 M11a 的 api.py 的关系：本文件是 M11b 的**权威实现**，api.py 末尾以 +`from .m11b_api import ...` 覆盖同名契约，使 + * pbl_runtime_event_append → 单事务 append-only 写事件（幂等） + * pbl_entity_state_apply → 单事务「事件 + 实体状态(+快照)」+ 提交后广播 + * pbl_runtime_event_poll → 先命中广播缓冲、未命中退回查库（3s 兜底） + * pbl_world_broadcast → 提交后广播状态查询 / 主动重播 + * pbl_runtime_broadcast_pull → 广播增量拉取（新增契约） +全部走同一条 tx_write 主链路，避免两套实现漂移。 + +约定：核心 tx_write 是同步阻塞（DB 驱动是同步的），契约层用线程池执行， +不阻塞事件循环；异常统一转结构化 {"ok": False, "error": {...}}， +错误码可断言（SLA/回滚/幂等/租户缺失）。 +""" +import asyncio +import functools +import logging + +from .m11b_config import ( + ERR_BAD_PAYLOAD, + ERR_DB, + ERR_STATE_CONFLICT, + ERR_TX_FAILED, + PBL_RT_TENANT_MISSING_CODE, +) +from .rtx_db import PblRtError, tenant_id as resolve_tenant +from .tx_write import ( + apply_runtime_event, + broadcast_stats, + poll_events, + read_states, +) + +log = logging.getLogger("pbl_runtime_ext.m11b_api") + +# 租户缺失在 pbl_common 侧的历史错误码，保持一致便于前端断言 +TENANT_MISSING_CODE = PBL_RT_TENANT_MISSING_CODE + + +def _int(v, default=0): + try: + return int(v) + except (TypeError, ValueError): + return int(default or 0) + + +def _client_authoritative_fields(kw): + """挑出客户端提交里的「服务端权威字段」，用于 fail-closed 拒绝。""" + from .m11b_config import FORBIDDEN_CLIENT_FIELDS + + return [k for k in (kw or {}) if k in FORBIDDEN_CLIENT_FIELDS + and (kw or {}).get(k) not in (None, "")] + + +def _run(fn, *a, **kw): + """把同步 DB 调用丢到默认线程池，保持事件循环响应（端到端预算含排队）。""" + loop = asyncio.get_event_loop() + return loop.run_in_executor(None, functools.partial(fn, *a, **kw)) + + +def _error(exc): + if isinstance(exc, PblRtError): + return {"ok": False, "error": exc.to_dict(), "code": exc.code} + return {"ok": False, "error": {"code": ERR_DB, "message": str(exc)[:300]}, + "code": ERR_DB} + + +# ------------------------------------------------------------------ 契约 +async def pbl_runtime_event_append(**kw): + """追加事件（append-only + 幂等 + 单事务；不写状态时状态列表为空）。""" + try: + tid = resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + if not kw.get("event_type"): + return {"ok": False, "code": ERR_BAD_PAYLOAD, + "error": {"code": ERR_BAD_PAYLOAD, "message": "缺少 event_type"}} + try: + res = await _run( + apply_runtime_event, + tenant_id=tid, + session_id=_int(kw.get("session_id") or kw.get("session")), + event_type=kw.get("event_type"), + payload=kw.get("payload") if kw.get("payload") is not None else kw.get("payload_json"), + world_id=_int(kw.get("world_id")), + entity_id=_int(kw.get("entity_id")), + actor_id=kw.get("actor_id"), + source=kw.get("source") or "runtime", + idem_key=kw.get("idem_key") or kw.get("event_uid"), + client_seq=kw.get("client_seq"), + causation_id=kw.get("causation_id"), + state_updates=None, + snapshot=bool(kw.get("snapshot")), + broadcast=kw.get("broadcast", True), + tx_group=kw.get("tx_group"), + client_fields=_client_authoritative_fields(kw), + ) + res["contract"] = "pbl_runtime_event_append" + return res + except Exception as exc: # noqa: BLE001 + log.exception("pbl_runtime_event_append 失败") + return _error(exc) + + +async def pbl_entity_state_apply(**kw): + """服务端裁决：单事务写事件 + 实体状态（+快照），提交后广播。 + + 客户端提交的 state_version 一律忽略（出现在禁写清单里 → 显式回告 + `client_version_ignored`；若同时出现其他权威字段则 fail-closed 拒绝）。 + """ + try: + tid = resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + session_id = _int(kw.get("session_id") or kw.get("session")) + entity_id = kw.get("entity_id") + if not session_id or entity_id in (None, ""): + return {"ok": False, "code": ERR_BAD_PAYLOAD, + "error": {"code": ERR_BAD_PAYLOAD, "message": "缺少 session_id/entity_id"}} + updates = kw.get("state_updates") + if not updates: + updates = [{"entity_id": entity_id, + "state_key": kw.get("state_key") or "default", + "state_value": kw.get("state_value") or kw.get("state_json") or {}, + "world_id": _int(kw.get("world_id")), + "merge": kw.get("merge", True), + "if_state_version": kw.get("if_state_version")}] + leaked = _client_authoritative_fields(kw) + strict = bool(kw.get("strict", False)) + if leaked and strict: + return {"ok": False, "code": ERR_BAD_PAYLOAD, + "error": {"code": ERR_BAD_PAYLOAD, + "message": "服务端权威字段客户端禁写: %s" % ",".join(leaked), + "detail": {"fields": leaked}}} + try: + res = await _run( + apply_runtime_event, + tenant_id=tid, + session_id=session_id, + event_type=kw.get("event_type") or "entity.state.changed", + payload=kw.get("payload") or {"entity_id": entity_id, + "state_key": (updates[0] or {}).get("state_key")}, + world_id=_int(kw.get("world_id")), + entity_id=_int(entity_id), + actor_id=kw.get("actor_id"), + source=kw.get("source") or "client_intent", + idem_key=kw.get("idem_key") or kw.get("event_uid"), + client_seq=kw.get("client_seq"), + causation_id=kw.get("causation_id"), + state_updates=updates, + snapshot=bool(kw.get("snapshot")), + broadcast=kw.get("broadcast", True), + if_state_version=kw.get("if_state_version"), + tx_group=kw.get("tx_group"), + client_fields=leaked, + ) + res["contract"] = "pbl_entity_state_apply" + res["client_version_ignored"] = kw.get("state_version") not in (None, "") + res["authoritative_fields_rejected"] = leaked if strict else [] + return res + except PblRtError as exc: + if exc.code in (ERR_STATE_CONFLICT, ERR_TX_FAILED): + log.info("状态裁决被拒 %s: %s", exc.code, exc.message) + return _error(exc) + except Exception as exc: # noqa: BLE001 + log.exception("pbl_entity_state_apply 失败") + return _error(exc) + + +async def pbl_runtime_event_poll(**kw): + """3s 轮询兜底：先命中广播缓冲，未命中退回查库（幂等，不重复消费）。""" + try: + tid = resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + try: + res = await _run( + poll_events, + tenant_id=tid, + session_id=_int(kw.get("session_id") or kw.get("session")), + since_seq=_int(kw.get("since_seq") or kw.get("since")), + limit=_int(kw.get("limit") or 100, 100), + try_buffer=bool(kw.get("try_buffer", True)), + ) + res["contract"] = "pbl_runtime_event_poll" + res["poll_interval_ms"] = 3000 + return res + except Exception as exc: # noqa: BLE001 + log.exception("pbl_runtime_event_poll 失败") + return _error(exc) + + +async def pbl_runtime_broadcast_pull(**kw): + """新增契约：从广播环形缓冲拉增量（掉线补齐，比查库轻）。""" + try: + resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + from .broadcast import get_hub + + res = get_hub().pull("s:%s" % _int(kw.get("session_id")), + _int(kw.get("since_seq")), + _int(kw.get("limit") or 200, 200)) + res.update({"contract": "pbl_runtime_broadcast_pull", "ok": True, + "session_id": _int(kw.get("session_id")), + "stats": broadcast_stats() if kw.get("with_stats") else None}) + return res + + +async def pbl_world_broadcast(**kw): + """广播状态查询 / 主动重播（重播只推已提交事件，不重复写库）。""" + try: + tid = resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + session_id = _int(kw.get("session_id") or kw.get("session")) + replay = kw.get("replay") + try: + if replay: + res = await _run(_replay, tid, session_id, replay) + else: + res = {"ok": True, "stats": broadcast_stats(), "session_id": session_id} + res["contract"] = "pbl_world_broadcast" + res["fallback_poll_interval_ms"] = 3000 + res["note"] = "广播失败不撤销已提交事件，客户端 3s 轮询补齐（不丢数据）" + return res + except Exception as exc: # noqa: BLE001 + log.exception("pbl_world_broadcast 失败") + return _error(exc) + + +def _replay(tid, session_id, since_seq): + """把已提交但可能未送达的事件重新入队广播（不写库）。""" + from .broadcast import get_hub + + got = poll_events(tid, session_id, since_seq=_int(since_seq), limit=200, + try_buffer=False) + hub = get_hub() + published = 0 + for ev in got.get("events") or []: + hub.publish("s:%s" % session_id, { + "seq_no": ev.get("seq_no"), "event_id": ev.get("id"), + "event_type": ev.get("event_type"), "session_id": session_id, + "world_id": ev.get("world_id"), "state_version": ev.get("state_version"), + "states": [], "ts": ev.get("created_at")}) + published += 1 + return {"ok": True, "replayed": published, "events": got.get("count", 0), + "last_seq": got.get("last_seq"), "stats": hub.stats()} + + +async def pbl_entity_state_get(**kw): + """读实体状态（服务端权威版本；只读，不改基表）。""" + try: + tid = resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + try: + res = await _run( + read_states, + tenant_id=tid, + session_id=_int(kw.get("session_id") or kw.get("session")), + entity_id=kw.get("entity_id"), + since_version=_int(kw.get("since_version")), + limit=_int(kw.get("limit") or 200, 200), + ) + res["ok"] = True + res["contract"] = "pbl_entity_state_get" + return res + except Exception as exc: # noqa: BLE001 + log.exception("pbl_entity_state_get 失败") + return _error(exc) + + +async def pbl_runtime_broadcast_stats(**kw): + """广播观测：累计发布/投递/失败/最大耗时与 SLA 判定（监控与测试断言用）。""" + try: + resolve_tenant(kw.get("tenant_id")) + except PblRtError as exc: + return _error(exc) + st = broadcast_stats() + st["contract"] = "pbl_runtime_broadcast_stats" + st["sla"] = {"adjudicate_ms": 200, "broadcast_ms": 300, "e2e_ms": 1000, + "poll_fallback_interval_ms": 3000} + st["ok"] = True + return st + + +def contracts(): + """契约注册表（供 init.py 注册到 ServerEnv / 路由）。""" + return { + "pbl_runtime_event_append": pbl_runtime_event_append, + "pbl_runtime_event_poll": pbl_runtime_event_poll, + "pbl_runtime_broadcast_pull": pbl_runtime_broadcast_pull, + "pbl_entity_state_get": pbl_entity_state_get, + "pbl_entity_state_apply": pbl_entity_state_apply, + "pbl_world_broadcast": pbl_world_broadcast, + "pbl_runtime_broadcast_stats": pbl_runtime_broadcast_stats, + } diff --git a/pbl_runtime_ext/m11b_config.py b/pbl_runtime_ext/m11b_config.py new file mode 100644 index 0000000..7519482 --- /dev/null +++ b/pbl_runtime_ext/m11b_config.py @@ -0,0 +1,65 @@ +"""M11b 配置常量：单事务事件+状态写入、按月分区、提交后广播与时延承诺。 + +所有阈值集中在此，禁止在业务代码里散落魔法数字。 +时延承诺（需求 R-RT-03）： + - 裁决（adjudicate） <= 200 ms + - 广播（broadcast） <= 300 ms + - 端到端（e2e） <= 1000 ms + - 客户端轮询兜底间隔 = 3000 ms +""" + +MODULE_NAME = "pbl_runtime_ext" + +# ---------------------------------------------------------------- 表名 +T_EVENT = "pbl_runtime_event" +T_STATE = "pbl_entity_state" +T_SNAPSHOT = "pbl_world_state_snapshot" + +# ---------------------------------------------------------------- 分区 +PARTITION_SUFFIX = "pbl_runtime_event_" # + YYYYMM +PARTITION_RETAIN_MONTHS = 14 # 保留 12 个月 + 2 个月缓冲 +PARTITION_PRECREATE_MONTHS = 2 # 预建未来分区数 +PARTITION_LATEST_NAME = "pbl_runtime_event_pmax" # MAXVALUE 兜底分区 + +# ---------------------------------------------------------------- 时延预算(ms) +SLA_ADJUDICATE_MS = 200 +SLA_BROADCAST_MS = 300 +SLA_E2E_MS = 1000 +POLL_FALLBACK_INTERVAL_MS = 3000 # 广播丢失时客户端兜底轮询周期 +BROADCAST_THREADS = 2 # 提交后广播后台线程数 +BROADCAST_QUEUE_SIZE = 2000 # 每 world 环形缓冲容量 +POLL_MAX_LIMIT = 500 +POLL_DEFAULT_LIMIT = 100 + +# ---------------------------------------------------------------- 事务 +TX_MAX_RETRY = 3 # 乐观锁冲突重试次数 +TX_RETRY_BACKOFF_MS = 15 + +# ---------------------------------------------------------------- 错误码 +ERR_TENANT_MISSING = "PBL_RT_TENANT_MISSING" +ERR_BAD_PAYLOAD = "PBL_RT_BAD_PAYLOAD" +ERR_APPEND_ONLY = "PBL_RT_APPEND_ONLY" +ERR_STATE_CONFLICT = "PBL_RT_STATE_CONFLICT" +ERR_TX_FAILED = "PBL_RT_TX_FAILED" +ERR_SLA_BREACH = "PBL_RT_SLA_BREACH" +ERR_DB = "PBL_RT_DB" + +# 与 pbl_common 错误码风格对齐的别名（契约层引用） +PBL_RT_TENANT_MISSING_CODE = ERR_TENANT_MISSING + +# ---------------------------------------------------------------- 客户端禁写字段 +FORBIDDEN_CLIENT_FIELDS = ( + "state_version", + "server_version", + "seq_no", + "event_id", + "created_at", + "tenant_id", + "partition_key", +) + +# 事件必填字段 +EVENT_REQUIRED = ("world_id", "event_type") + +# 状态变更必填字段 +STATE_REQUIRED = ("entity_id",) diff --git a/pbl_runtime_ext/partitions.py b/pbl_runtime_ext/partitions.py new file mode 100644 index 0000000..ba51851 --- /dev/null +++ b/pbl_runtime_ext/partitions.py @@ -0,0 +1,257 @@ +"""M11b pbl_runtime_event 按月 RANGE 分区管理 + append-only 防护。 + +为什么分区：运行时事件是 append-only 高频写入流（每会话每 tick 都可能产生事件）， +单表膨胀后索引深度与范围扫描都会劣化，直接冲击「裁决 ≤200ms」的承诺。按月 +RANGE 分区把写入锁定在当月分区，历史月分区可整块归档/清理 +（DROP PARTITION 是元数据操作，比 DELETE 快几个数量级，也不产生 undo/binlog 洪峰）。 + +关键约束（MariaDB/MySQL）： + * RANGE 分区键必须包含在主键与所有唯一键中 → 事件表主键须为 (id, created_at)， + uk_tenant_idem 须扩为 (tenant_id, idem_key, created_at)。M11a 建的是非分区表 + （PK 只有 id），所以本模块提供 partition_migration_ddl() 做一次性改造； + * 用 TO_DAYS(created_at) 做 RANGE 边界，避免 DATETIME 直接比较的时区歧义； + * 必须保留 MAXVALUE 兜底分区（pmax），否则写入落在未建月份会直接报错。 + 本模块在写入前调用 ensure_forward_partitions 预建当月 + 未来 N 月， + pmax 只做最后一道保险（不会因为建分区失败而拒绝写事件）。 + +PostgreSQL 声明式分区：主表 PARTITION BY RANGE (created_at)，子表 +FOR VALUES FROM (start) TO (end)，索引建在父表上自动下发。 + +本模块所有 DDL 都幂等（先查 information_schema 再执行），且**绝不 DROP +不匹配 pbl_runtime_event_YYYYMM 命名的分区**（防误删业务表）。 +""" +import datetime +import logging +import re + +from .m11b_config import ( + PARTITION_LATEST_NAME, + PARTITION_PRECREATE_MONTHS, + PARTITION_RETAIN_MONTHS, + PARTITION_SUFFIX, + T_EVENT, +) + +log = logging.getLogger("pbl_runtime_ext.partitions") + +_MONTH_RE = re.compile(r"^%s(\d{4})(\d{2})$" % re.escape(PARTITION_SUFFIX)) + + +# ------------------------------------------------------------------ 月边界 +def month_start(dt=None): + dt = dt or datetime.date.today() + return datetime.date(dt.year, dt.month, 1) + + +def month_end(dt=None): + """下月 1 日（作为 EXCLUSIVE 上界）。""" + return add_months(month_start(dt), 1) + + +def add_months(d, n): + total = (d.year * 12 + (d.month - 1)) + n + year, month = divmod(total, 12) + return datetime.date(year, month + 1, 1) + + +def partition_name(dt): + return "%s%04d%02d" % (PARTITION_SUFFIX, dt.year, dt.month) + + +def partition_for(dt=None): + """给定日期 → (分区名, 起始日, 结束日)。""" + s = month_start(dt) + return partition_name(s), s, month_end(s) + + +def to_days(d): + """MySQL TO_DAYS 等价算法（公历 1582-10-15 之后有效，业务日期必在此后）。""" + a = (14 - d.month) // 12 + y = d.year + 4800 - a + m = d.month + 12 * a - 3 + jdn = d.day + (153 * m + 2) // 5 + 365 * y + y // 4 - y // 100 + y // 400 - 32045 + return jdn - 3652425 + + +def months_window(months=PARTITION_PRECREATE_MONTHS, base=None): + """[当月, +1 月, ... +months] 的月首日列表。""" + base = month_start(base) + return [add_months(base, i) for i in range(0, months + 1)] + + +def is_event_partition(name): + return bool(_MONTH_RE.match(name or "")) + + +# ------------------------------------------------------------------ DDL 生成 +def ddl_create_partition(dt, engine="mariadb"): + """单个月分区 DDL（幂等由调用方查表保证）。""" + name, start, end = partition_for(dt) + if "postgres" in str(engine).lower(): + return ('CREATE TABLE IF NOT EXISTS "%s" PARTITION OF "%s" ' + 'FOR VALUES FROM (\'%s\') TO (\'%s\')' + % (name, T_EVENT, start.isoformat(), end.isoformat())) + return ("ALTER TABLE `%s` ADD PARTITION (PARTITION `%s` VALUES LESS THAN (TO_DAYS('%s')))" + % (T_EVENT, name, end.isoformat())) + + +def ddl_drop_partition(name, engine="mariadb"): + if not is_event_partition(name): + raise ValueError("拒绝删除非事件月分区: %s" % name) + if "postgres" in str(engine).lower(): + return 'DROP TABLE IF EXISTS "%s"' % name + return "ALTER TABLE `%s` DROP PARTITION `%s`" % (T_EVENT, name) + + +def ddl_base_partitioned(engine="mariadb"): + """建表期分区子句（新建环境直接用，供 sql/ 脚本与建表函数复用）。""" + if "postgres" in str(engine).lower(): + return 'PARTITION BY RANGE ("created_at")' + return ("PARTITION BY RANGE (TO_DAYS(`created_at`)) (\n" + " PARTITION `%s` VALUES LESS THAN (TO_DAYS('1970-01-02')),\n" + " PARTITION `%s` VALUES LESS THAN MAXVALUE\n)" + % (PARTITION_SUFFIX + "197001", PARTITION_LATEST_NAME)) + + +def partition_migration_ddl(months=PARTITION_PRECREATE_MONTHS, engine="mariadb"): + """把 M11a 的非分区事件表改造成按月 RANGE 分区（一次性，幂等由 DBA 保证）。 + + MariaDB 要求分区键进入所有唯一键，因此先改主键/唯一键，再 PARTITION BY， + 最后逐月 ADD PARTITION。语句顺序不可调换。 + """ + stmts = [] + if "postgres" in str(engine).lower(): + stmts.append('-- PG: 需重建为 PARTITION BY RANGE ("created_at") 声明式分区表') + for dt in months_window(months): + stmts.append(ddl_create_partition(dt, engine)) + return stmts + stmts.append("ALTER TABLE `%s` DROP PRIMARY KEY, " + "ADD PRIMARY KEY (`id`, `created_at`)" % T_EVENT) + stmts.append("ALTER TABLE `%s` DROP INDEX `uk_tenant_idem`, " + "ADD UNIQUE KEY `uk_tenant_idem` (`tenant_id`, `idem_key`, `created_at`)" % T_EVENT) + parts = ",\n".join( + [" PARTITION `%s` VALUES LESS THAN (TO_DAYS('%s'))" + % (partition_name(dt), month_end(dt).isoformat()) for dt in months_window(months)] + + [" PARTITION `%s` VALUES LESS THAN MAXVALUE" % PARTITION_LATEST_NAME]) + stmts.append("ALTER TABLE `%s` PARTITION BY RANGE (TO_DAYS(`created_at`)) (\n%s\n)" + % (T_EVENT, parts)) + return stmts + + +# ------------------------------------------------------------------ 运行期维护 +def existing_partitions(): + """事件表已存在的分区名列表；表未分区/不支持时返回 []。""" + from .rtx_db import PblRtError, dbname, is_postgres, q_all + + try: + if is_postgres(): + rows = q_all('SELECT c.relname AS name FROM pg_inherits i ' + 'JOIN pg_class c ON c.oid = i.inhrelid ' + 'JOIN pg_class p ON p.oid = i.inhparent ' + 'WHERE p.relname = ${t}$', {'t': T_EVENT}) + else: + rows = q_all('SELECT PARTITION_NAME AS name FROM information_schema.PARTITIONS ' + 'WHERE TABLE_SCHEMA = ${s}$ AND TABLE_NAME = ${t}$ ' + 'AND PARTITION_NAME IS NOT NULL', {'s': dbname(), 't': T_EVENT}) + except PblRtError as exc: + log.warning("查询分区列表失败（按未分区处理）: %s", exc) + return [] + except Exception as exc: # noqa: BLE001 + log.warning("查询分区列表异常（按未分区处理）: %s", exc) + return [] + out = [] + for r in rows or []: + nm = r.get("name") or r.get("NAME") or r.get("partition_name") or r.get("Name") + if nm: + out.append(str(nm)) + return out + + +def _default_executor(): + from .rtx_db import sql_exec + + return lambda sql: sql_exec(sql) + + +def ensure_forward_partitions(months=PARTITION_PRECREATE_MONTHS, engine=None, executor=None): + """预建当月 + 未来 N 个月分区。幂等；DDL 失败不抛（写事件仍可落 pmax 兜底）。 + + 返回 {"ok", "created", "skipped", "partitions", "errors"}。 + **绝不允许因为建分区失败而拒绝写事件** —— 事件落库是事实源，分区只是性能优化。 + """ + from .rtx_db import engine as detect_engine + + engine = (engine or detect_engine()).lower() + have = set(existing_partitions()) + created, skipped, errors = [], [], [] + exec_fn = executor or _default_executor() + for dt in months_window(months): + name = partition_name(dt) + if name in have: + skipped.append(name) + continue + sql = ddl_create_partition(dt, engine) + try: + if exec_fn is None: + errors.append({"partition": name, "reason": "no_executor", "ddl": sql}) + continue + exec_fn(sql) + created.append(name) + have.add(name) + except Exception as exc: # noqa: BLE001 + errors.append({"partition": name, "reason": str(exc)[:200], "ddl": sql}) + log.warning("预建分区 %s 失败: %s", name, exc) + return {"ok": not errors, "created": created, "skipped": skipped, + "partitions": sorted(have), "errors": errors} + + +def drop_expired_partitions(retain=PARTITION_RETAIN_MONTHS, engine=None, dry_run=True): + """清理超出保留期的月分区（默认 dry_run，显式 dry_run=False 才真删）。 + + 安全：只删匹配 pbl_runtime_event_YYYYMM 的分区；pmax 与非规范命名永不删。 + """ + from .rtx_db import engine as detect_engine + + engine = (engine or detect_engine()).lower() + cutoff = add_months(month_start(), -int(retain)) + doomed, kept = [], [] + for name in existing_partitions(): + m = _MONTH_RE.match(name) + if not m: + kept.append(name) + continue + try: + bound = datetime.date(int(m.group(1)), int(m.group(2)), 1) + except ValueError: + kept.append(name) + continue + (doomed if bound < cutoff else kept).append(name) + dropped = [] + if not dry_run and doomed: + exec_fn = _default_executor() + for name in doomed: + if exec_fn is None: + break + try: + exec_fn(ddl_drop_partition(name, engine)) + dropped.append(name) + except Exception as exc: # noqa: BLE001 + log.warning("删除过期分区 %s 失败: %s", name, exc) + return {"dry_run": dry_run, "cutoff_month": cutoff.isoformat(), + "expired": doomed, "dropped": dropped, "retained": kept} + + +def partition_status(dt=None): + """给定日期所在分区是否就绪（写入前置检查，供契约接口返回）。""" + name, start, end = partition_for(dt) + have = set(existing_partitions()) + return {"table": T_EVENT, "partition": name, "exists": name in have, + "from": start.isoformat(), "to": end.isoformat(), + "fallback_partition": PARTITION_LATEST_NAME, + "partitioned": bool(have), + "partitions": sorted(have)} + + +def partition_maintenance_sql(months=PARTITION_PRECREATE_MONTHS, engine="mariadb"): + """输出维护 SQL 清单（供 DBA 脚本/部署文档使用，不执行）。""" + return [ddl_create_partition(dt, engine) for dt in months_window(months)] diff --git a/pbl_runtime_ext/rtx_db.py b/pbl_runtime_ext/rtx_db.py new file mode 100644 index 0000000..6c48c58 --- /dev/null +++ b/pbl_runtime_ext/rtx_db.py @@ -0,0 +1,348 @@ +"""M11b 数据库适配层：dbname 解析、引擎探测、语句执行、事务上下文。 + +铁律： + * 库名一律 ServerEnv().get_module_dbname('pbl_runtime_ext')，禁止硬编码 DBNAME； + * 查询走 pbl_common.api 的 q_all/q_one（已适配），本层只做写/事务与兜底； + * 任何缺少 tenant_id 的读写立即 fail-closed（抛 PblRtError，不落库）。 +""" +import logging +import threading + +from .m11b_config import ( + ERR_DB, + ERR_TENANT_MISSING, + MODULE_NAME, +) + +log = logging.getLogger("pbl_runtime_ext.db") + +_local = threading.local() + + +class PblRtError(Exception): + """运行时扩展业务异常：带 code，供 .dspy 层转成结构化错误响应。""" + + def __init__(self, code, message, detail=None): + super().__init__(message) + self.code = code + self.message = message + self.detail = detail or {} + + def to_dict(self): + return {"code": self.code, "message": self.message, "detail": self.detail} + + +# ------------------------------------------------------------------ dbname +def dbname(): + try: + from apppublic import ServerEnv # 运行期注入，导入失败再走兜底 + + name = ServerEnv().get_module_dbname(MODULE_NAME) + if name: + return name + except Exception as exc: # pragma: no cover - 环境缺 apppublic 时兜底 + log.debug("ServerEnv().get_module_dbname 不可用: %s", exc) + try: + from pbl_common import api as pbl_api + + getter = getattr(pbl_api, "module_dbname", None) + if callable(getter): + name = getter(MODULE_NAME) + if name: + return name + except Exception as exc: # pragma: no cover + log.debug("pbl_common.api.module_dbname 不可用: %s", exc) + raise PblRtError(ERR_DB, "无法解析模块库名（dbname），fail-closed") + + +def engine(): + """探测 SQL 方言：mariadb / postgresql。默认 mariadb（与 pbls 应用一致）。""" + try: + from apppublic import ServerEnv + + cfg = ServerEnv() + for attr in ("get_db_engine", "db_engine"): + val = getattr(cfg, attr, None) + if callable(val): + val = val() + if isinstance(val, str) and val: + return val.lower() + except Exception: # pragma: no cover + pass + try: + from pbl_common import api as pbl_api + + eng = getattr(pbl_api, "DB_ENGINE", None) + if isinstance(eng, str) and eng: + return eng.lower() + except Exception: # pragma: no cover + pass + return "mariadb" + + +def is_postgres(): + return "postgres" in engine() + + +# ------------------------------------------------------------------ tenant +def tenant_id(explicit=None): + """取租户：显式优先，其次 pbl_common.api.tenant_id()；缺失 fail-closed。""" + tid = explicit + if not tid: + try: + from pbl_common import api as pbl_api + + getter = getattr(pbl_api, "tenant_id", None) + if callable(getter): + tid = getter() + except Exception as exc: # pragma: no cover + log.debug("pbl_common.api.tenant_id 不可用: %s", exc) + if not tid: + raise PblRtError( + ERR_TENANT_MISSING, + "tenant_id 缺失：所有运行时事件/状态读写必须带租户，fail-closed", + ) + return str(tid) + + +# ------------------------------------------------------------------ 执行 +def _resolve_executors(): + """按优先级收集可用的 SQL 执行器（返回 callable(sql, params, db)）。""" + cands = [] + + def wrap(fn, style): + def _call(sql, params, db): + if style == "positional": + return fn(sql, params, db) + if style == "kw": + return fn(sql, params=params, dbname=db) + return fn(sql, params) + + return _call + + try: + from pbl_common import api as pbl_api + + for name, style in ( + ("sql_exec", "kw"), + ("execute_sql", "kw"), + ("q_exec", "positional"), + ): + fn = getattr(pbl_api, name, None) + if callable(fn): + cands.append(wrap(fn, style)) + except Exception: # pragma: no cover + pass + + try: + import sqlor + + if callable(getattr(sqlor, "sqlExe", None)): + cands.append(wrap(sqlor.sqlExe, "kw")) + except Exception: # pragma: no cover + pass + + try: + import sqlor + + sor = getattr(sqlor, "sor", None) + if sor is not None and callable(getattr(sor, "sqlExe", None)): + cands.append(wrap(sor.sqlExe, "kw")) + except Exception: # pragma: no cover + pass + + return cands + + +def sql_exec(sql, params=None): + """执行一条语句。返回受影响行数（拿不到时返回 None）。失败 fail-closed。""" + db = dbname() + executors = getattr(_local, "executors", None) + if executors is None: + executors = _resolve_executors() + _local.executors = executors + if not executors: + raise PblRtError( + ERR_DB, + "无可用 SQL 执行器（pbl_common.api / sqlor.sqlExe 均未加载），fail-closed", + ) + last_err = None + for call in executors: + try: + res = call(sql, params or {}, db) + if isinstance(res, bool): + return None + if isinstance(res, int): + return res + if res is None: + return None + return getattr(res, "rowcount", None) + except Exception as exc: + last_err = exc + log.debug("sql executor 尝试失败，换下一个: %s", exc) + raise PblRtError(ERR_DB, "SQL 执行失败: %s" % last_err) + + +def q_all(sql, params=None): + """查询多行：优先复用 pbl_common.api.q_all（已适配驱动差异）。""" + db = dbname() + try: + from pbl_common import api as pbl_api + + fn = getattr(pbl_api, "q_all", None) + if callable(fn): + try: + return fn(sql, params or {}, db) + except TypeError: + return fn(sql, params or {}) + except Exception as exc: # pragma: no cover + log.debug("pbl_common.api.q_all 不可用，走兜底: %s", exc) + return _fallback_rows(sql, params) + + +def q_one(sql, params=None): + rows = q_all(sql, params) or [] + return rows[0] if rows else None + + +def _fallback_rows(sql, params): + """兜底行查询（仅在 pbl_common 未提供 q_all 时使用）。""" + db = dbname() + try: + import sqlor + + fn = getattr(sqlor, "R", None) or getattr(sqlor, "I", None) + if callable(fn): + res = fn(sql, params or {}, db) + return list(res or []) + except Exception as exc: # pragma: no cover + raise PblRtError(ERR_DB, "行查询失败: %s" % exc) + raise PblRtError(ERR_DB, "无可用行查询执行器，fail-closed") + + +# ------------------------------------------------------------------ 事务 +def in_tx(): + return getattr(_local, "tx_depth", 0) > 0 + + +class transaction(object): + """单事务上下文管理器。 + + 用法:: + + with transaction() as tx: + tx.exec("INSERT ...", {...}) + tx.exec("UPDATE ...", {...}) + # 正常退出 COMMIT，异常退出 ROLLBACK 并向上抛 + + 嵌套调用不产生新事务（savepoint 语义由上层避免），保证 + 「事件 + 实体状态」始终落在同一个事务里。 + """ + + def __init__(self, label="runtime_tx"): + self.label = label + self.db = None + self._conn = None + self._outer = False + + def __enter__(self): + if in_tx(): + self._outer = True + _local.tx_depth += 1 + return self + self._outer = False + _local.tx_depth = getattr(_local, "tx_depth", 0) + 1 + self.db = dbname() + self._conn = _open_conn(self.db) + if self._conn is not None: + self._conn.begin() + return self + + def exec(self, sql, params=None): + if self._conn is not None: + cur = self._conn.cursor() + try: + cur.execute(sql, params or {}) + affected = cur.rowcount + finally: + try: + cur.close() + except Exception: + pass + return affected + return sql_exec(sql, params) + + def query(self, sql, params=None): + if self._conn is not None: + cur = self._conn.cursor() + try: + cur.execute(sql, params or {}) + cols = [d[0] for d in (cur.description or [])] + return [dict(zip(cols, row)) for row in (cur.fetchall() or [])] + finally: + try: + cur.close() + except Exception: + pass + return q_all(sql, params) + + def commit(self): + if self._conn is not None: + self._conn.commit() + + def rollback(self): + if self._conn is not None: + try: + self._conn.rollback() + except Exception as exc: # pragma: no cover + log.warning("rollback 失败: %s", exc) + + def __exit__(self, exc_type, exc_val, exc_tb): + if self._outer: + _local.tx_depth = max(0, getattr(_local, "tx_depth", 1) - 1) + return False + try: + if exc_type is None: + self.commit() + else: + self.rollback() + finally: + if self._conn is not None: + try: + self._conn.close() + except Exception: + pass + _local.tx_depth = max(0, getattr(_local, "tx_depth", 1) - 1) + return False + + +def _open_conn(db): + """尝试拿一个可事务控制的连接；拿不到返回 None（退化为逐条 sql_exec）。""" + try: + from pbl_common import api as pbl_api + + for name in ("get_conn", "connection", "get_connection"): + fn = getattr(pbl_api, name, None) + if callable(fn): + try: + return fn(db) + except TypeError: + return fn() + except Exception as exc: # pragma: no cover + log.debug("pbl_common 连接获取不可用: %s", exc) + try: + import sqlor + + fn = getattr(sqlor, "get_conn", None) + if callable(fn): + return fn(db) + except Exception: # pragma: no cover + pass + return None + + +def tx_guard_sql(): + """返回当前方言的事务控制语句（供无连接对象时使用）。""" + if is_postgres(): + return ("BEGIN", "COMMIT", "ROLLBACK") + return ("START TRANSACTION", "COMMIT", "ROLLBACK") diff --git a/pbl_runtime_ext/tx_write.py b/pbl_runtime_ext/tx_write.py new file mode 100644 index 0000000..8823e58 --- /dev/null +++ b/pbl_runtime_ext/tx_write.py @@ -0,0 +1,557 @@ +"""M11b 核心：单事务写「事件(append-only) + 实体状态(+快照)」，提交后广播。 + +对外主函数 +---------- +``apply_runtime_event(...)`` —— 一次调用完成： + 1. fail-closed 校验（tenant_id 缺失 / 必填缺失 / 客户端试图写权威字段）； + 2. 幂等闸门（idem_key 命中 uk_tenant_idem → 直接返回既有事件，不重复写、不重复广播）； + 3. **单事务**：pbl_runtime_event INSERT + pbl_entity_state UPSERT(+快照) 一起提交， + 任一条失败整体回滚，返回/抛出前不留半条脏数据； + 4. 提交成功后才广播（broadcast.py::BroadcastHub），广播失败不撤销事实， + 客户端 3s 轮询（``poll_events`` 先查内存缓冲、未命中退回查库）补齐； + 5. 全程分段计时：裁决(含事务)/广播/端到端，与 SLA 预算对比后随响应返回。 + +append-only 由 ``assert_append_only`` 在代码层强制：本模块任何路径都不会 +UPDATE/DELETE pbl_runtime_event（事件是不可变事实，状态才是可变视图）。 +""" +import hashlib +import json +import logging +import re +import time +import uuid + +from .broadcast import get_hub +from .latency import Stopwatch, evaluate_sla +from .m11b_config import ( + ERR_APPEND_ONLY, + ERR_BAD_PAYLOAD, + ERR_STATE_CONFLICT, + ERR_TX_FAILED, + FORBIDDEN_CLIENT_FIELDS, + STATE_REQUIRED, + TX_MAX_RETRY, + TX_RETRY_BACKOFF_MS, +) +from .partitions import ensure_forward_partitions, partition_for +from .rtx_db import PblRtError, transaction + +log = logging.getLogger("pbl_runtime_ext.tx_write") + +T_EVENT_TABLE = "pbl_runtime_event" +T_STATE_TABLE = "pbl_entity_state" +T_SNAPSHOT_TABLE = "pbl_world_state_snapshot" + +# append-only 守卫：禁止的语句形态（作用于事件表） +_MUTATION_RE = re.compile( + r"^\s*(UPDATE|DELETE|REPLACE|TRUNCATE|ALTER|DROP)\b", re.IGNORECASE) +_EVENT_TABLE_RE = re.compile(r"\bpbl_runtime_event\b", re.IGNORECASE) + +_partition_cache = {"ts": 0.0, "result": None} +_PARTITION_TTL_SECONDS = 3600 * 6 + + +# ------------------------------------------------------------------ 工具 +def _dumps(obj): + if obj is None: + return "{}" + if isinstance(obj, str): + return obj + try: + return json.dumps(obj, ensure_ascii=False, default=str, sort_keys=True) + except (TypeError, ValueError): + return json.dumps(str(obj)) + + +def _loads(v, default=None): + if isinstance(v, (dict, list)): + return v + if not v: + return default if default is not None else {} + try: + return json.loads(v) + except (ValueError, TypeError): + return default if default is not None else {} + + +def _now(): + return time.strftime("%Y-%m-%d %H:%M:%S") + + +def _checksum(text): + return hashlib.sha256(text.encode("utf-8")).hexdigest()[:32] + + +def derive_idem_key(tenant_id, session_id, event_type, payload, client_seq=None): + """无显式 idem_key 时按业务要素派生（同输入同键 → 天然幂等）。""" + raw = "%s|%s|%s|%s|%s" % (tenant_id, session_id, event_type, + _dumps(payload or {}), client_seq if client_seq is not None else "") + return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:64] + + +def assert_append_only(sql): + """append-only 守卫：任何改写 pbl_runtime_event 的语句一律拒绝执行。""" + if _EVENT_TABLE_RE.search(sql or "") and _MUTATION_RE.match(sql or ""): + raise PblRtError(ERR_APPEND_ONLY, + "pbl_runtime_event 是 append-only 事实流，禁止 UPDATE/DELETE/REPLACE", + {"sql": (sql or "")[:160]}) + return True + + +# ------------------------------------------------------------------ 校验 +def _validate(tenant_id, session_id, event_type, payload, state_updates, client_fields): + if not tenant_id: + raise PblErrorTenant("tenant_id 缺失：运行时事件/状态读写必须带租户（fail-closed）") + if session_id in (None, "", 0): + raise PblRtError(ERR_BAD_PAYLOAD, "缺少 session_id") + if not event_type: + raise PblRtError(ERR_BAD_PAYLOAD, "缺少 event_type") + submitted = list(client_fields.keys()) if isinstance(client_fields, dict) \ + else list(client_fields or []) + leaked = [f for f in submitted if f in FORBIDDEN_CLIENT_FIELDS] + if leaked: + raise PblRtError(ERR_BAD_PAYLOAD, + "服务端权威字段客户端禁写: %s" % ",".join(sorted(set(leaked))), + {"fields": sorted(set(leaked))}) + for su in state_updates or []: + for f in STATE_REQUIRED: + if su.get(f) in (None, ""): + raise PblRtError(ERR_BAD_PAYLOAD, "state_updates 缺少 %s" % f) + return True + + +class PblErrorTenant(PblRtError): + def __init__(self, message): + super(PblErrorTenant, self).__init__("PBL-TENANT-0001", message) + + +# ------------------------------------------------------------------ 分区 +def ensure_partition_ready(force=False): + """写入前确保当月+未来分区存在（进程内 TTL 缓存，失败不阻塞写入）。""" + now = time.time() + if not force and _partition_cache["result"] is not None and \ + (now - _partition_cache["ts"]) < _PARTITION_TTL_SECONDS: + return _partition_cache["result"] + try: + res = ensure_forward_partitions() + except Exception as exc: # noqa: BLE001 + res = {"ok": False, "created": [], "skipped": [], "errors": [str(exc)[:200]]} + _partition_cache["ts"] = now + _partition_cache["result"] = res + return res + + +# ------------------------------------------------------------------ 主流程 +def apply_runtime_event(tenant_id, session_id, event_type, payload=None, + world_id=0, entity_id=0, actor_id=None, source="runtime", + idem_key=None, client_seq=None, causation_id=None, + state_updates=None, snapshot=False, broadcast=True, + if_state_version=None, tx_group=None, client_fields=None): + """单事务写入 + 提交后广播。返回结构化结果（含 latency/SLA 实测）。 + + state_updates: [{"entity_id":..,"state_key":..,"state_value":{..}, + "world_id":..,"merge":True,"if_state_version":..}] + """ + sw = Stopwatch() + tid = str(tenant_id or "") + _validate(tid, session_id, event_type, payload, state_updates, client_fields) + payload_text = _dumps(payload or {}) + idem = idem_key or derive_idem_key(tid, session_id, event_type, payload, client_seq) + part_name, _start, _end = partition_for() + partition = ensure_partition_ready() + + existing = _find_by_idem(tid, idem) + if existing: + sw.mark("tx_ms") + latency, _ = evaluate_sla(sw, {"adjudicate_ms": sw.snapshot().get("tx_ms", 0)}) + return {"ok": True, "dedup": True, "event": existing, "event_id": existing.get("id"), + "event_code": existing.get("event_code"), "seq_no": existing.get("seq_no"), + "state_version": existing.get("state_version"), "idem_key": idem, + "append_only": True, "partition": part_name, "state": [], + "broadcast": {"skipped": True, "reason": "dedup"}, + "transaction": "single", "latency": latency, + "note": "幂等命中：未重复写入、未重复广播"} + + attempt, last_err = 0, None + while attempt < max(1, TX_MAX_RETRY): + attempt += 1 + try: + result = _run_tx(tid, session_id, event_type, payload_text, world_id, + entity_id, actor_id, source, idem, causation_id, + state_updates, snapshot, if_state_version, tx_group, + part_name) + break + except PblRtError as exc: + if exc.code == ERR_STATE_CONFLICT: + raise + last_err = exc + log.warning("单事务第 %d 次失败: %s", attempt, exc) + time.sleep(TX_RETRY_BACKOFF_MS * attempt / 1000.0) + else: + raise PblRtError(ERR_TX_FAILED, + "单事务写入失败已回滚（无脏数据）：%s" % (last_err or "unknown")) + + tx_ms = sw.mark("tx_ms") + result["transaction"] = "single" + result["attempts"] = attempt + result["partition"] = part_name + result["partition_maintenance"] = {"ok": partition.get("ok"), + "created": partition.get("created")} + + # ---- 提交后广播（绝不提交前广播） + bcast = {"skipped": True, "reason": "broadcast_disabled"} + if broadcast and result.get("event"): + hub = get_hub() + bcast = hub.publish("s:%s" % session_id, _broadcast_frame(result)) + bcast_ms = sw.mark("broadcast_ms") + bcast["cost_ms"] = bcast_ms + bcast["budget_ms"] = _bcast_budget() + bcast["fallback_poll_interval_ms"] = 3000 + result["broadcast"] = bcast + + adjudicate_ms = tx_ms + _adjudicate_overhead(sw) + result["adjudicate_ms"] = adjudicate_ms + latency, breached = evaluate_sla(sw, {"adjudicate_ms": adjudicate_ms, + "tx_ms": tx_ms, "broadcast_ms": bcast_ms}) + result["latency"] = latency + result["append_only"] = True + result["server_authoritative"] = True + result["within_sla"] = latency["within_sla"] + if breached: + result["sla_warning"] = breached + return result + + +def _bcast_budget(): + from .m11b_config import SLA_BROADCAST_MS + + return SLA_BROADCAST_MS + + +def _adjudicate_overhead(sw): + """裁决 = 校验 + 幂等查询 + 事务（事务外的前置开销也计入裁决预算）。""" + marks = sw.snapshot() + return max(0, sw.total_ms() - marks.get("broadcast_ms", 0)) + + +def _broadcast_frame(result): + ev = result.get("event") or {} + return {"seq_no": result.get("seq_no"), "event_id": result.get("event_id"), + "event_type": ev.get("event_type"), "session_id": ev.get("session_id"), + "world_id": ev.get("world_id"), "state_version": result.get("state_version"), + "states": result.get("state") or [], "ts": result.get("created_at") or _now()} + + +# ------------------------------------------------------------------ 事务体 +def _run_tx(tid, session_id, event_type, payload_text, world_id, entity_id, + actor_id, source, idem, causation_id, state_updates, snapshot, + if_state_version, tx_group, part_name): + """在**一个**事务里写事件 + 实体状态(+快照)。异常自动回滚。""" + now = _now() + event_code = "PBLRE%s" % uuid.uuid4().hex[:16] + with transaction("apply_runtime_event") as tx: + seq_no = _next_seq(tx, tid, session_id) + base_version = _max_state_version(tx, tid, session_id) + state_rows = _apply_states(tx, tid, session_id, world_id, entity_id, + state_updates, base_version, actor_id, now, + if_state_version) + nxt_version = max([base_version] + [r["state_version"] for r in state_rows]) \ + if state_rows else base_version + 1 + ev_params = { + "code": event_code, "t": tid, "idem": idem, "w": world_id or 0, + "s": session_id, "e": entity_id or 0, "type": event_type, + "payload": payload_text, "seq": seq_no, "src": source or "runtime", + "st": "applied", "txg": tx_group or "", "ver": nxt_version, + "by": actor_id or 0, "ts": now, + } + assert_append_only(INSERT_EVENT_SQL) + tx.exec(INSERT_EVENT_SQL, ev_params) + snap = None + if snapshot: + snap = _write_snapshot(tx, tid, session_id, world_id, seq_no, + state_rows, ev_params["by"], now) + event_row = { + "id": None, "event_code": event_code, "tenant_id": tid, + "session_id": session_id, "world_id": world_id or 0, + "entity_id": entity_id or 0, "event_type": event_type, + "payload": _loads(payload_text), "seq_no": seq_no, "source": source, + "state": "applied", "tx_group": tx_group or "", "idem_key": idem, + "causation_id": causation_id, "state_version": nxt_version, + "created_at": now, "partition": part_name, + } + event_row["id"] = _last_insert_id(tx) + return {"ok": True, "dedup": False, "event": event_row, + "event_id": event_row["id"], "event_code": event_code, + "seq_no": seq_no, "state_version": nxt_version, + "idem_key": idem, "causation_id": causation_id, + "state": state_rows, "snapshot": snap, "created_at": now} + + +INSERT_EVENT_SQL = ( + "INSERT INTO `pbl_runtime_event` " + "(`event_code`,`tenant_id`,`idem_key`,`world_id`,`session_id`,`entity_id`," + "`event_type`,`payload`,`seq_no`,`source`,`state`,`tx_group`,`state_version`," + "`created_by`,`created_at`) VALUES " + "(${code}$,${t}$,${idem}$,${w}$,${s}$,${e}$,${type}$,${payload}$,${seq}$," + "${src}$,${st}$,${txg}$,${ver}$,${by}$,${ts}$)" +) + +UPSERT_STATE_MARIADB = ( + "INSERT INTO `pbl_entity_state` " + "(`state_code`,`tenant_id`,`world_id`,`session_id`,`entity_id`,`state_key`," + "`state_value`,`state_version`,`last_event_id`,`created_at`,`updated_at`) VALUES " + "(${code}$,${t}$,${w}$,${s}$,${e}$,${k}$,${v}$,${ver}$,${ev}$,${ts}$,${ts}$) " + "ON DUPLICATE KEY UPDATE `state_value` = VALUES(`state_value`), " + "`state_version` = VALUES(`state_version`), `last_event_id` = VALUES(`last_event_id`), " + "`updated_at` = VALUES(`updated_at`)" +) + +UPSERT_STATE_PG = ( + 'INSERT INTO "pbl_entity_state" ' + '("state_code","tenant_id","world_id","session_id","entity_id","state_key",' + '"state_value","state_version","last_event_id","created_at","updated_at") VALUES ' + '(${code}$,${t}$,${w}$,${s}$,${e}$,${k}$,${v}$,${ver}$,${ev}$,${ts}$,${ts}$) ' + 'ON CONFLICT ("tenant_id","session_id","entity_id","state_key") DO UPDATE SET ' + '"state_value" = EXCLUDED."state_value", "state_version" = EXCLUDED."state_version", ' + '"last_event_id" = EXCLUDED."last_event_id", "updated_at" = EXCLUDED."updated_at"' +) + + +def _next_seq(tx, tid, session_id): + rows = tx.query("SELECT COALESCE(MAX(`seq_no`),0) AS s FROM `pbl_runtime_event` " + "WHERE `tenant_id` = ${t}$ AND `session_id` = ${s}$", + {"t": tid, "s": session_id}) + return int((rows[0].get("s") if rows else 0) or 0) + 1 + + +def _max_state_version(tx, tid, session_id): + rows = tx.query("SELECT COALESCE(MAX(`state_version`),0) AS v FROM `pbl_entity_state` " + "WHERE `tenant_id` = ${t}$ AND `session_id` = ${s}$", + {"t": tid, "s": session_id}) + return int((rows[0].get("v") if rows else 0) or 0) + + +def _last_insert_id(tx): + try: + rows = tx.query("SELECT LAST_INSERT_ID() AS i", {}) + val = rows[0].get("i") if rows else None + return int(val) if val is not None else None + except Exception: # noqa: BLE001 # PG/驱动不支持时返回 None，不影响事实已提交 + return None + + +def _apply_states(tx, tid, session_id, world_id, entity_id, state_updates, + base_version, actor_id, now, if_state_version): + """UPSERT 实体状态：state_version 由服务端单调分配，客户端传值一律忽略。""" + from .rtx_db import is_postgres + + updates = list(state_updates or []) + if not updates and entity_id: + updates = [{"entity_id": entity_id, "state_key": "default", + "state_value": {}, "merge": False}] + sql = UPSERT_STATE_PG if is_postgres() else UPSERT_STATE_MARIADB + out, ver = [], base_version + for su in updates: + eid = su.get("entity_id") or entity_id + if eid in (None, ""): + raise PblRtError(ERR_BAD_PAYLOAD, "state_updates 缺少 entity_id") + key = su.get("state_key") or "default" + cur = _read_state(tx, tid, session_id, eid, key) + cur_ver = int((cur or {}).get("state_version") or 0) + expect = su.get("if_state_version", if_state_version) + if expect not in (None, "") and int(expect) != cur_ver: + raise PblRtError(ERR_STATE_CONFLICT, + "乐观锁失败：期望 state_version=%s 实际=%s（服务端权威）" + % (expect, cur_ver), + {"entity_id": eid, "state_key": key, + "expected": int(expect), "actual": cur_ver}) + value = su.get("state_value", su.get("state_json", {})) + if su.get("merge", True) and cur: + value = dict(_loads(cur.get("state_value"), {}), **_loads(value, {})) + ver = max(ver, cur_ver) + 1 + vtext = _dumps(value) + tx.exec(sql, {"code": "PBLES%s" % uuid.uuid4().hex[:14], "t": tid, + "w": su.get("world_id", world_id) or 0, "s": session_id, + "e": eid, "k": key, "v": vtext, "ver": ver, + "ev": su.get("last_event_id") or 0, "ts": now}) + out.append({"entity_id": eid, "state_key": key, "state_version": ver, + "checksum": _checksum(vtext), "prev_version": cur_ver, + "updated_by": actor_id or 0}) + return out + + +def _read_state(tx, tid, session_id, entity_id, state_key): + rows = tx.query("SELECT `state_value`,`state_version` FROM `pbl_entity_state` " + "WHERE `tenant_id` = ${t}$ AND `session_id` = ${s}$ " + "AND `entity_id` = ${e}$ AND `state_key` = ${k}$ LIMIT 1", + {"t": tid, "s": session_id, "e": entity_id, "k": state_key}) + return rows[0] if rows else None + + +def _write_snapshot(tx, tid, session_id, world_id, seq_no, state_rows, by, now): + body = _dumps({"states": state_rows, "seq_no": seq_no}) + code = "PBLSN%s%06d" % (uuid.uuid4().hex[:10], int(seq_no or 0)) + tx.exec("INSERT INTO `pbl_world_state_snapshot` " + "(`snapshot_code`,`tenant_id`,`world_id`,`session_id`,`seq_no`,`state`," + "`entity_count`,`checksum`,`created_by`,`created_at`) VALUES " + "(${code}$,${t}$,${w}$,${s}$,${q}$,${st}$,${n}$,${ck}$,${by}$,${ts}$)", + {"code": code, "t": tid, "w": world_id or 0, "s": session_id, "q": seq_no, + "st": body, "n": len(state_rows), "ck": _checksum(body), "by": by, "ts": now}) + return {"snapshot_code": code, "seq_no": seq_no, "entity_count": len(state_rows), + "checksum": _checksum(body)} + + +def _find_by_idem(tid, idem): + from .rtx_db import q_one + + try: + row = q_one("SELECT `id`,`event_code`,`session_id`,`world_id`,`entity_id`," + "`event_type`,`payload`,`seq_no`,`state_version`,`state`,`created_at` " + "FROM `pbl_runtime_event` WHERE `tenant_id` = ${t}$ AND `idem_key` = ${i}$ " + "LIMIT 1", {"t": tid, "i": idem}) + except Exception as exc: # noqa: BLE001 表不存在等：按未命中处理，交给事务写入报错 + log.debug("幂等查询跳过: %s", exc) + return None + if not row: + return None + row = dict(row) + row["payload"] = _loads(row.get("payload")) + row["dedup"] = True + return row + + +# ------------------------------------------------------------------ 读侧 +def poll_events(tenant_id, session_id, since_seq=0, limit=100, try_buffer=True): + """增量拉取：先命中广播环形缓冲（毫秒级），未命中退回查库；3s 兜底周期。""" + tid = str(tenant_id or "") + if not tid: + raise PblErrorTenant("tenant_id 缺失：轮询必须带租户（fail-closed）") + since = int(since_seq or 0) + lim = max(1, min(int(limit or 100), 500)) + if try_buffer: + buf = get_hub().pull("s:%s" % session_id, since, lim) + if buf.get("events") or (buf.get("buffered") and not buf.get("need_poll")): + buf["source"] = "broadcast_buffer" + buf["session_id"] = session_id + return buf + from .rtx_db import q_all + + rows = q_all("SELECT `id`,`event_code`,`event_type`,`entity_id`,`world_id`," + "`payload`,`seq_no`,`state_version`,`source`,`state`,`created_at` " + "FROM `pbl_runtime_event` WHERE `tenant_id` = ${t}$ " + "AND `session_id` = ${s}$ AND `seq_no` > ${q}$ " + "ORDER BY `seq_no` ASC LIMIT ${n}$", + {"t": tid, "s": session_id, "q": since, "n": lim}) or [] + for r in rows: + r["payload"] = _loads(r.get("payload")) + return {"events": rows, "count": len(rows), + "last_seq": int(rows[-1]["seq_no"]) if rows else since, + "source": "db", "buffered": False, "need_poll": False, + "session_id": session_id, "poll_interval_ms": 3000, + "append_only": True} + + +def read_states(tenant_id, session_id, entity_id=None, since_version=0, limit=200): + tid = str(tenant_id or "") + if not tid: + raise PblErrorTenant("tenant_id 缺失：状态读取必须带租户（fail-closed）") + from .rtx_db import q_all + + sql = ("SELECT `entity_id`,`state_key`,`state_value`,`state_version`," + "`last_event_id`,`updated_at` FROM `pbl_entity_state` " + "WHERE `tenant_id` = ${t}$ AND `session_id` = ${s}$") + params = {"t": tid, "s": session_id} + if entity_id not in (None, ""): + sql += " AND `entity_id` = ${e}$" + params["e"] = entity_id + if int(since_version or 0): + sql += " AND `state_version` > ${v}$" + params["v"] = int(since_version) + params["n"] = max(1, min(int(limit or 200), 500)) + rows = q_all(sql + " ORDER BY `state_version` ASC LIMIT ${n}$", params) or [] + for r in rows: + r["state_value"] = _loads(r.get("state_value")) + return {"data": rows, "count": len(rows), "server_authoritative": True, + "forbidden_client_fields": list(FORBIDDEN_CLIENT_FIELDS)} + + +def broadcast_stats(): + return get_hub().stats() + + +def self_check(): + """M11b 契约自检（离线，不连库）：append-only 守卫 + 月分区算法 + 幂等键派生。""" + msgs, ok = [], True + upd_event = "UPDATE " + "`pbl_runtime_event`" + " SET `state` = 'x' WHERE 1 = 1" + try: + assert_append_only(upd_event) + ok = False + msgs.append("自检[append-only 守卫] FAIL：UPDATE 未被拒绝") + except PblRtError as exc: + if exc.code == ERR_APPEND_ONLY: + msgs.append("自检[append-only 守卫] PASS UPDATE 被拒绝") + else: + ok = False + msgs.append("自检[append-only 守卫] FAIL code=%s" % exc.code) + try: + assert_append_only("SELECT * FROM `pbl_runtime_event` WHERE `seq_no` > 0") + assert_append_only("INSERT INTO `pbl_runtime_event` (`a`) VALUES (1)") + assert_append_only("UPDATE `pbl_entity_state` SET `state_version` = 1") + msgs.append("自检[append-only 不误伤] PASS SELECT/INSERT/状态 UPDATE 放行") + except PblRtError as exc: + ok = False + msgs.append("自检[append-only 不误伤] FAIL：%s" % exc.message) + + name, start, end = partition_for(datetime_date(2026, 3, 17)) + if (name, start.isoformat(), end.isoformat()) != ("pbl_runtime_event_202603", + "2026-03-01", "2026-04-01"): + ok = False + msgs.append("自检[月分区边界] FAIL %s/%s/%s" % (name, start, end)) + else: + msgs.append("自检[月分区边界] PASS %s [%s,%s)" % (name, start, end)) + y12, s12, e12 = partition_for(datetime_date(2026, 12, 31)) + if (y12, e12.isoformat()) != ("pbl_runtime_event_202612", "2027-01-01"): + ok = False + msgs.append("自检[跨年分区] FAIL %s→%s" % (y12, e12)) + else: + msgs.append("自检[跨年分区] PASS 202612 上界 2027-01-01") + + k1 = derive_idem_key("t", 1, "move", {"x": 1}) + k2 = derive_idem_key("t", 1, "move", {"x": 1}) + k3 = derive_idem_key("t", 1, "move", {"x": 2}) + if k1 != k2 or k1 == k3 or len(k1) != 64: + ok = False + msgs.append("自检[幂等键派生] FAIL") + else: + msgs.append("自检[幂等键派生] PASS 同输入同键/异输入异键") + + if "INSERT INTO `pbl_runtime_event`" not in INSERT_EVENT_SQL: + ok = False + msgs.append("自检[事件写入语句] FAIL：不是 INSERT") + else: + msgs.append("自检[事件写入语句] PASS 事件表只有 INSERT（append-only）") + if ("UPDATE " + "`pbl_runtime_event`") in s_all() or \ + ("DELETE FROM " + "`pbl_runtime_event`") in s_all(): + ok = False + msgs.append("自检[全模块扫描] FAIL：出现事件表 UPDATE") + else: + msgs.append("自检[全模块扫描] PASS 模块内无事件表 UPDATE/DELETE") + return ok, msgs + + +def s_all(): + """本模块源码全文（自检用：确认不存在改写事件表的语句）。""" + import os + + try: + with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), + "tx_write.py"), encoding="utf-8") as fh: + return fh.read() + except Exception: # noqa: BLE001 + return "" + + +def datetime_date(y, m, d): + import datetime as _dt + + return _dt.date(y, m, d) diff --git a/scripts/load_path.py b/scripts/load_path.py index dd505d6..b167547 100644 --- a/scripts/load_path.py +++ b/scripts/load_path.py @@ -21,6 +21,8 @@ PATHS = [ ('/pbl_runtime_ext/api/pbl_entity_state_apply.dspy', 'logined'), ('/pbl_runtime_ext/api/pbl_world_broadcast.dspy', 'logined'), ('/pbl_runtime_ext/api/pbl_session_member_add.dspy', 'logined'), + ('/pbl_runtime_ext/api/pbl_runtime_broadcast_pull.dspy', 'logined'), + ('/pbl_runtime_ext/api/pbl_runtime_broadcast_stats.dspy', 'logined'), ] diff --git a/scripts/m11b_selftest.py b/scripts/m11b_selftest.py new file mode 100644 index 0000000..b439726 --- /dev/null +++ b/scripts/m11b_selftest.py @@ -0,0 +1,339 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M11b 离线自检脚本：单事务事件+状态写入、幂等、回滚、提交后广播、3s 轮询兜底。 + +不连真实数据库（用内存 SQL 替身），验证的是**代码行为**而非环境： + P1 单事务写入：事件 1 行 + 实体状态 1 行，提交后可见 + P2 幂等：同 idem_key 二次调用不重复写、不重复广播（dedup=True） + P3 回滚：状态写入抛错 → 事件行不落库（无脏数据） + P4 提交后广播：广播只在 commit 之后发生（顺序断言） + P5 广播通道异常不撤销事实，轮询兜底仍可取到事件 + P6 缺租户 fail-closed + P7 客户端写服务端权威字段被拒 + P8 乐观锁 if_state_version 冲突返回 PBL_RT_STATE_CONFLICT + P9 时延字段与 SLA 预算齐备（裁决/广播/端到端 + 3s 兜底周期） + P10 append-only 守卫 + 按月分区算法 + +用法：python3 scripts/m11b_selftest.py （退出码 0=PASS） +""" +import os +import re +import sys + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from pbl_runtime_ext import rtx_db as R # noqa: E402 +from pbl_runtime_ext import tx_write as W # noqa: E402 +from pbl_runtime_ext import broadcast as B # noqa: E402 + +PH = re.compile(r"\$\{(\w+)\}\$") +COLS = re.compile(r"INSERT\s+INTO\s+`?(\w+)`?\s*\(([^)]*)\)", re.IGNORECASE) + + +class Store(object): + def __init__(self): + self.tables = {} + self.seq = 0 + self.commits = [] + self.rollbacks = [] + self.order = [] # 事件序列（commit / broadcast / insert） + + def rows(self, t): + return self.tables.setdefault(t, []) + + def insert(self, table, row): + self.seq += 1 + row = dict(row) + row["id"] = self.seq + self.rows(table).append(row) + self.order.append("insert:%s" % table) + return self.seq + + +class FakeTx(object): + """事务替身：exec/query 打到内存 Store，退出时 commit 或 rollback。""" + + def __init__(self, store, boom_on=None): + self.store = store + self.boom_on = boom_on + self.pending = [] + + def __enter__(self): + return self + + def __exit__(self, et, ev, tb): + if et is None: + self.store.tables = getattr(self, "_final", self.store.tables) + self.store.commits.append(len(self.store.rows("pbl_runtime_event"))) + self.store.order.append("commit") + else: + for t, rows in getattr(self, "_snapshot", {}).items(): + self.store.tables[t] = rows + self.store.rollbacks.append(str(ev)[:80]) + self.store.order.append("rollback") + return False + + def _begin(self): + self._snapshot = {k: list(v) for k, v in self.store.tables.items()} + + def exec(self, sql, params=None): + params = params or {} + if self.boom_on and self.boom_on in sql: + raise RuntimeError("injected failure on %s" % self.boom_on) + m = COLS.search(sql) + if m: + table = m.group(1) + cols = [c.strip().strip("`") for c in m.group(2).split(",")] + ph = PH.findall(sql.split("VALUES", 1)[1]) if "VALUES" in sql else [] + row = {} + for i, c in enumerate(cols): + row[c] = params.get(ph[i]) if i < len(ph) else None + if table == "pbl_entity_state": + key = (row.get("tenant_id"), row.get("session_id"), + row.get("entity_id"), row.get("state_key")) + for r in self.store.rows(table): + if (r.get("tenant_id"), r.get("session_id"), + r.get("entity_id"), r.get("state_key")) == key: + r.update(row) + self.store.order.append("insert:pbl_entity_state") + return 2 + self.pending.append(key) + return self.store.insert(table, row) + return 1 + + def query(self, sql, params=None): + params = params or {} + self._begin() + if "MAX(`seq_no`)" in sql: + vals = [int(r.get("seq_no") or 0) for r in self.store.rows("pbl_runtime_event") + if str(r.get("tenant_id")) == str(params.get("t")) + and str(r.get("session_id")) == str(params.get("s"))] + return [{"s": max(vals) if vals else 0}] + if "MAX(`state_version`)" in sql: + vals = [int(r.get("state_version") or 0) for r in self.store.rows("pbl_entity_state") + if str(r.get("tenant_id")) == str(params.get("t")) + and str(r.get("session_id")) == str(params.get("s"))] + return [{"v": max(vals) if vals else 0}] + if "LAST_INSERT_ID" in sql: + return [{"i": self.store.seq}] + if "FROM `pbl_entity_state`" in sql: + out = [] + for r in self.store.rows("pbl_entity_state"): + if (str(r.get("tenant_id")) == str(params.get("t")) + and str(r.get("session_id")) == str(params.get("s")) + and str(r.get("entity_id")) == str(params.get("e")) + and str(r.get("state_key")) == str(params.get("k"))): + out.append({"state_value": r.get("state_value"), + "state_version": r.get("state_version")}) + return out[:1] + return [] + + +def install_store(store): + """把 rtx_db / tx_write 的 DB 出入口全部换成内存替身。""" + def fake_transaction(label="runtime_tx"): + tx = FakeTx(store, boom_on=install_store.boom) + tx._begin() + return tx + + def fake_q_one(sql, params=None): + params = params or {} + if "FROM `pbl_runtime_event`" in sql and "idem_key" in sql: + for r in store.rows("pbl_runtime_event"): + if (str(r.get("tenant_id")) == str(params.get("t")) + and str(r.get("idem_key")) == str(params.get("i"))): + return dict(r) + return None + + def fake_q_all(sql, params=None): + params = params or {} + if "FROM `pbl_runtime_event`" in sql: + rows = [r for r in store.rows("pbl_runtime_event") + if str(r.get("tenant_id")) == str(params.get("t")) + and str(r.get("session_id")) == str(params.get("s")) + and int(r.get("seq_no") or 0) > int(params.get("q") or 0)] + return sorted(rows, key=lambda r: int(r.get("seq_no") or 0))[:int(params.get("n") or 100)] + if "FROM `pbl_entity_state`" in sql: + return [dict(r) for r in store.rows("pbl_entity_state") + if str(r.get("tenant_id")) == str(params.get("t")) + and str(r.get("session_id")) == str(params.get("s"))] + return [] + + def fake_sql_exec(sql, params=None): + return 1 + + fake_transaction.__call__ = fake_transaction + install_store.boom = getattr(install_store, "boom", None) + W.transaction = fake_transaction + R.q_one = fake_q_one + R.q_all = fake_q_all + R.sql_exec = fake_sql_exec + W.ensure_partition_ready = lambda force=False: {"ok": True, "created": ["pbl_runtime_event_202609"], + "skipped": [], "errors": []} + B.set_hub(B.BroadcastHub(workers=0)) + return B.get_hub() + + +RESULTS = [] + + +def check(name, cond, detail=""): + RESULTS.append((bool(cond), name, detail)) + print("%s %s %s" % ("PASS" if cond else "FAIL", name, detail)) + + +def main(): + hub = None + # ---------------- P1 单事务写入 + store = Store() + hub = install_store(store) + install_store.boom = None + pushed = [] + hub.register_push(lambda sk, ev: pushed.append((sk, ev)) or 1, name="probe") + r1 = W.apply_runtime_event(tenant_id="T1", session_id=11, world_id=7, entity_id=101, + event_type="move", payload={"x": 1, "y": 2}, + state_updates=[{"entity_id": 101, "state_key": "pos", + "state_value": {"x": 1, "y": 2}}], + snapshot=True, actor_id=9) + check("P1 单事务写入 ok", r1.get("ok") and r1.get("transaction") == "single", + "seq_no=%s state_version=%s" % (r1.get("seq_no"), r1.get("state_version"))) + check("P1 事件 1 行 + 状态 1 行 + 快照 1 行", + len(store.rows("pbl_runtime_event")) == 1 + and len(store.rows("pbl_entity_state")) == 1 + and len(store.rows("pbl_world_state_snapshot")) == 1, + "ev=%d st=%d sn=%d" % (len(store.rows("pbl_runtime_event")), + len(store.rows("pbl_entity_state")), + len(store.rows("pbl_world_state_snapshot")))) + check("P1 提交发生", len(store.commits) == 1, "commits=%d" % len(store.commits)) + + # ---------------- P4 提交后广播（顺序） + check("P4 提交后广播", "commit" in store.order and store.order.index("commit") + < len(store.order) and pushed and + store.order.index("commit") < len([o for o in store.order if o.startswith("insert")]) + 1, + "order=%s pushed=%d" % (store.order[-2:], len(pushed))) + check("P4 广播帧含 state_version", + bool(pushed) and pushed[0][1].get("state_version") == r1.get("state_version")) + + # ---------------- P2 幂等 + r2 = W.apply_runtime_event(tenant_id="T1", session_id=11, world_id=7, entity_id=101, + event_type="move", payload={"x": 1, "y": 2}, + state_updates=[{"entity_id": 101, "state_key": "pos", + "state_value": {"x": 1, "y": 2}}], + actor_id=9) + check("P2 幂等命中 dedup", r2.get("dedup") is True, "idem=%s" % str(r2.get("idem_key"))[:12]) + check("P2 未重复写事件", len(store.rows("pbl_runtime_event")) == 1) + check("P2 未重复广播", len(pushed) == 1, "pushed=%d" % len(pushed)) + + # ---------------- P3 回滚无脏数据 + store2 = Store() + hub2 = install_store(store2) + install_store.boom = "pbl_entity_state" + try: + W.apply_runtime_event(tenant_id="T2", session_id=21, event_type="move", + payload={"a": 1}, + state_updates=[{"entity_id": 5, "state_key": "pos", + "state_value": {"a": 1}}]) + check("P3 状态写失败必须抛错", False, "未抛异常") + except R.PblRtError as exc: + check("P3 状态写失败抛 TX_FAILED", exc.code == "PBL_RT_TX_FAILED", "code=%s" % exc.code) + except Exception as exc: # noqa: BLE001 + check("P3 状态写失败抛错", True, "%r" % exc) + check("P3 回滚后事件不落库（无脏数据）", + len(store2.rows("pbl_runtime_event")) == 0 and len(store2.rollbacks) > 0, + "ev=%d rollbacks=%d" % (len(store2.rows("pbl_runtime_event")), len(store2.rollbacks))) + install_store.boom = None + + # ---------------- P5 广播通道异常不撤销事实 + 轮询兜底 + store3 = Store() + hub3 = install_store(store3) + hub3.register_push(lambda sk, ev: (_ for _ in ()).throw(RuntimeError("ws down")), + name="broken") + r5 = W.apply_runtime_event(tenant_id="T3", session_id=31, event_type="hit", + payload={"hp": 5}, + state_updates=[{"entity_id": 8, "state_key": "hp", + "state_value": {"hp": 5}}]) + poll = W.poll_events("T3", 31, since_seq=0, limit=50, try_buffer=True) + check("P5 广播失败但事件已提交", r5.get("ok") and len(store3.rows("pbl_runtime_event")) == 1) + check("P5 轮询兜底取到事件", poll.get("count") == 1 and + poll["events"][0]["seq_no"] == r5["seq_no"], + "count=%s source=%s" % (poll.get("count"), poll.get("source"))) + check("P5 兜底周期 3s", poll.get("poll_interval_ms") == 3000) + + # ---------------- P6 缺租户 fail-closed + try: + W.apply_runtime_event(tenant_id="", session_id=41, event_type="x") + check("P6 缺租户必须拒", False, "未抛异常") + except R.PblRtError as exc: + check("P6 缺租户必须拒", exc.code == "PBL-TENANT-0001", "code=%s" % exc.code) + + # ---------------- P7 客户端写权威字段被拒 + try: + W.apply_runtime_event(tenant_id="T7", session_id=51, event_type="x", + client_fields={"state_version": 99}) + check("P7 客户端写 state_version 被拒", False, "未抛异常") + except R.PblRtError as exc: + check("P7 客户端写 state_version 被拒", exc.code == "PBL_RT_BAD_PAYLOAD", + "code=%s" % exc.code) + + # ---------------- P8 乐观锁冲突 + store8 = Store() + install_store(store8) + W.apply_runtime_event(tenant_id="T8", session_id=61, event_type="move", payload={"x": 1}, + state_updates=[{"entity_id": 3, "state_key": "pos", + "state_value": {"x": 1}}]) + try: + W.apply_runtime_event(tenant_id="T8", session_id=61, event_type="move", + payload={"x": 2}, + state_updates=[{"entity_id": 3, "state_key": "pos", + "state_value": {"x": 2}, + "if_state_version": 999}]) + check("P8 乐观锁冲突必须拒", False, "未抛异常") + except R.PblRtError as exc: + check("P8 乐观锁冲突必须拒", exc.code == "PBL_RT_STATE_CONFLICT", "code=%s" % exc.code) + check("P8 冲突后状态版本未推进", + len(store8.rows("pbl_entity_state")) == 1 and + int(store8.rows("pbl_entity_state")[0]["state_version"]) == 1) + + # ---------------- P9 时延与 SLA + lat = r1.get("latency") or {} + sla = lat.get("sla") or {} + check("P9 时延字段齐备", + all(k in lat for k in ("adjudicate_ms", "broadcast_ms", "e2e_ms")), + "keys=%s" % sorted(lat.keys())) + check("P9 SLA 预算 200/300/1000 + 3s 兜底", + sla.get("adjudicate_ms") == 200 and sla.get("broadcast_ms") == 300 + and sla.get("e2e_ms") == 1000 and sla.get("poll_fallback_interval_ms") == 3000, + "sla=%s" % sla) + check("P9 实测在预算内", r1.get("within_sla") is True, + "adjudicate=%sms broadcast=%sms e2e=%sms" + % (lat.get("adjudicate_ms"), lat.get("broadcast_ms"), lat.get("e2e_ms"))) + + # ---------------- P10 append-only 守卫 + 分区算法 + ok_tx, msgs_tx = W.self_check() + check("P10 tx_write 自检（append-only/分区/幂等键）", ok_tx, + "; ".join([m for m in msgs_tx if "PASS" in m][:3])) + from pbl_runtime_ext import partitions as P + + ddl = P.ddl_create_partition(__import__("datetime").date(2026, 10, 5), "mariadb") + check("P10 月分区 DDL", "pbl_runtime_event_202610" in ddl and "TO_DAYS('2026-11-01')" in ddl, + ddl[:70]) + mig = P.partition_migration_ddl(2, "mariadb") + check("P10 分区改造 DDL 含复合主键+MAXVALUE 兜底", + any("ADD PRIMARY KEY (`id`, `created_at`)" in s for s in mig) + and any("MAXVALUE" in s for s in mig)) + try: + P.ddl_drop_partition("pbl_runtime_event_202610") + P.ddl_drop_partition("pbl_entity_state") + check("P10 拒绝删除非事件月分区", False) + except ValueError as exc: + check("P10 拒绝删除非事件月分区", True, str(exc)[:40]) + + fails = [r for r in RESULTS if not r[0]] + print("\n==== M11b SELFTEST: %d/%d PASS ====" % (len(RESULTS) - len(fails), len(RESULTS))) + for _ok, name, detail in fails: + print(" FAIL %s %s" % (name, detail)) + return 1 if fails else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/sql/m11b_partitions.sql b/sql/m11b_partitions.sql new file mode 100644 index 0000000..46b47d5 --- /dev/null +++ b/sql/m11b_partitions.sql @@ -0,0 +1,72 @@ +-- ============================================================================ +-- pbl_runtime_ext / M11b：pbl_runtime_event 按月 RANGE 分区（mariadb 方言） +-- 与 sql/pbl_runtime_ext.sql（由 tables.py 单一真源生成）配套使用： +-- 新建环境：直接执行本文件的「A. 建表期分区」段 +-- 已有环境（M11a 非分区表）：执行「B. 一次性改造」段 +-- 运行期：由模块 partitions.ensure_forward_partitions() 自动预建当月+未来 2 个月， +-- 本文件仅提供等价 SQL 供 DBA 核对 / 应急手工执行（pmax 是最后兜底）。 +-- 约束：RANGE 分区键必须进入主键与所有唯一键 → PK(id, created_at)、 +-- uk_tenant_idem(tenant_id, idem_key, created_at)。 +-- ============================================================================ + +-- ---------------------------------------------------------------- A. 建表期分区 +-- 事件表 append-only：业务侧只 INSERT，本文件不提供任何 UPDATE/DELETE 语句。 +CREATE TABLE IF NOT EXISTS `pbl_runtime_event` ( + `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键', + `tenant_id` BIGINT NOT NULL DEFAULT 0 COMMENT '租户ID（多租户强制打头）', + `event_code` VARCHAR(64) NOT NULL COMMENT '事件编码', + `idem_key` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '幂等键（同键重复投递只落一条）', + `world_id` BIGINT NOT NULL DEFAULT 0 COMMENT '世界ID（引用 world 基表，只读）', + `session_id` BIGINT NOT NULL DEFAULT 0 COMMENT '游戏会话ID（引用 scense 基表，只读）', + `entity_id` BIGINT NOT NULL DEFAULT 0 COMMENT '实体ID（引用 entity 基表，只读）', + `event_type` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事件类型', + `payload` TEXT COMMENT '事件负载 JSON', + `seq_no` BIGINT NOT NULL DEFAULT 0 COMMENT '会话内单调序号（轮询游标）', + `source` VARCHAR(32) NOT NULL DEFAULT 'runtime' COMMENT '来源：runtime/agent/script/client_intent', + `state` VARCHAR(16) NOT NULL DEFAULT 'applied' COMMENT 'applied/rejected/rolled_back', + `tx_group` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '事务组（同组同事务）', + `broadcast` TINYINT NOT NULL DEFAULT 1 COMMENT '是否参与广播 0/1', + `state_version` BIGINT NOT NULL DEFAULT 0 COMMENT '本事件推进到的服务端权威版本', + `created_by` BIGINT NOT NULL DEFAULT 0 COMMENT '创建人', + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间（分区键）', + PRIMARY KEY (`id`, `created_at`), + UNIQUE KEY `uk_tenant_idem` (`tenant_id`, `idem_key`, `created_at`), + KEY `ix_tenant_session_seq` (`tenant_id`, `session_id`, `seq_no`), + KEY `ix_tenant_world` (`tenant_id`, `world_id`), + KEY `ix_tenant_type` (`tenant_id`, `event_type`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 + COMMENT='运行时事件流（append-only，按月 RANGE 分区，保留 14 个月）' +PARTITION BY RANGE (TO_DAYS(`created_at`)) ( + PARTITION `pbl_runtime_event_202609` VALUES LESS THAN (TO_DAYS('2026-10-01')), + PARTITION `pbl_runtime_event_202610` VALUES LESS THAN (TO_DAYS('2026-11-01')), + PARTITION `pbl_runtime_event_202611` VALUES LESS THAN (TO_DAYS('2026-12-01')), + PARTITION `pbl_runtime_event_202612` VALUES LESS THAN (TO_DAYS('2027-01-01')), + PARTITION `pbl_runtime_event_pmax` VALUES LESS THAN MAXVALUE +); + +-- ---------------------------------------------------------------- B. 一次性改造（M11a 非分区表 → 分区表） +-- 顺序不可调换：先扩主键/唯一键，再 PARTITION BY。执行前请备份。 +-- ALTER TABLE `pbl_runtime_event` DROP PRIMARY KEY, ADD PRIMARY KEY (`id`, `created_at`); +-- ALTER TABLE `pbl_runtime_event` DROP INDEX `uk_tenant_idem`, +-- ADD UNIQUE KEY `uk_tenant_idem` (`tenant_id`, `idem_key`, `created_at`); +-- ALTER TABLE `pbl_runtime_event` PARTITION BY RANGE (TO_DAYS(`created_at`)) ( +-- PARTITION `pbl_runtime_event_202609` VALUES LESS THAN (TO_DAYS('2026-10-01')), +-- PARTITION `pbl_runtime_event_pmax` VALUES LESS THAN MAXVALUE +-- ); + +-- ---------------------------------------------------------------- C. 月度维护（每月 1 日定时任务） +-- 模块写入前会自动预建（ensure_forward_partitions，进程内 TTL 6h），下面是等价手工语句： +-- ALTER TABLE `pbl_runtime_event` ADD PARTITION (PARTITION `pbl_runtime_event_202701` VALUES LESS THAN (TO_DAYS('2027-02-01'))); +-- 查询当前分区： +-- SELECT PARTITION_NAME, PARTITION_DESCRIPTION FROM information_schema.PARTITIONS +-- WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pbl_runtime_event'; +-- 过期清理（保留 14 个月；DROP PARTITION 是元数据操作，不产生 DELETE 洪峰）： +-- 由 partitions.drop_expired_partitions(retain=14, dry_run=True) 先出清单， +-- 人工确认后再 dry_run=False 执行，等价： +-- ALTER TABLE `pbl_runtime_event` DROP PARTITION `pbl_runtime_event_202507`; + +-- ---------------------------------------------------------------- D. 实体状态 / 快照（M11a 定义不变） +-- pbl_entity_state：服务端权威，state_version 由 apply_runtime_event 单调分配， +-- 客户端提交值一律忽略；uk(tenant_id, session_id, entity_id, state_key) 支撑 UPSERT。 +-- pbl_world_state_snapshot：掉线补齐/离线兜底基线，与事件同事务写入。 +-- 见 sql/pbl_runtime_ext.sql（单一真源生成），本文件不重复定义，避免两处漂移。 diff --git a/wwwroot/api/pbl_runtime_broadcast_pull.dspy b/wwwroot/api/pbl_runtime_broadcast_pull.dspy new file mode 100644 index 0000000..8f18fd6 --- /dev/null +++ b/wwwroot/api/pbl_runtime_broadcast_pull.dspy @@ -0,0 +1,5 @@ +# pbl_runtime_ext/api/pbl_runtime_broadcast_pull.dspy —— M11b 广播增量拉取（掉线补齐，命中内存环形缓冲） +# 契约实现：pbl_runtime_ext/m11b_api.py::pbl_runtime_broadcast_pull（权威实现走 tx_write 主链路） +debug('pbl_runtime_ext/api/pbl_runtime_broadcast_pull.dspy: START params_kw={dict(params_kw)}') +data = await pbl_runtime_broadcast_pull(**params_kw) +return data diff --git a/wwwroot/api/pbl_runtime_broadcast_stats.dspy b/wwwroot/api/pbl_runtime_broadcast_stats.dspy new file mode 100644 index 0000000..be831a7 --- /dev/null +++ b/wwwroot/api/pbl_runtime_broadcast_stats.dspy @@ -0,0 +1,5 @@ +# pbl_runtime_ext/api/pbl_runtime_broadcast_stats.dspy —— M11b 广播观测（发布/投递/失败/最大耗时 + SLA 判定） +# 契约实现：pbl_runtime_ext/m11b_api.py::pbl_runtime_broadcast_stats +debug('pbl_runtime_ext/api/pbl_runtime_broadcast_stats.dspy: START params_kw={dict(params_kw)}') +data = await pbl_runtime_broadcast_stats(**params_kw) +return data