433 lines
16 KiB
Python
433 lines
16 KiB
Python
# -*- 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
|