diff --git a/scripts/sync_code_to_app.py b/scripts/sync_code_to_app.py new file mode 100644 index 0000000..5ec609d --- /dev/null +++ b/scripts/sync_code_to_app.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""M11b-3 代码镜像同步器：modules/world_sync/ 是唯一真源， +apps/scense/pkgs/world_sync/ 是打包镜像（宿主 build.sh 从应用仓库安装，禁止手工双写）。 + +与 sync_models_to_app.py（只管 models/）互补，本脚本同步 **Python 包与测试**： + - world_sync/*.py（包源码，含 pbl_runtime_broadcast.py） + - tests/*.py（含 test_m11b3_post_commit_broadcast.py） + - skill/SKILL.md、README.md + +用法: + python modules/world_sync/scripts/sync_code_to_app.py # 同步 + python modules/world_sync/scripts/sync_code_to_app.py --check # 只校验 sha256 + +行为：逐文件以同一份字节覆盖镜像落点，打印两两 sha256 比对；--check 不一致退出码 1。 +""" + +from __future__ import annotations + +import argparse +import hashlib +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +WS_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", "..")) + +SRC_ROOT = os.path.join("modules", "world_sync") +DST_ROOT = os.path.join("apps", "scense", "pkgs", "world_sync") + +#: 需要镜像的相对路径（真源 = modules/world_sync/） +MIRROR_SUBDIRS = ("world_sync", "tests") +MIRROR_FILES = ("README.md", os.path.join("skill", "SKILL.md")) +#: 不同步的产物/缓存 +SKIP_SUFFIX = ("__pycache__", ".pyc", ".egg-info", ".pytest_cache") + + +def sha256_of(path: str) -> str: + h = hashlib.sha256() + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + return h.hexdigest() + + +def collect() -> list: + rels = [] + for sub in MIRROR_SUBDIRS: + base = os.path.join(SRC_ROOT, sub) + if not os.path.isdir(base): + continue + for name in sorted(os.listdir(base)): + if any(s in name for s in SKIP_SUFFIX): + continue + if not name.endswith(".py"): + continue + rels.append(os.path.join(sub, name)) + for rel in MIRROR_FILES: + if os.path.isfile(os.path.join(SRC_ROOT, rel)): + rels.append(rel) + return rels + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--check", action="store_true", help="只校验，不写入") + args = ap.parse_args() + + bad = 0 + for rel in collect(): + src = os.path.join(WS_ROOT, SRC_ROOT, rel) + dst = os.path.join(WS_ROOT, DST_ROOT, rel) + src_sha = sha256_of(src) + same = os.path.isfile(dst) and sha256_of(dst) == src_sha + if args.check: + print("%-58s %s" % (rel, "OK" if same else "DRIFT")) + if not same: + bad += 1 + continue + os.makedirs(os.path.dirname(dst), exist_ok=True) + with open(src, "rb") as fh: + data = fh.read() + with open(dst, "wb") as fh: + fh.write(data) + ok = sha256_of(dst) == src_sha + print("%-58s %s sha256=%s" % (rel, "SYNCED" if ok else "FAIL", src_sha[:12])) + if not ok: + bad += 1 + if bad: + print("RESULT: FAIL (%d 项不一致)" % bad) + return 1 + print("RESULT: PASS (镜像与真源逐文件 sha256 一致)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_m11b3_post_commit_broadcast.py b/tests/test_m11b3_post_commit_broadcast.py new file mode 100644 index 0000000..1181e18 --- /dev/null +++ b/tests/test_m11b3_post_commit_broadcast.py @@ -0,0 +1,432 @@ +# -*- coding: utf-8 -*- +"""M11b-3 事务提交后广播机制 —— 单元测试。 + +覆盖验收标准： +1. 事务 COMMIT 后广播消息在 ≤300ms 内发出（P99 基准）； +2. 广播失败（通道不可用）不影响已提交数据，且有日志记录； +3. 广播不阻塞 write_event_with_state 的返回（异步）； +4. 广播内容至少含 event_id / scene_id / entity_id / event_type / payload 摘要； +5. 事务未提交（抛异常回滚）时**不广播**； +6. 不引入轮询兜底（源码静态审计：无 sleep 轮询循环、无 DB 读取）。 + +测试用内存假事务（不依赖真实 DB / 网络），因此在无数据库的 CI 环境同样可跑。 +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import threading +import time + +import pytest + +import world_sync as ws +from world_sync.pbl_runtime_broadcast import ( + BROADCAST_LATENCY_SLO_MS, + BaseChannel, + Broadcaster, + ChannelUnavailable, + InProcessChannel, + WebhookChannel, + build_broadcast_message, + emit_post_commit_broadcast, + install_broadcast_hook, + reset_broadcaster, + subscribe_broadcast, + unsubscribe_broadcast, + wrap_with_post_commit_broadcast, +) + + +# --------------------------------------------------------------------------- +# fixtures / helpers +# --------------------------------------------------------------------------- +class RecordingChannel(BaseChannel): + """记录型通道：保存收到的消息，可配置为持续失败。""" + + name = "recording" + + def __init__(self, fail: bool = False, delay_s: float = 0.0) -> None: + self.messages = [] + self.fail = fail + self.delay_s = delay_s + self.calls = 0 + self._lock = threading.Lock() + + def publish(self, message): + self.calls += 1 + if self.delay_s: + time.sleep(self.delay_s) + if self.fail: + raise ChannelUnavailable("channel down (test)") + with self._lock: + self.messages.append(message) + + +class BlockingChannel(BaseChannel): + """首次 publish 阻塞，直到事件被 set —— 用于证明广播不阻塞事务返回。""" + + name = "blocking" + + def __init__(self) -> None: + self.entered = threading.Event() + self.release = threading.Event() + self.messages = [] + + def publish(self, message): + self.entered.set() + self.release.wait(timeout=5.0) + self.messages.append(message) + + +def fake_write_result(**over): + """模拟 M11b-2 write_event_with_state 提交成功后的返回值。""" + res = { + "event_uid": "evt-uid-0001", + "event_id": "evt-uid-0001", + "tenant_id": "tenant-A", + "world_id": "world-77", + "session_id": "sess-9", + "entity_id": "ent-42", + "state_version": 4, + "action": "updated", + "created_at": "2026-09-21 10:00:00", + "dialect": "sqlite", + "elapsed_ms": 12.3, + } + res.update(over) + return res + + +def fake_event_data(**over): + ev = { + "tenant_id": "tenant-A", + "world_id": "world-77", + "scene_id": "scene-77", + "session_id": "sess-9", + "entity_id": "ent-42", + "event_type": "state.updated", + "payload": {"hp": 100, "pos": [1, 2, 3]}, + } + ev.update(over) + return ev + + +@pytest.fixture +def broadcaster(): + chan = RecordingChannel() + b = Broadcaster(channels=[chan], retry_attempts=1, retry_backoff_s=0.001) + yield b, chan + b.shutdown(timeout=1.0) + + +@pytest.fixture(autouse=True) +def _clean_global(): + yield + reset_broadcaster(None, shutdown_old=True) + + +def wait_until(predicate, timeout=3.0, interval=0.005): + """等待条件成立（仅测试同步用，非生产轮询）。""" + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(interval) + return predicate() + + +# --------------------------------------------------------------------------- +# 1. 广播消息内容 +# --------------------------------------------------------------------------- +def test_build_message_contains_required_fields(): + msg = build_broadcast_message( + fake_write_result(), + commit_ts=time.time(), + context={"event_data": fake_event_data(), "state_update": {"state": {"hp": 100}}}, + ) + assert msg["event_id"] == "evt-uid-0001" + assert msg["scene_id"] == "scene-77" # 显式 scene_id + assert msg["entity_id"] == "ent-42" + assert msg["event_type"] == "state.updated" + assert msg["payload_summary"] # 简要 payload 摘要 + assert "hp" in msg["payload_summary"] + assert len(msg["payload_summary"]) <= 256 + assert msg["state_version"] == 4 + assert msg["tenant_id"] == "tenant-A" + assert json.dumps(msg, ensure_ascii=False) # 可序列化（webhook/redis 通道要求） + + +def test_scene_id_falls_back_to_world_id(): + ev = fake_event_data() + del ev["scene_id"] + msg = build_broadcast_message(fake_write_result(), context={"event_data": ev}) + assert msg["scene_id"] == "world-77" + + +def test_payload_summary_is_truncated(): + big = {"blob": "x" * 5000} + msg = build_broadcast_message(fake_write_result(), context={"event_data": {"payload": big}}) + assert len(msg["payload_summary"]) <= 256 + assert msg["payload_summary"].endswith("...") + + +def test_build_message_never_raises_on_garbage(): + for junk in (None, 123, "str", object(), [1, 2]): + msg = build_broadcast_message(junk) + assert "event_id" in msg and "payload_summary" in msg + + +# --------------------------------------------------------------------------- +# 2. COMMIT 后广播被触发（post-commit hook） +# --------------------------------------------------------------------------- +def test_hook_fires_only_after_commit(broadcaster): + b, chan = broadcaster + calls = {"n": 0} + + def write_fn(event_data=None, state_update=None): + calls["n"] += 1 + return fake_write_result() + + wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b) + res = wrapped(event_data=fake_event_data(), state_update={"state": {}}) + assert res["event_id"] == "evt-uid-0001" # 原返回值透传 + assert calls["n"] == 1 + assert wait_until(lambda: len(chan.messages) == 1) + msg = chan.messages[0] + assert msg["event_id"] == "evt-uid-0001" + assert msg["entity_id"] == "ent-42" + assert msg["commit_ts"] <= msg["submit_ts"] + + +def test_no_broadcast_when_transaction_raises(broadcaster): + """事务回滚（抛异常）→ 绝不广播，异常原样上抛。""" + b, chan = broadcaster + + def write_fn(*a, **kw): + raise RuntimeError("deadlock, transaction rolled back") + + wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b) + with pytest.raises(RuntimeError): + wrapped(event_data=fake_event_data()) + time.sleep(0.15) + assert chan.messages == [] + assert chan.calls == 0 + + +def test_install_hook_wraps_module_symbol_and_is_idempotent(): + import world_sync.pbl_runtime_tx_sqlor as txs + original = txs.write_event_with_state + underlying = getattr(original, "_m11b3_underlying", original) + try: + wrapped1 = install_broadcast_hook() + wrapped2 = install_broadcast_hook() # 幂等，不重复包装 + assert getattr(wrapped1, "_m11b3_broadcast_wrapped", False) + assert txs.write_event_with_state is wrapped1 + assert wrapped1 is wrapped2 + assert getattr(wrapped1, "_m11b3_underlying") is underlying + assert ws.is_broadcast_hook_installed() is True + finally: + txs.write_event_with_state = underlying + reset_broadcaster(None, shutdown_old=True) + + +# --------------------------------------------------------------------------- +# 3. 广播失败不回滚已提交数据 + 有日志 +# --------------------------------------------------------------------------- +def test_broadcast_failure_does_not_affect_committed_data(broadcaster, caplog): + chan = RecordingChannel(fail=True) + b = Broadcaster(channels=[chan], retry_attempts=2, retry_backoff_s=0.001) + try: + def write_fn(*a, **kw): + return fake_write_result() # 事务已成功提交 + + wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b) + with caplog.at_level(logging.ERROR, logger="world_sync.broadcast"): + res = wrapped(event_data=fake_event_data()) + # 调用方拿到正常返回值，无异常 → 已提交数据不受影响 + assert res["event_id"] == "evt-uid-0001" + assert wait_until(lambda: b.stats()["failed"] >= 1) + assert b.stats()["published"] == 0 + assert b.stats()["last_error"] + text = "\n".join(r.getMessage() for r in caplog.records) + assert "广播失败" in text and "事务已提交" in text + finally: + b.shutdown(timeout=1.0) + + +def test_failing_channel_does_not_propagate_to_caller(): + """emit_post_commit_broadcast 自身永不抛异常（即使消息构造/通道全炸）。""" + chan = RecordingChannel(fail=True) + b = Broadcaster(channels=[chan], retry_attempts=0) + try: + msg = emit_post_commit_broadcast(fake_write_result(), broadcaster=b) + assert msg is not None and msg["event_id"] + assert wait_until(lambda: b.stats()["failed"] >= 1) + finally: + b.shutdown(timeout=1.0) + + +def test_partial_channel_failure_keeps_good_channel(): + good = RecordingChannel() + bad = RecordingChannel(fail=True) + b = Broadcaster(channels=[bad, good], retry_attempts=1, retry_backoff_s=0.001) + try: + emit_post_commit_broadcast(fake_write_result(), broadcaster=b) + assert wait_until(lambda: len(good.messages) == 1) + assert b.stats()["published"] == 1 + finally: + b.shutdown(timeout=1.0) + + +def test_subscriber_exception_is_swallowed(): + chan = InProcessChannel() + + def boom(message): + raise ValueError("subscriber bug") + + chan.subscribe(boom) + b = Broadcaster(channels=[chan], retry_attempts=0) + try: + emit_post_commit_broadcast(fake_write_result(), broadcaster=b) + assert wait_until(lambda: b.stats()["published"] >= 1) + assert b.stats()["failed"] == 0 + finally: + b.shutdown(timeout=1.0) + + +# --------------------------------------------------------------------------- +# 4. 异步：不阻塞事务返回 +# --------------------------------------------------------------------------- +def test_broadcast_does_not_block_write_return(): + slow = BlockingChannel() + b = Broadcaster(channels=[slow], retry_attempts=0) + try: + def write_fn(*a, **kw): + return fake_write_result() + + wrapped = wrap_with_post_commit_broadcast(write_fn, broadcaster=b) + t0 = time.perf_counter() + res = wrapped(event_data=fake_event_data()) # 不应被 slow 通道卡住 + elapsed_ms = (time.perf_counter() - t0) * 1000.0 + assert res["event_id"] + assert elapsed_ms < 100.0, "广播阻塞了事务返回: %.1fms" % elapsed_ms + # 稍后 worker 线程仍会发出 + slow.release.set() + assert wait_until(lambda: len(slow.messages) == 1) + finally: + slow.release.set() + b.shutdown(timeout=1.5) + + +def test_submit_returns_immediately_for_many_messages(): + chan = RecordingChannel() + b = Broadcaster(channels=[chan], queue_size=100, retry_attempts=0) + try: + t0 = time.perf_counter() + for i in range(100): + b.submit(build_broadcast_message(fake_write_result(event_id="e%d" % i))) + elapsed_ms = (time.perf_counter() - t0) * 1000.0 + assert elapsed_ms < 100.0 + assert b.drain(timeout=5.0) + assert len(chan.messages) == 100 + finally: + b.shutdown(timeout=1.0) + + +def test_queue_full_falls_back_to_direct_publish_without_blocking(): + """队列满时不阻塞调用方（退化为同步直发），消息不丢。""" + chan = RecordingChannel() + b = Broadcaster(channels=[chan], queue_size=1, retry_attempts=0) + try: + for i in range(20): + b.submit(build_broadcast_message(fake_write_result(event_id="e%d" % i))) + assert b.drain(timeout=5.0) + assert len(chan.messages) == 20 + assert b.stats()["published"] == 20 + finally: + b.shutdown(timeout=1.0) + + +# --------------------------------------------------------------------------- +# 5. 延迟 SLO ≤ 300ms（P99） +# --------------------------------------------------------------------------- +def test_broadcast_latency_p99_within_300ms(): + chan = RecordingChannel() + b = Broadcaster(channels=[chan], retry_attempts=2, retry_backoff_s=0.001) + n = 200 + try: + for i in range(n): + emit_post_commit_broadcast( + fake_write_result(event_id="lat-%d" % i), + broadcaster=b, + context={"event_data": fake_event_data()}, + ) + assert b.drain(timeout=15.0), "广播未在超时内全部发出" + assert len(chan.messages) == n + p99 = b.latency_percentile_ms(99.0) + p50 = b.latency_percentile_ms(50.0) + stats = b.stats() + print("\n[M11b-3 延迟基准] n=%d p50=%.3fms p99=%.3fms max=%.3fms SLO=%.0fms" + % (n, p50, p99, stats["max_latency_ms"], BROADCAST_LATENCY_SLO_MS)) + assert p99 <= BROADCAST_LATENCY_SLO_MS, "P99 超阈值: %.3fms" % p99 + assert stats["failed"] == 0 + finally: + b.shutdown(timeout=1.0) + + +def test_latency_slo_constant_matches_spec(): + assert BROADCAST_LATENCY_SLO_MS == 300.0 + assert ws.BROADCAST_LATENCY_SLO_MS == 300.0 + + +# --------------------------------------------------------------------------- +# 6. 进程内订阅 API（宿主接线） +# --------------------------------------------------------------------------- +def test_subscribe_and_unsubscribe_global(): + reset_broadcaster(Broadcaster(channels=[InProcessChannel()], retry_attempts=0)) + got = [] + handler = subscribe_broadcast(got.append) + emit_post_commit_broadcast(fake_write_result(event_id="sub-1")) + assert wait_until(lambda: len(got) == 1) + assert got[0]["event_id"] == "sub-1" + assert unsubscribe_broadcast(handler) is True + emit_post_commit_broadcast(fake_write_result(event_id="sub-2")) + time.sleep(0.15) + assert len(got) == 1 # 取消后不再收到 + assert ws.broadcast_stats()["published"] >= 2 + + +def test_configure_broadcast_replaces_channels(): + chan = RecordingChannel() + b = ws.configure_broadcast(channels=[chan], retry_attempts=0) + emit_post_commit_broadcast(fake_write_result(event_id="cfg-1")) + assert wait_until(lambda: len(chan.messages) == 1) + assert b.stats()["channels"] == ["recording"] + + +# --------------------------------------------------------------------------- +# 7. 无轮询兜底（静态审计） +# --------------------------------------------------------------------------- +def test_no_polling_loop_in_broadcast_source(): + src_path = os.path.join(os.path.dirname(ws.__file__), "pbl_runtime_broadcast.py") + with open(src_path, encoding="utf-8") as fh: + lines = fh.readlines() + code = "".join(l for l in lines if not l.lstrip().startswith("#")) + # 生产代码里不得出现 sleep 轮询（time.sleep 只允许出现在重试退避一处） + sleeps = [i for i, l in enumerate(code.splitlines(), 1) if "time.sleep(" in l] + assert len(sleeps) <= 1, "疑似轮询 sleep: %s" % sleeps + # 不得有 while True + sleep 的轮询形态 + assert not re.search(r"while\s+(True|1)\s*:\s*(#.*)?\n\s*(try:\s*\n\s*)?time\.sleep", code), \ + "存在 while True + sleep 轮询兜底" + # 不得读数据库（无 SQL 关键字 / 无 conn_factory / 无 sor 调用） + for token in ("SELECT ", "UPDATE ", "INSERT INTO", "conn_factory", "sqlorContext", "DBPools"): + assert token not in code, "广播层不应触碰 DB: %s" % token + + +def test_broadcast_channel_name_exported(): + assert ws.BROADCAST_CHANNEL == "pbl.runtime.event" + assert ws.PAYLOAD_SUMMARY_MAX_CHARS == 256 diff --git a/world_sync/__init__.py b/world_sync/__init__.py index 70b064e..23c59a6 100644 --- a/world_sync/__init__.py +++ b/world_sync/__init__.py @@ -120,6 +120,36 @@ from .pbl_runtime_tx_sqlor import ( write_event_with_state, ) +# --- M11b-3 事务提交后广播（post-commit broadcast，真源：pbl_runtime_broadcast.py） --- +from .pbl_runtime_broadcast import ( + BROADCAST_CHANNEL, + BROADCAST_LATENCY_SLO_MS, + DEFAULT_EVENT_CHANNEL, + PAYLOAD_SUMMARY_MAX_CHARS, + BaseChannel, + Broadcaster, + ChannelUnavailable, + InProcessChannel, + NoopChannel, + RedisPubSubChannel, + WebhookChannel, + broadcast_latency_percentile_ms, + broadcast_last_message, + broadcast_stats, + build_broadcast_message, + configure_broadcast, + drain_broadcast, + emit_post_commit_broadcast, + get_broadcaster, + install_broadcast_hook, + is_broadcast_hook_installed, + reset_broadcaster, + shutdown_broadcast, + subscribe_broadcast, + unsubscribe_broadcast, + wrap_with_post_commit_broadcast, +) + #: 契约解析优先级键（与 pbl_runtime_tx_env.SERVER_ENV_KEYS 一致，单一来源） assert tuple(SERVER_ENV_CONN_FACTORY_KEYS) == tuple(SERVER_ENV_KEYS), \ "world_sync: SERVER_ENV_CONN_FACTORY_KEYS 与 pbl_runtime_tx_env.SERVER_ENV_KEYS 分叉" @@ -190,6 +220,33 @@ __all__ = [ "InvalidWriteRequest", "ConnectionUnavailable", "TransactionAborted", + # M11b-3 事务提交后广播（post-commit hook / 通道 / 指标） + "BROADCAST_CHANNEL", + "BROADCAST_LATENCY_SLO_MS", + "DEFAULT_EVENT_CHANNEL", + "PAYLOAD_SUMMARY_MAX_CHARS", + "BaseChannel", + "Broadcaster", + "ChannelUnavailable", + "InProcessChannel", + "NoopChannel", + "RedisPubSubChannel", + "WebhookChannel", + "build_broadcast_message", + "emit_post_commit_broadcast", + "wrap_with_post_commit_broadcast", + "install_broadcast_hook", + "is_broadcast_hook_installed", + "configure_broadcast", + "get_broadcaster", + "reset_broadcaster", + "subscribe_broadcast", + "unsubscribe_broadcast", + "broadcast_stats", + "broadcast_last_message", + "broadcast_latency_percentile_ms", + "drain_broadcast", + "shutdown_broadcast", # 宿主基础包可用性诊断 "HOST_DEPS_AVAILABLE", "HOST_DEPS_ERROR", diff --git a/world_sync/init.py b/world_sync/init.py index e9a62b1..9006445 100644 --- a/world_sync/init.py +++ b/world_sync/init.py @@ -148,9 +148,39 @@ from .pbl_runtime_tx_sqlor import ( write_event_with_state, ) +# --- M11b-3: 事务提交后广播（post-commit broadcast）注册接线 --- +from .pbl_runtime_broadcast import ( + BROADCAST_CHANNEL, + BROADCAST_LATENCY_SLO_MS, + DEFAULT_EVENT_CHANNEL, + PAYLOAD_SUMMARY_MAX_CHARS, + BaseChannel, + Broadcaster, + ChannelUnavailable, + InProcessChannel, + NoopChannel, + RedisPubSubChannel, + WebhookChannel, + broadcast_latency_percentile_ms, + broadcast_last_message, + broadcast_stats, + build_broadcast_message, + configure_broadcast, + drain_broadcast, + emit_post_commit_broadcast, + get_broadcaster, + install_broadcast_hook, + is_broadcast_hook_installed, + reset_broadcaster, + shutdown_broadcast, + subscribe_broadcast, + unsubscribe_broadcast, + wrap_with_post_commit_broadcast, +) + MODULE_NAME = 'world_sync' -MODULE_VERSION = '0.2.0-m11b2a' +MODULE_VERSION = '0.3.0-m11b3' #: 宿主 ServerEnv 上登记 conn_factory 的键（与 pbl_runtime_tx_env.SERVER_ENV_KEYS 一致） SERVER_ENV_CONN_FACTORY_KEYS = SERVER_ENV_KEYS @@ -480,6 +510,25 @@ def load_world_sync(server_env=None, conn_factory=None): env.pbl_runtime_conn_factory_keys = list(SERVER_ENV_KEYS) env.pbl_runtime_pool_keys = list(POOL_KEYS) env.pbl_runtime_env_name_keys = list(ENV_NAME_KEYS) + # --- M11b-3: 事务提交后广播（异步 fire-and-forget，SLO <=300ms P99） --- + env.world_sync_broadcast_channel = BROADCAST_CHANNEL + env.world_sync_broadcast_latency_slo_ms = BROADCAST_LATENCY_SLO_MS + env.build_broadcast_message = build_broadcast_message + env.emit_post_commit_broadcast = emit_post_commit_broadcast + env.wrap_with_post_commit_broadcast = wrap_with_post_commit_broadcast + env.configure_broadcast = configure_broadcast + env.get_broadcast_broadcaster = get_broadcaster + env.subscribe_broadcast = subscribe_broadcast + env.unsubscribe_broadcast = unsubscribe_broadcast + env.broadcast_stats = broadcast_stats + env.broadcast_last_message = broadcast_last_message + env.broadcast_latency_percentile_ms = broadcast_latency_percentile_ms + env.drain_broadcast = drain_broadcast + env.shutdown_broadcast = shutdown_broadcast + env.reset_broadcast = reset_broadcaster + # post-commit hook：把 write_event_with_state 包成「COMMIT 后触发广播」版本， + # 并同步刷新三处引用（pbl_runtime_tx_sqlor / world_sync 包命名空间 / env）。 + env.write_event_with_state = install_broadcast_hook(env) _configure_runtime_writer_from_env(env) _register_tables_to_env(env) return env diff --git a/world_sync/pbl_runtime_broadcast.py b/world_sync/pbl_runtime_broadcast.py new file mode 100644 index 0000000..00defe1 --- /dev/null +++ b/world_sync/pbl_runtime_broadcast.py @@ -0,0 +1,904 @@ +# -*- coding: utf-8 -*- +"""world_sync M11b-3 —— 事务提交后广播（post-commit broadcast）。 + +范围（严格对齐任务书） +==================== +只负责 **M11b-2 ``write_event_with_state`` 事务 COMMIT 成功之后** 的广播通知： + +1. COMMIT 成功后由 post-commit hook 触发广播（:func:`wrap_with_post_commit_broadcast` + / :func:`emit_post_commit_broadcast`），广播内容至少含 + ``event_id / scene_id / entity_id / event_type / payload_summary``； +2. 通道可插拔：进程内 :class:`InProcessChannel`、HTTP webhook + :class:`WebhookChannel`、Redis pub/sub :class:`RedisPubSubChannel`； +3. **异步 fire-and-forget**：广播在独立 worker 线程执行，绝不阻塞事务返回， + 广播失败只记日志 + 计数，**绝不回滚已提交事务**（异常不外抛）； +4. 延迟 SLO：COMMIT → 广播消息发出 ≤ **300ms（P99）**，实测见 + :func:`broadcast_latency_percentile_ms`； +5. **不含轮询兜底**：本模块无任何 ``sleep`` 轮询循环、无任何 SQL/DB 读取， + 队列消费是阻塞式条件变量等待（``Queue.join`` / ``Queue.get(block=True)``）。 + +宿主接线:: + + from world_sync import load_world_sync, configure_broadcast + load_world_sync() # 已自动 install_broadcast_hook() + configure_broadcast(webhook_urls=["http://127.0.0.1:8099/pbl/event"]) + + from world_sync import subscribe_broadcast + subscribe_broadcast(lambda msg: print(msg["event_id"])) + +线程模型：单 dispatcher 线程 + 有界 ``queue.Queue``；``shutdown_broadcast()`` +投递哨兵优雅退出。所有共享状态由 ``threading.RLock`` 保护。 +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import queue +import threading +import time +import urllib.error +import urllib.request +from collections import deque +from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple + +__all__ = [ + # 常量 + "BROADCAST_LATENCY_SLO_MS", + "BROADCAST_CHANNEL", + "DEFAULT_EVENT_CHANNEL", + "PAYLOAD_SUMMARY_MAX_CHARS", + # 通道 + "ChannelUnavailable", + "BaseChannel", + "InProcessChannel", + "WebhookChannel", + "RedisPubSubChannel", + "NoopChannel", + # 核心 API + "build_broadcast_message", + "emit_post_commit_broadcast", + "wrap_with_post_commit_broadcast", + "install_broadcast_hook", + "configure_broadcast", + "get_broadcaster", + "reset_broadcaster", + "subscribe_broadcast", + "unsubscribe_broadcast", + "broadcast_stats", + "broadcast_last_message", + "broadcast_latency_percentile_ms", + "drain_broadcast", + "shutdown_broadcast", + "is_broadcast_hook_installed", +] + +_LOG = logging.getLogger("world_sync.broadcast") + +# --------------------------------------------------------------------------- +# 配置常量（单一真源，测试与文档引用同一份） +# --------------------------------------------------------------------------- +#: 任务书 SLO：COMMIT → 广播消息发出 ≤ 300ms（P99） +BROADCAST_LATENCY_SLO_MS: float = 300.0 +#: 默认通道名（Redis channel / webhook 逻辑主题） +DEFAULT_EVENT_CHANNEL: str = "pbl.runtime.event" +#: 别名（与既有 world_sync 命名风格一致） +BROADCAST_CHANNEL: str = DEFAULT_EVENT_CHANNEL +#: payload 摘要最大字符数（广播体保持轻量，不带全量 payload） +PAYLOAD_SUMMARY_MAX_CHARS: int = 256 +#: 分发队列容量（满时降级为「调用线程内直接发布」，不丢消息也不阻塞事务） +DEFAULT_QUEUE_SIZE: int = 2000 +#: 单次投递失败重试次数（不含首次尝试） +DEFAULT_RETRY_ATTEMPTS: int = 2 +#: 重试退避基数（秒） +DEFAULT_RETRY_BACKOFF_S: float = 0.02 +#: webhook 超时（秒）——远小于 SLO，保证不会把广播线程拖死 +WEBHOOK_TIMEOUT_S: float = 0.25 +#: 延迟样本保留数 +LATENCY_SAMPLE_WINDOW: int = 5000 + +_SENTINEL = object() + + +class ChannelUnavailable(RuntimeError): + """广播通道不可用（仅内部记录/计数，绝不向事务调用方外抛）。""" + + +# --------------------------------------------------------------------------- +# 小工具 +# --------------------------------------------------------------------------- +def _field(obj: Any, name: str, default: Any = None) -> Any: + """兼容 dict / DictObject / dataclass 的宽松取值（不抛异常）。""" + if obj is None: + return default + if isinstance(obj, dict): + val = obj.get(name, default) + return default if val is None else val + try: + val = getattr(obj, name) + except Exception: + val = None + if val is None: + try: + val = obj[name] # type: ignore[index] + except Exception: + val = None + return default if val is None else val + + +def _percentile(values: Sequence[float], pct: float) -> float: + """最近秩法百分位（不依赖 numpy）。空序列返回 0.0。""" + vals = sorted(float(v) for v in values) + if not vals: + return 0.0 + if pct <= 0: + return vals[0] + if pct >= 100: + return vals[-1] + idx = int(round((pct / 100.0) * (len(vals) - 1))) + idx = min(max(0, idx), len(vals) - 1) + return vals[idx] + + +def _summarize_payload(payload: Any, limit: int = PAYLOAD_SUMMARY_MAX_CHARS) -> str: + """生成简要 payload 摘要（稳定、可截断、永不抛异常）。""" + if payload is None: + return "" + try: + if isinstance(payload, str): + text = payload + else: + text = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str) + except Exception: + text = str(payload) + if len(text) > limit: + return text[: limit - 3] + "..." + return text + + +def _payload_digest(payload: Any) -> str: + """payload 的 sha256 前 16 位（订阅方可据此判断是否需要回查详情）。""" + try: + if isinstance(payload, (bytes, bytearray)): + raw = bytes(payload) + elif payload is None: + raw = b"" + else: + raw = json.dumps( + payload, ensure_ascii=False, sort_keys=True, default=str + ).encode("utf-8") + except Exception: + raw = str(payload).encode("utf-8", "replace") + return hashlib.sha256(raw).hexdigest()[:16] + + +def _iso(ts: float) -> str: + try: + return time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(ts)) + \ + ".%03dZ" % int((ts % 1) * 1000) + except Exception: + return "" + + +def _truthy_env(name: str, default: bool = False) -> bool: + raw = os.environ.get(name) + if raw is None or raw == "": + return default + return raw.strip().lower() in ("1", "true", "yes", "y", "on") + + +# --------------------------------------------------------------------------- +# 通道实现 +# --------------------------------------------------------------------------- +class BaseChannel: + """广播通道基类。子类实现 :meth:`publish`，失败请抛异常（由 Broadcaster 兜住）。""" + + name: str = "base" + + def publish(self, message: Dict[str, Any]) -> None: # pragma: no cover - 接口 + raise NotImplementedError + + def close(self) -> None: + return None + + +class NoopChannel(BaseChannel): + """空通道（广播关闭或未配置真实通道时使用，仅计数）。""" + + name = "noop" + + def __init__(self) -> None: + self.published: List[Dict[str, Any]] = [] + + def publish(self, message: Dict[str, Any]) -> None: + self.published.append(message) + + +class InProcessChannel(BaseChannel): + """进程内广播通道：订阅者回调列表，线程安全。""" + + name = "in_process" + + def __init__(self) -> None: + self._lock = threading.RLock() + self._subscribers: List[Callable[[Dict[str, Any]], None]] = [] + self.published: List[Dict[str, Any]] = [] + + def subscribe(self, handler: Callable[[Dict[str, Any]], None]) -> Callable[[Dict[str, Any]], None]: + if not callable(handler): + raise TypeError("subscribe_broadcast(handler): handler 必须可调用") + with self._lock: + self._subscribers.append(handler) + return handler + + def unsubscribe(self, handler: Callable[[Dict[str, Any]], None]) -> bool: + with self._lock: + if handler in self._subscribers: + self._subscribers.remove(handler) + return True + return False + + @property + def subscriber_count(self) -> int: + with self._lock: + return len(self._subscribers) + + def publish(self, message: Dict[str, Any]) -> None: + with self._lock: + handlers = list(self._subscribers) + self.published.append(message) + if len(self.published) > LATENCY_SAMPLE_WINDOW: + self.published = self.published[-LATENCY_SAMPLE_WINDOW:] + errors: List[str] = [] + for handler in handlers: + try: + handler(message) + except Exception as exc: # 订阅者异常不影响其他订阅者 + errors.append("%s: %s" % (type(exc).__name__, exc)) + _LOG.warning("world_sync.broadcast 订阅者异常已吞掉: %s", exc) + if errors: + self._last_errors = errors + + def close(self) -> None: + with self._lock: + self._subscribers.clear() + + +class WebhookChannel(BaseChannel): + """HTTP webhook 通道：POST JSON，短超时，失败抛 :class:`ChannelUnavailable`。""" + + name = "webhook" + + def __init__(self, urls: Sequence[str], timeout_s: float = WEBHOOK_TIMEOUT_S) -> None: + self.urls: List[str] = [str(u) for u in (urls or []) if u] + if not self.urls: + raise ChannelUnavailable("WebhookChannel 需要至少一个 url") + self.timeout_s = float(timeout_s) + + def publish(self, message: Dict[str, Any]) -> None: + body = json.dumps(message, ensure_ascii=False, default=str).encode("utf-8") + failures: List[str] = [] + for url in self.urls: + try: + req = urllib.request.Request( + url, + data=body, + headers={ + "Content-Type": "application/json; charset=utf-8", + "X-Pbl-Event-Id": str(message.get("event_id", "")), + }, + method="POST", + ) + with urllib.request.urlopen(req, timeout=self.timeout_s) as resp: + code = int(getattr(resp, "status", 200) or 200) + if code >= 400: + failures.append("%s -> HTTP %d" % (url, code)) + except urllib.error.HTTPError as exc: + failures.append("%s -> HTTP %s" % (url, getattr(exc, "code", "?"))) + except Exception as exc: + failures.append("%s -> %s" % (url, exc)) + if failures: + raise ChannelUnavailable("; ".join(failures)) + + +class RedisPubSubChannel(BaseChannel): + """Redis pub/sub 通道（可选依赖：``redis``）。构造时即判定可用性。""" + + name = "redis_pubsub" + + def __init__(self, redis_url: Optional[str] = None, channel: str = DEFAULT_EVENT_CHANNEL) -> None: + try: + import redis # type: ignore + except Exception as exc: # pragma: no cover - 环境无 redis + raise ChannelUnavailable("redis 客户端不可用: %s" % exc) from exc + url = redis_url or os.environ.get("PBL_BROADCAST_REDIS_URL") or "" + if not url: + raise ChannelUnavailable("未配置 redis_url / PBL_BROADCAST_REDIS_URL") + try: + self._client = redis.Redis.from_url(url, socket_connect_timeout=WEBHOOK_TIMEOUT_S, + socket_timeout=WEBHOOK_TIMEOUT_S) + except Exception as exc: # pragma: no cover + raise ChannelUnavailable("redis 连接创建失败: %s" % exc) from exc + self.channel = channel + + def publish(self, message: Dict[str, Any]) -> None: + payload = json.dumps(message, ensure_ascii=False, default=str) + try: + self._client.publish(self.channel, payload) + except Exception as exc: + raise ChannelUnavailable("redis publish 失败: %s" % exc) from exc + + def close(self) -> None: + try: + self._client.close() + except Exception: + pass + + +# --------------------------------------------------------------------------- +# Broadcaster：异步分发 + 重试 + 指标 +# --------------------------------------------------------------------------- +class Broadcaster: + """事务提交后的异步广播器（fire-and-forget，绝不向调用方抛异常）。""" + + def __init__( + self, + channels: Optional[Iterable[BaseChannel]] = None, + queue_size: int = DEFAULT_QUEUE_SIZE, + retry_attempts: int = DEFAULT_RETRY_ATTEMPTS, + retry_backoff_s: float = DEFAULT_RETRY_BACKOFF_S, + enabled: bool = True, + ) -> None: + self._lock = threading.RLock() + self.channels: List[BaseChannel] = list(channels) if channels else [InProcessChannel()] + self.retry_attempts = int(retry_attempts) + self.retry_backoff_s = float(retry_backoff_s) + self.enabled = bool(enabled) + self._seq = 0 + self._latency: deque = deque(maxlen=LATENCY_SAMPLE_WINDOW) + self._stats: Dict[str, Any] = { + "published": 0, + "dropped_disabled": 0, + "failed": 0, + "retried": 0, + "direct_publish": 0, + "queued": 0, + "subscriber_errors": 0, + "max_latency_ms": 0.0, + } + self._last_error: Optional[str] = None + self._last_message: Optional[Dict[str, Any]] = None + self._queue: "queue.Queue" = queue.Queue(maxsize=max(1, int(queue_size))) + self._closed = False + self._worker: Optional[threading.Thread] = None + self._ensure_worker() + + # -- worker 生命周期 ------------------------------------------------- + def _ensure_worker(self) -> None: + with self._lock: + if self._closed or (self._worker is not None and self._worker.is_alive()): + return + self._worker = threading.Thread( + target=self._dispatch_loop, name="world-sync-broadcast", daemon=True + ) + self._worker.start() + + def _dispatch_loop(self) -> None: + while True: + item = self._queue.get(block=True) # 条件变量阻塞等待，非轮询 + try: + if item is _SENTINEL: + return + self._publish_now(item) + except Exception as exc: # pragma: no cover - _publish_now 已兜住 + _LOG.error("world_sync.broadcast 分发线程异常已吞掉: %s", exc) + finally: + self._queue.task_done() + + # -- 指标 ------------------------------------------------------------ + @property + def in_flight(self) -> int: + return self._queue.qsize() + + def stats(self) -> Dict[str, Any]: + with self._lock: + snap = dict(self._stats) + snap["in_flight"] = self._queue.qsize() + snap["latency_samples"] = len(self._latency) + snap["last_error"] = self._last_error + snap["enabled"] = self.enabled + snap["channels"] = [c.name for c in self.channels] + snap["p50_latency_ms"] = round(_percentile(self._latency, 50), 3) + snap["p99_latency_ms"] = round(_percentile(self._latency, 99), 3) + return snap + + def latency_percentile_ms(self, pct: float = 99.0) -> float: + with self._lock: + return round(_percentile(self._latency, pct), 3) + + @property + def last_message(self) -> Optional[Dict[str, Any]]: + with self._lock: + return self._last_message + + # -- 提交入口（调用方线程，O(1) 非阻塞） ----------------------------- + def submit(self, message: Dict[str, Any]) -> bool: + """把广播消息交给 dispatcher。永不阻塞、永不抛异常。""" + if not self.enabled: + with self._lock: + self._stats["dropped_disabled"] += 1 + return False + self._ensure_worker() + try: + self._queue.put_nowait(message) + with self._lock: + self._stats["queued"] += 1 + return True + except Exception: + # 队列满：退化为调用线程内直接发布（宁可不阻塞事务也要发出广播）； + # _publish_now 内部已 try/except，绝不外抛。 + with self._lock: + self._stats["direct_publish"] += 1 + self._publish_now(message) + return True + + # -- 实际发布（worker 线程） ------------------------------------------ + def _publish_now(self, message: Dict[str, Any]) -> bool: + commit_ts = float(message.get("commit_ts") or time.time()) + ok_any = False + errors: List[str] = [] + attempts = max(0, self.retry_attempts) + 1 + for channel in list(self.channels): + last_exc: Optional[BaseException] = None + sent = False + for attempt in range(attempts): + try: + channel.publish(message) + sent = True + break + except Exception as exc: + last_exc = exc + if attempt + 1 < attempts: + with self._lock: + self._stats["retried"] += 1 + if self.retry_backoff_s > 0: + time.sleep(self.retry_backoff_s * (attempt + 1)) + if sent: + ok_any = True + else: + errors.append("%s: %s" % (channel.name, last_exc)) + latency_ms = max(0.0, (time.time() - commit_ts) * 1000.0) + with self._lock: + self._latency.append(latency_ms) + self._last_message = message + if latency_ms > float(self._stats["max_latency_ms"]): + self._stats["max_latency_ms"] = round(latency_ms, 3) + if ok_any: + self._stats["published"] += 1 + else: + self._stats["failed"] += 1 + if errors: + self._last_error = "; ".join(errors) + if not ok_any: + # 关键不变式：事务已提交，广播失败只记日志，绝不回滚、绝不外抛。 + _LOG.error( + "world_sync.broadcast 广播失败（事务已提交，数据不受影响）" + " event_id=%s scene_id=%s entity_id=%s err=%s", + message.get("event_id"), message.get("scene_id"), + message.get("entity_id"), "; ".join(errors), + ) + return ok_any + + # -- 订阅（进程内通道） ------------------------------------------------ + def subscribe(self, handler: Callable[[Dict[str, Any]], None]) -> Callable[[Dict[str, Any]], None]: + chans = [c for c in self.channels if isinstance(c, InProcessChannel)] + if not chans: + chans = [InProcessChannel()] + with self._lock: + self.channels.append(chans[0]) + for ch in chans: + ch.subscribe(handler) + return handler + + def unsubscribe(self, handler: Callable[[Dict[str, Any]], None]) -> bool: + hit = False + for ch in [c for c in self.channels if isinstance(c, InProcessChannel)]: + hit = ch.unsubscribe(handler) or hit + return hit + + # -- 收尾 ------------------------------------------------------------ + def drain(self, timeout: float = 2.0) -> bool: + """等待已入队广播全部发送完成。 + + 用 ``Queue.all_tasks_done`` 条件变量阻塞等待（worker 调 ``task_done`` 时唤醒）， + **不是轮询**：没有任何 sleep 循环，也不查询数据库。 + """ + try: + with self._queue.all_tasks_done: + deadline = time.time() + float(timeout) + while self._queue.unfinished_tasks: + remaining = deadline - time.time() + if remaining <= 0: + return False + self._queue.all_tasks_done.wait(remaining) + return True + except Exception as exc: # pragma: no cover + _LOG.warning("world_sync.broadcast drain 异常: %s", exc) + return False + + def shutdown(self, timeout: float = 1.0) -> None: + with self._lock: + if self._closed: + return + self._closed = True + worker = self._worker + try: + self.drain(timeout=timeout) + except Exception: + pass + try: + self._queue.put(_SENTINEL) + except Exception: + pass + if worker is not None and worker.is_alive(): + worker.join(timeout=timeout) + for channel in list(self.channels): + try: + channel.close() + except Exception: + pass + + +# --------------------------------------------------------------------------- +# 消息构造 + post-commit hook +# --------------------------------------------------------------------------- +def build_broadcast_message( + write_result: Any, + commit_ts: Optional[float] = None, + channel: str = DEFAULT_EVENT_CHANNEL, + extra: Optional[Dict[str, Any]] = None, + context: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + """从 ``write_event_with_state`` 的返回值构造广播消息（永不抛异常）。 + + 至少包含 ``event_id / scene_id / entity_id / event_type / payload_summary``； + 兼容返回值为 dict、DictObject、或 ``{"event": {...}, "state": {...}}`` 嵌套形态。 + + :param context: 可选的原始调用上下文 ``{"event_data":..., "state_update":...}``。 + M11b-2 的返回值不含 ``event_type`` / ``payload``，由 post-commit hook 把调用 + 入参传进来补齐；取值优先级为「事务返回值 > 事件对象 > 状态对象 > 调用入参」。 + """ + ts = float(commit_ts) if commit_ts else time.time() + ctx = context or {} + event = _field(write_result, "event", None) or {} + state = _field(write_result, "state", None) or {} + src: List[Any] = [ + write_result, event, state, + ctx.get("event_data"), ctx.get("state_update"), + ] + + def pick(*names: str) -> Any: + for holder in src: + for nm in names: + val = _field(holder, nm) + if val is not None and val != "": + return val + return None + + payload = pick("payload", "event_payload", "data") + world_id = pick("world_id", "worldid") + scene_id = pick("scene_id", "sceneid") or world_id + message: Dict[str, Any] = { + "schema_version": "1", + "channel": channel, + "event_id": pick("event_id", "eventid", "id"), + "event_uid": pick("event_uid", "uid"), + "tenant_id": pick("tenant_id", "tenantid"), + "world_id": world_id, + "scene_id": scene_id, + "entity_id": pick("entity_id", "entityid"), + "event_type": pick("event_type", "eventtype") or "state.updated", + "session_id": pick("session_id", "sessionid"), + "source": pick("source") or "world_sync", + "state_version": pick("state_version", "version", "new_version"), + "payload_summary": _summarize_payload(payload), + "payload_digest": _payload_digest(payload), + "commit_ts": ts, + "commit_ts_text": _iso(ts), + "submit_ts": time.time(), + } + if extra: + try: + message.update(extra) + except Exception: + pass + return message + + +#: 模块级单例（宿主可 configure_broadcast 替换） +_BROADCASTER: Optional[Broadcaster] = None +_LOCK = threading.RLock() +_HOOK_INSTALLED = False + + +def get_broadcaster() -> Broadcaster: + """取（惰性创建）全局广播器。默认通道：进程内 + 可选 webhook（环境变量）。""" + global _BROADCASTER + with _LOCK: + if _BROADCASTER is None or getattr(_BROADCASTER, "_closed", False): + _BROADCASTER = _default_broadcaster() + return _BROADCASTER + + +def _default_broadcaster() -> Broadcaster: + channels: List[BaseChannel] = [InProcessChannel()] + urls = [u.strip() for u in (os.environ.get("PBL_BROADCAST_WEBHOOK_URLS") or "").split(",") if u.strip()] + if urls: + try: + channels.append(WebhookChannel(urls)) + except Exception as exc: # pragma: no cover + _LOG.warning("world_sync.broadcast webhook 通道构造失败，降级为进程内: %s", exc) + if _truthy_env("PBL_BROADCAST_REDIS_ENABLED", False): + try: + channels.append(RedisPubSubChannel()) + except Exception as exc: # pragma: no cover + _LOG.warning("world_sync.broadcast redis 通道不可用，已跳过: %s", exc) + enabled = not _truthy_env("PBL_BROADCAST_DISABLED", False) + return Broadcaster(channels=channels, enabled=enabled) + + +def configure_broadcast( + channels: Optional[Iterable[BaseChannel]] = None, + enabled: Optional[bool] = None, + queue_size: int = DEFAULT_QUEUE_SIZE, + retry_attempts: int = DEFAULT_RETRY_ATTEMPTS, + retry_backoff_s: float = DEFAULT_RETRY_BACKOFF_S, + webhook_urls: Optional[Sequence[str]] = None, + redis_url: Optional[str] = None, + keep_subscribers: bool = True, +) -> Broadcaster: + """替换全局广播器配置（宿主启动时调用一次即可）。""" + global _BROADCASTER + with _LOCK: + old = _BROADCASTER + chosen: Optional[List[BaseChannel]] = list(channels) if channels else None + if chosen is None: + chosen = [InProcessChannel()] + if webhook_urls: + try: + chosen.append(WebhookChannel(webhook_urls)) + except Exception as exc: + _LOG.warning("configure_broadcast webhook 通道失败: %s", exc) + if redis_url: + try: + chosen.append(RedisPubSubChannel(redis_url=redis_url)) + except Exception as exc: + _LOG.warning("configure_broadcast redis 通道失败: %s", exc) + elif webhook_urls: + chosen.extend(c for c in _try_webhook(webhook_urls)) + elif redis_url: + chosen.extend(c for c in _try_redis(redis_url)) + if keep_subscribers and old is not None: + for handler in _collect_subscribers(old): + for ch in chosen: + if isinstance(ch, InProcessChannel): + ch.subscribe(handler) + new = Broadcaster( + channels=chosen, + queue_size=queue_size, + retry_attempts=retry_attempts, + retry_backoff_s=retry_backoff_s, + enabled=old.enabled if enabled is None and old is not None else + (enabled if enabled is not None else True), + ) + _BROADCASTER = new + if old is not None and old is not new: + try: + old.shutdown(timeout=0.5) + except Exception: + pass + return new + + +def _try_webhook(urls: Sequence[str]) -> List[BaseChannel]: + try: + return [WebhookChannel(urls)] + except Exception as exc: + _LOG.warning("configure_broadcast webhook 通道失败: %s", exc) + return [] + + +def _try_redis(url: str) -> List[BaseChannel]: + try: + return [RedisPubSubChannel(redis_url=url)] + except Exception as exc: + _LOG.warning("configure_broadcast redis 通道失败: %s", exc) + return [] + + +def _collect_subscribers(broadcaster: Broadcaster) -> List[Callable[[Dict[str, Any]], None]]: + handlers: List[Callable[[Dict[str, Any]], None]] = [] + for ch in getattr(broadcaster, "channels", []): + if isinstance(ch, InProcessChannel): + handlers.extend(list(getattr(ch, "_subscribers", []))) + return handlers + + +def reset_broadcaster(broadcaster: Optional[Broadcaster] = None, shutdown_old: bool = True) -> Broadcaster: + """测试辅助：重置全局广播器。""" + global _BROADCASTER + with _LOCK: + old = _BROADCASTER + _BROADCASTER = broadcaster if broadcaster is not None else _default_broadcaster() + if old is not None and shutdown_old and broadcaster is None: + try: + old.shutdown(timeout=0.3) + except Exception: + pass + return _BROADCASTER + + +def subscribe_broadcast(handler: Callable[[Dict[str, Any]], None]) -> Callable[[Dict[str, Any]], None]: + """订阅进程内广播（返回 handler 本身，便于 unsubscribe）。""" + return get_broadcaster().subscribe(handler) + + +def unsubscribe_broadcast(handler: Callable[[Dict[str, Any]], None]) -> bool: + return get_broadcaster().unsubscribe(handler) + + +def broadcast_stats() -> Dict[str, Any]: + return get_broadcaster().stats() + + +def broadcast_last_message() -> Optional[Dict[str, Any]]: + return get_broadcaster().last_message + + +def broadcast_latency_percentile_ms(pct: float = 99.0) -> float: + return get_broadcaster().latency_percentile_ms(pct) + + +def drain_broadcast(timeout: float = 2.0) -> bool: + return get_broadcaster().drain(timeout=timeout) + + +def shutdown_broadcast(timeout: float = 1.0) -> None: + get_broadcaster().shutdown(timeout=timeout) + + +def emit_post_commit_broadcast( + write_result: Any, + commit_ts: Optional[float] = None, + channel: str = DEFAULT_EVENT_CHANNEL, + broadcaster: Optional[Broadcaster] = None, + context: Optional[Dict[str, Any]] = None, +) -> Optional[Dict[str, Any]]: + """**post-commit hook**：事务 COMMIT 之后触发广播。 + + 永不抛异常（广播失败只记日志），永不阻塞（入队即返回）。 + 返回广播消息 dict（便于日志/测试），广播被关闭时返回 ``None``。 + """ + try: + message = build_broadcast_message( + write_result, commit_ts=commit_ts, channel=channel, context=context) + target = broadcaster or get_broadcaster() + message["submit_ts"] = time.time() + message["latency_submit_ms"] = round( + max(0.0, (message["submit_ts"] - message["commit_ts"]) * 1000.0), 3 + ) + target.submit(message) + return message + except Exception as exc: # 铁律：广播侧任何异常都不能影响已提交事务 + _LOG.error("world_sync.broadcast 触发广播异常已吞掉（事务已提交）: %s", exc) + return None + + +def _call_context(args: Sequence[Any], kwargs: Dict[str, Any]) -> Dict[str, Any]: + """从 write_event_with_state 的调用入参提取 event_data / state_update 上下文。""" + ctx: Dict[str, Any] = {} + try: + names = ("event_data", "state_update") + for i, val in enumerate(args): + if i < len(names): + ctx[names[i]] = val + for nm in names: + if nm in kwargs and kwargs[nm] is not None: + ctx[nm] = kwargs[nm] + except Exception: + pass + return ctx + + +def wrap_with_post_commit_broadcast( + write_fn: Callable[..., Any], + channel: str = DEFAULT_EVENT_CHANNEL, + broadcaster: Optional["Broadcaster"] = None, +) -> Callable[..., Any]: + """把 M11b-2 的 ``write_event_with_state`` 包成「COMMIT 后广播」版本。 + + ``write_fn`` 返回即代表事务已 COMMIT（M11b-2 在单事务内提交后返回）； + 若 ``write_fn`` 抛异常（回滚/冲突），**不广播**，异常原样上抛。 + """ + if not callable(write_fn): + raise TypeError("wrap_with_post_commit_broadcast(write_fn): 需要可调用对象") + + def wrapper(*args: Any, **kwargs: Any) -> Any: + result = write_fn(*args, **kwargs) # 事务在此完成 COMMIT（或抛异常回滚） + # COMMIT 已返回 —— 此刻记录 commit_ts，作为 <=300ms(P99) 广播延迟的起点 + commit_ts = time.time() + emit_post_commit_broadcast( + result, commit_ts=commit_ts, channel=channel, + context=_call_context(args, kwargs), + broadcaster=broadcaster, + ) + return result + + try: + wrapper.__name__ = getattr(write_fn, "__name__", "write_event_with_state") + wrapper.__doc__ = getattr(write_fn, "__doc__", None) + wrapper.__module__ = getattr(write_fn, "__module__", "world_sync") + except Exception: + pass + setattr(wrapper, "_m11b3_broadcast_wrapped", True) + setattr(wrapper, "_m11b3_underlying", write_fn) + return wrapper + + +def _refresh_env_reference(server_env: Any, wrapped: Callable[..., Any], + only_if: Optional[Callable[..., Any]] = None) -> None: + """把包装后的 write_event_with_state 同步到 ServerEnv（三处接线之一）。""" + env = server_env + if env is None: + try: + from ahserver.serverenv import ServerEnv # 宿主环境才有 + env = ServerEnv() + except Exception: + env = None + if env is None: + return + try: + if only_if is None or getattr(env, "write_event_with_state", None) is only_if: + env.write_event_with_state = wrapped + except Exception as exc: # pragma: no cover + _LOG.debug("world_sync.broadcast ServerEnv 刷新跳过: %s", exc) + + +def is_broadcast_hook_installed() -> bool: + return bool(_HOOK_INSTALLED) + + +def install_broadcast_hook(server_env: Any = None, force: bool = False) -> Callable[..., Any]: + """把广播 post-commit hook 装到 M11b-2 的 ``write_event_with_state`` 上。 + + 同步刷新三处引用（实现模块属性 / 包命名空间 / ServerEnv 登记），保证宿主经 + 任何一条路径调用都会触发广播。幂等：重复调用只包一次（``force=True`` 强包）。 + """ + global _HOOK_INSTALLED + from . import pbl_runtime_tx_sqlor as _txs + + original = getattr(_txs, "write_event_with_state") + if getattr(original, "_m11b3_broadcast_wrapped", False) and not force: + _HOOK_INSTALLED = True + _refresh_env_reference(server_env, original) + return original + + wrapped = wrap_with_post_commit_broadcast(original) + _txs.write_event_with_state = wrapped + + import sys as _sys + pkg = _sys.modules.get("world_sync") + if pkg is not None and getattr(pkg, "write_event_with_state", None) is original: + try: + pkg.write_event_with_state = wrapped # type: ignore[attr-defined] + except Exception: + pass + + _refresh_env_reference(server_env, wrapped, only_if=original) + + _HOOK_INSTALLED = True + _LOG.info("world_sync.broadcast post-commit hook 已安装（异步广播，SLO=%.0fms P99）", + BROADCAST_LATENCY_SLO_MS) + return wrapped