From fc0ac8118a12fb53d5b22ba9aa98cdca5c4d98a8 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 11:49:28 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- world_sync/__init__.py | 51 +++ world_sync/init.py | 50 +++ world_sync/pbl_runtime_latency.py | 188 +++++++++++ world_sync/pbl_runtime_polling.py | 544 ++++++++++++++++++++++++++++++ 4 files changed, 833 insertions(+) create mode 100644 world_sync/pbl_runtime_latency.py create mode 100644 world_sync/pbl_runtime_polling.py diff --git a/world_sync/__init__.py b/world_sync/__init__.py index 23c59a6..2a56539 100644 --- a/world_sync/__init__.py +++ b/world_sync/__init__.py @@ -150,6 +150,34 @@ from .pbl_runtime_broadcast import ( wrap_with_post_commit_broadcast, ) +# --- M11b-4 3s 轮询兜底 + 端到端时延验证（真源：pbl_runtime_polling.py / pbl_runtime_latency.py） --- +from .pbl_runtime_polling import ( + FALLBACK_SLO_MS, + POLL_INTERVAL_S, + POLL_LIMIT, + POLL_TIMEOUT_S, + SEEN_IDS_IN_SQL, + PollingFallback, + build_poll_sql, + configure_fallback, + explain_poll_query as polling_explain_poll_query, + fallback_stats, + get_fallback, + normalize_watermark, + plan_uses_scene_index, + poll_since, + reset_fallback, + run_fallback, + wait_event, +) +from .pbl_runtime_latency import ( + E2E_LATENCY_SLO_MS, + latency_report, + measure_once, + run_chain, + summarize_latency, +) + #: 契约解析优先级键（与 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 分叉" @@ -247,6 +275,29 @@ __all__ = [ "broadcast_latency_percentile_ms", "drain_broadcast", "shutdown_broadcast", + # M11b-4 3s 轮询兜底（消费端侧，只读，不侵入写入端） + "POLL_INTERVAL_S", + "POLL_TIMEOUT_S", + "POLL_LIMIT", + "FALLBACK_SLO_MS", + "SEEN_IDS_IN_SQL", + "PollingFallback", + "build_poll_sql", + "configure_fallback", + "fallback_stats", + "get_fallback", + "normalize_watermark", + "plan_uses_scene_index", + "poll_since", + "reset_fallback", + "run_fallback", + "wait_event", + # M11b-4 端到端时延基准 + "E2E_LATENCY_SLO_MS", + "measure_once", + "run_chain", + "summarize_latency", + "latency_report", # 宿主基础包可用性诊断 "HOST_DEPS_AVAILABLE", "HOST_DEPS_ERROR", diff --git a/world_sync/init.py b/world_sync/init.py index 9006445..003097d 100644 --- a/world_sync/init.py +++ b/world_sync/init.py @@ -178,6 +178,32 @@ from .pbl_runtime_broadcast import ( wrap_with_post_commit_broadcast, ) +# --- M11b-4: 3s 轮询兜底 + 端到端时延验证 注册接线 --- +from .pbl_runtime_polling import ( + FALLBACK_SLO_MS, + POLL_INTERVAL_S, + POLL_LIMIT, + POLL_TIMEOUT_S, + PollingFallback, + build_poll_sql, + configure_fallback, + fallback_stats, + get_fallback, + normalize_watermark, + plan_uses_scene_index, + poll_since, + reset_fallback, + run_fallback, + wait_event, +) +from .pbl_runtime_latency import ( + E2E_LATENCY_SLO_MS, + latency_report, + measure_once, + run_chain, + summarize_latency, +) + MODULE_NAME = 'world_sync' MODULE_VERSION = '0.3.0-m11b3' @@ -526,6 +552,30 @@ def load_world_sync(server_env=None, conn_factory=None): env.drain_broadcast = drain_broadcast env.shutdown_broadcast = shutdown_broadcast env.reset_broadcast = reset_broadcaster + # --- M11b-4: 3s 轮询兜底（消费端侧，只读；不侵入 M11b-2 写入端） --- + env.POLL_INTERVAL_S = POLL_INTERVAL_S + env.POLL_TIMEOUT_S = POLL_TIMEOUT_S + env.POLL_LIMIT = POLL_LIMIT + env.FALLBACK_SLO_MS = FALLBACK_SLO_MS + env.E2E_LATENCY_SLO_MS = E2E_LATENCY_SLO_MS + env.PollingFallback = PollingFallback + env.build_poll_sql = build_poll_sql + env.normalize_watermark = normalize_watermark + env.plan_uses_scene_index = plan_uses_scene_index + env.configure_fallback = configure_fallback + env.get_fallback = get_fallback + env.reset_fallback = reset_fallback + env.fallback_stats = fallback_stats + env.poll_since = poll_since + env.run_fallback = run_fallback + env.wait_event = wait_event + # 消费端订阅侧标准接线：广播到点 → 推进兜底水位线 + 停止轮询 + env.notify_broadcast = get_fallback().notify_broadcast + # M11b-4 端到端时延基准 + env.measure_once = measure_once + env.run_chain = run_chain + env.summarize_latency = summarize_latency + env.latency_report = latency_report # post-commit hook：把 write_event_with_state 包成「COMMIT 后触发广播」版本， # 并同步刷新三处引用（pbl_runtime_tx_sqlor / world_sync 包命名空间 / env）。 env.write_event_with_state = install_broadcast_hook(env) diff --git a/world_sync/pbl_runtime_latency.py b/world_sync/pbl_runtime_latency.py new file mode 100644 index 0000000..54c104c --- /dev/null +++ b/world_sync/pbl_runtime_latency.py @@ -0,0 +1,188 @@ +# -*- coding: utf-8 -*- +"""M11b-4：端到端时延基准（adjudication → write → COMMIT → broadcast → consumer）。 + +链路分段（对应任务验收「P99 ≤ 1s」）:: + + t0 裁决触发（run_chain 进入，模拟裁决服务决定下发事件） + t1 write_event_with_state 返回（内部已完成事件+状态单事务 COMMIT，M11b-2） + t2 post-commit hook 广播发出完成（M11b-3 install_broadcast_hook 包装后的函数） + t3 消费端回调收到广播消息（感知） + + e2e_latency = t3 - t0 + +本模块 **只组合调用** M11b-2/M11b-3 已交付的公开 API，不修改其核心逻辑： +- 写入：``world_sync.write_event_with_state``（若宿主已 ``load_world_sync()`` 则 + 该引用已被 ``install_broadcast_hook`` 包成 COMMIT 后自动广播版本）； +- 广播：``subscribe_broadcast`` / ``drain_broadcast`` / ``broadcast_stats``。 + +消费端感知判定：优先「广播回调到达」（真实链路）；若广播丢失，则由 +:mod:`world_sync.pbl_runtime_polling` 的 3s 轮询兜底补漏到达（此时 e2e 必然 +>1s，报告会把它归入 ``fallback_recovered`` 并单独统计，供瓶颈分析）。 +""" + +import time +from typing import Any, Callable, Dict, List, Optional + +from .pbl_runtime_broadcast import ( + broadcast_stats, + drain_broadcast, + get_broadcaster, + subscribe_broadcast, + unsubscribe_broadcast, +) +from .pbl_runtime_errors import TxConfigError +from .pbl_runtime_polling import FALLBACK_SLO_MS, get_fallback +from .pbl_runtime_sql import percentile + +#: 端到端时延 SLO（毫秒）——任务规定 P99 ≤ 1s +E2E_LATENCY_SLO_MS = 1000.0 + +__all__ = [ + "E2E_LATENCY_SLO_MS", + "FALLBACK_SLO_MS", + "measure_once", + "run_chain", + "summarize_latency", + "latency_report", +] + + +def _now_ms() -> float: + return time.monotonic() * 1000.0 + + +def measure_once(write_fn: Callable[..., Any], event_data: Dict[str, Any], + state_update: Optional[Dict[str, Any]] = None, + wait_ms: float = 1000.0, + poll_fallback: bool = False) -> Dict[str, Any]: + """跑一条事件的端到端测量。 + + :param write_fn: 写入函数（应为已安装 post-commit hook 的 + ``write_event_with_state``；纯写入端时延也能测，广播段记 None）。 + :param poll_fallback: True 时广播丢失不判 fail，改由轮询兜底补漏计时。 + """ + scene_id = str(event_data.get("scene_id") or event_data.get("world_id") or "") + fb = get_fallback() + holder: Dict[str, Any] = {} + t_done = {"v": None} + + def _on_msg(msg: Any) -> None: + if holder.get("t_recv_ms") is None: + holder["t_recv_ms"] = _now_ms() + holder["msg"] = msg + # 广播正常路径：推进兜底水位线（消费端标准接线姿势） + try: + fb.notify_broadcast(msg) + except Exception: + pass + + holder["t_recv_ms"] = None + sub = subscribe_broadcast(_on_msg) + try: + t0 = _now_ms() + try: + result = write_fn(event_data, state_update or {}) + finally: + t1 = _now_ms() + # post-commit hook 在 write_fn 内部同步/异步发出；drain 确保已派发 + try: + drain_broadcast(timeout=0.5) + except Exception: + pass + t2 = _now_ms() + # 等消费端感知（广播回调） + deadline = _now_ms() + wait_ms + while holder.get("t_recv_ms") is None and _now_ms() < deadline: + time.sleep(0.001) + t3 = holder.get("t_recv_ms") + via = "broadcast" + if t3 is None and poll_fallback: + # 广播丢失 → 3s 轮询兜底补漏（真实节奏：interval=3s） + eid = None + if isinstance(result, dict): + eid = result.get("event_id") or result.get("id") + w = fb.wait_event(scene_id, str(eid or ""), timeout=FALLBACK_SLO_MS / 1000.0) + via = "polling_fallback" + t3 = _now_ms() if w.get("found") else None + e2e = (t3 - t0) if t3 is not None else None + return { + "scene_id": scene_id, + "event_id": (result.get("event_id") if isinstance(result, dict) else None), + "t_write_ms": round(t1 - t0, 3), + "t_broadcast_emit_ms": round(t2 - t1, 3), + "t_consume_ms": round(t3 - t1, 3) if t3 is not None else None, + "e2e_ms": round(e2e, 3) if e2e is not None else None, + "via": via, + "perceived": t3 is not None, + } + finally: + unsubscribe_broadcast(sub) + + +def run_chain(write_fn: Callable[..., Any], events: List[Dict[str, Any]], + state_builder: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, + wait_ms: float = 1000.0, poll_fallback: bool = False) -> List[Dict[str, Any]]: + """批量端到端测量（裁决触发→…→消费端感知），返回逐条样本。""" + samples: List[Dict[str, Any]] = [] + for ev in events: + st = state_builder(ev) if state_builder else {} + samples.append(measure_once(write_fn, ev, st, wait_ms=wait_ms, + poll_fallback=poll_fallback)) + return samples + + +def summarize_latency(samples: List[Dict[str, Any]], + slo_ms: float = E2E_LATENCY_SLO_MS) -> Dict[str, Any]: + """汇总：P50/P90/P99/max + SLO 判定 + 分段瓶颈定位。""" + e2e = [s["e2e_ms"] for s in samples if s.get("e2e_ms") is not None] + write_seg = [s["t_write_ms"] for s in samples if s.get("t_write_ms") is not None] + emit_seg = [s["t_broadcast_emit_ms"] for s in samples + if s.get("t_broadcast_emit_ms") is not None] + consume_seg = [s["t_consume_ms"] for s in samples if s.get("t_consume_ms") is not None] + p99 = round(percentile(e2e, 99), 3) if e2e else None + via_broadcast = sum(1 for s in samples if s.get("via") == "broadcast") + via_polling = sum(1 for s in samples if s.get("via") == "polling_fallback") + lost = sum(1 for s in samples if not s.get("perceived")) + # 瓶颈定位：P99 分段最大者 + seg_p99 = { + "write_commit": round(percentile(write_seg, 99), 3) if write_seg else 0.0, + "broadcast_emit": round(percentile(emit_seg, 99), 3) if emit_seg else 0.0, + "consume_wait": round(percentile(consume_seg, 99), 3) if consume_seg else 0.0, + } + bottleneck = max(seg_p99.items(), key=lambda kv: kv[1])[0] if seg_p99 else None + return { + "n": len(samples), + "perceived": len(samples) - lost, + "lost": lost, + "via_broadcast": via_broadcast, + "via_polling_fallback": via_polling, + "p50_ms": round(percentile(e2e, 50), 3) if e2e else None, + "p90_ms": round(percentile(e2e, 90), 3) if e2e else None, + "p99_ms": p99, + "max_ms": round(max(e2e), 3) if e2e else None, + "segment_p99_ms": seg_p99, + "bottleneck_segment": bottleneck, + "slo_ms": slo_ms, + "slo_passed": (p99 is not None and p99 <= slo_ms), + "broadcast_stats": broadcast_stats(), + } + + +def latency_report(write_fn: Callable[..., Any], events: List[Dict[str, Any]], + state_builder: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]] = None, + slo_ms: float = E2E_LATENCY_SLO_MS, + wait_ms: float = 1000.0, + poll_fallback: bool = False) -> Dict[str, Any]: + """一键基准：跑链路 + 汇总 + 结论（供集成测试 / scripts/m11b4_e2e_bench.py 复用）。""" + if write_fn is None or not callable(write_fn): + raise TxConfigError("latency_report: write_fn must be a callable " + "(world_sync.write_event_with_state)") + samples = run_chain(write_fn, events, state_builder=state_builder, + wait_ms=wait_ms, poll_fallback=poll_fallback) + summary = summarize_latency(samples, slo_ms=slo_ms) + summary["samples"] = samples + summary["verdict"] = ("PASS" if summary["slo_passed"] else + "FAIL: P99 %.1fms > SLO %.0fms, bottleneck=%s" + % (summary["p99_ms"] or -1, slo_ms, + summary["bottleneck_segment"])) + return summary diff --git a/world_sync/pbl_runtime_polling.py b/world_sync/pbl_runtime_polling.py new file mode 100644 index 0000000..df2e028 --- /dev/null +++ b/world_sync/pbl_runtime_polling.py @@ -0,0 +1,544 @@ +# -*- coding: utf-8 -*- +"""M11b-4：广播不可靠时的 **3 秒轮询兜底**（polling fallback，消费端侧）。 + +设计定位（与 M11b-1/2/3 的关系） +================================ +- M11b-1（表/索引）：补漏查询 **只按 ``scene_id + created_at`` 过滤**，命中 + ``models/pbl_runtime_event.json`` 的复合索引 ``ix_scene_created(scene_id, + created_at)``，不做全表扫描；MySQL/MariaDB 侧用 :func:`explain_poll_query` + EXPLAIN 验证（:func:`plan_uses_scene_index` 判定）。 +- M11b-2（单事务写入）：本模块 **零侵入** —— 不 import、不包装、不修改 + ``write_event_with_state``；只读 ``pbl_runtime_event`` 表。 +- M11b-3（提交后广播）：本模块是广播的 **对偶兜底**。广播正常时消费端调用 + :meth:`PollingFallback.notify_broadcast` 记到点，轮询循环立即停止；广播丢失时 + 由 3s 定时器主动查库补漏（水位线之后的行 = 漏网事件）。 + +行为契约（对应任务验收标准） +============================ +1. 轮询间隔 :data:`POLL_INTERVAL_S` = 3.0s（可配），总超时 :data:`POLL_TIMEOUT_S`； + 「收到广播后停止，或超时后停止」（见 :meth:`PollingFallback.run` 的 + ``stopped_by`` ∈ {``broadcast_received``, ``event_matched``, ``timeout``}）。 +2. 广播丢失场景：下一轮轮询（≤ interval + 单次查询耗时）内把水位线之后的事件全部 + 补回（:meth:`PollingFallback.poll_once` 的 ``recovered``）。 +3. 独立封装：类 :class:`PollingFallback` + 进程级单例门面 + （:func:`get_fallback` / :func:`configure_fallback` / :func:`reset_fallback` / + :func:`fallback_stats` / :func:`wait_event`），消费端按需调用。 + +两个必须守住的正确性细节（写在本模块而非靠调用方自觉） +====================================================== +A. **时间格式必须与库内一致**。``write_event_with_state`` 落库的 ``created_at`` 是 + :func:`world_sync.pbl_runtime_tx.utc_now_text` 生成的 ``'YYYY-MM-DD HH:MM:SS'`` + （空格分隔）。M11b-3 广播消息里的 ``commit_ts_text`` 是 ISO ``'...T...Z'`` 格式， + 字典序恒大于空格格式（``'T'``=0x54 > ``' '``=0x20），**若误用它推进水位线会把 + 同场景后续事件全部跳过（静默丢事件）**。因此 :meth:`set_watermark` 只接受空格 + 格式时间戳，其余一律忽略（:func:`normalize_watermark`）。 +B. **秒级时间戳会撞秒**。同一秒内多条事件 ``created_at`` 相同，用 ``>`` 比较会漏。 + 因此查询用 ``created_at >= watermark`` + ``event_uid NOT IN (已投递集合)`` 双保险： + 同秒事件不丢、重复行不重复投递。已投递集合以 ``event_uid`` 为键（广播消息与写入 + 返回值都含该键；表列 ``id`` 是物理主键、``event_uid`` 才是跨系统追溯口径）。 +""" + +import re +import threading +import time +from typing import Any, Callable, Dict, List, Optional, Sequence + +from . import pbl_runtime_sql as ps +from .pbl_runtime_errors import TxConfigError +from .pbl_runtime_tx_env import resolve_conn_factory + +#: 兜底轮询间隔（秒）——任务规定 3s +POLL_INTERVAL_S = 3.0 +#: 单次兜底循环总超时（秒）：超时后定时器自行停止 +POLL_TIMEOUT_S = 30.0 +#: 单次补漏查询 LIMIT +POLL_LIMIT = 500 +#: 消费端感知 SLO（毫秒）：广播丢失时兜底须在 1 个轮询周期内补漏 +FALLBACK_SLO_MS = 3000.0 +#: 去重集合参与 SQL 的最大长度（防止 NOT IN 参数爆炸） +SEEN_IDS_IN_SQL = 200 + +_WATERMARK_RE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(\.\d{1,6})?$") + +__all__ = [ + "POLL_INTERVAL_S", + "POLL_TIMEOUT_S", + "POLL_LIMIT", + "FALLBACK_SLO_MS", + "SEEN_IDS_IN_SQL", + "PollingFallback", + "build_poll_sql", + "configure_fallback", + "explain_poll_query", + "fallback_stats", + "get_fallback", + "normalize_watermark", + "plan_uses_scene_index", + "poll_since", + "reset_fallback", + "run_fallback", + "wait_event", +] + + +# ---------------------------------------------------------------- 基础工具 +def _field(obj: Any, name: str, default: Any = None) -> Any: + """兼容 dict / DictObject / 行对象取字段（与 M11b-3 同口径，独立实现不互相依赖）。""" + if obj is None: + return default + if isinstance(obj, dict): + return obj.get(name, default) + try: + val = getattr(obj, name) + except Exception: + return default + return default if val is None else val + + +def normalize_watermark(value: Any) -> str: + """把 ``created_at`` 归一为库内口径 ``'YYYY-MM-DD HH:MM:SS[.ffffff]'``。 + + 只接受空格分隔格式（:func:`world_sync.pbl_runtime_tx.utc_now_text` 的产物）。 + ISO ``T``/``Z`` 格式（广播消息 ``commit_ts_text``）一律返回 ``''`` —— 见模块 + docstring 细节 A：混用会静默跳过后续事件。datetime 对象按 UTC 格式化。 + """ + if value is None: + return "" + if hasattr(value, "strftime") and not isinstance(value, str): + try: + return value.strftime("%Y-%m-%d %H:%M:%S") + except Exception: + return "" + text = str(value).strip() + if not text: + return "" + if _WATERMARK_RE.match(text): + return text + return "" + + +def _row_uid(row: Any) -> str: + return str(_field(row, "event_uid") or _field(row, "event_id") + or _field(row, "id") or "") + + +def _row_created(row: Any) -> str: + return normalize_watermark(_field(row, "created_at")) + + +def _row_to_dict(row: Any) -> Dict[str, Any]: + if isinstance(row, dict): + return dict(row) + return dict(row) + + +# ---------------------------------------------------------------- SQL 构造 +def build_poll_sql(dialect: str, watermark: str, scene_id: str = ":scene", + seen_uids: Optional[Sequence[Any]] = None, + limit: int = POLL_LIMIT): + """构造补漏查询，返回 ``(sql, params)``。 + + **索引契约**：WHERE 前缀必须是 ``scene_id = ? AND created_at >= ?``，与复合索引 + ``ix_scene_created(scene_id, created_at)`` 列序一致（等值列在前、范围列在后）； + ``ORDER BY created_at ASC`` 走同一索引，避免 filesort 扩大扫描面。 + ``LIMIT`` 为消毒后的整数直接内插（方言差异不涉及占位）。 + + ``scene_id`` 传默认哨兵字符串时仅作 SQL 文本预览/EXPLAIN 用，真实调用由 + :meth:`PollingFallback.query_since` 覆盖 ``params[0]``。 + """ + ph = ps.placeholder + limit = max(1, int(limit)) + sql = ( + "SELECT id, event_uid, tenant_id, world_id, scene_id, entity_id, " + "event_type, seq, seq_no, state_version, created_at " + "FROM %s WHERE scene_id = %s AND created_at >= %s" + % (ps.EVENT_TABLE, ph(dialect, 0), ph(dialect, 1)) + ) + params: List[Any] = [scene_id, normalize_watermark(watermark)] + uids = [str(u) for u in (seen_uids or []) if u is not None and str(u) != ""] + uids = uids[-SEEN_IDS_IN_SQL:] + if uids: + marks = ", ".join(ph(dialect, 2 + k) for k in range(len(uids))) + sql += " AND (event_uid IS NULL OR event_uid NOT IN (%s))" % marks + params.extend(uids) + sql += " ORDER BY created_at ASC LIMIT %d" % limit + return sql, params + + +# ---------------------------------------------------------------- 核心类 +class PollingFallback: + """3 秒轮询兜底器（消费端 / 订阅方侧，只读，不侵入写入端）。 + + :param conn_factory: DB-API 连接工厂；None 时走模块统一解析链 + （显式注入 → ServerEnv 登记 → 环境变量 ``PBL_RUNTIME_DB_URL`` → ``conf/db.json``）。 + :param interval: 轮询间隔秒（默认 :data:`POLL_INTERVAL_S` = 3s）。 + :param timeout: 单次兜底循环总超时秒（默认 :data:`POLL_TIMEOUT_S`）。 + """ + + def __init__(self, conn_factory: Optional[Callable[[], Any]] = None, + interval: float = POLL_INTERVAL_S, timeout: float = POLL_TIMEOUT_S, + limit: int = POLL_LIMIT): + self._conn_factory = conn_factory + self.interval = float(interval) + self.timeout = float(timeout) + self.limit = int(limit) + self._lock = threading.Lock() + # scene_id -> {"watermark": str, "notified": bool, "last_notify_ts": float} + self._scenes: Dict[str, Dict[str, Any]] = {} + # scene_id -> [event_uid]（保序，用于 NOT IN 去重） + self._seen: Dict[str, List[str]] = {} + self._counters = { + "polls": 0, "queries": 0, "recovered": 0, "duplicated": 0, + "missed_by_broadcast": 0, "timeouts": 0, "stops_on_notify": 0, + } + + # ---------------- 状态维护 ---------------- + def _scene_state(self, scene_id: str) -> Dict[str, Any]: + st = self._scenes.get(scene_id) + if st is None: + st = {"watermark": "", "notified": False, "last_notify_ts": 0.0} + self._scenes[scene_id] = st + return st + + def set_watermark(self, scene_id: str, created_at: Any) -> bool: + """消费端感知到事件后推进水位线（单调不减）。格式不合法则忽略并返回 False。""" + ts = normalize_watermark(created_at) + if not ts: + return False + with self._lock: + st = self._scene_state(str(scene_id or "")) + if ts > st["watermark"]: + st["watermark"] = ts + return True + return False + + def watermark(self, scene_id: str) -> str: + with self._lock: + return self._scenes.get(str(scene_id or ""), {}).get("watermark", "") + + def note_event_seen(self, scene_id: str, event_uid: Any) -> None: + """登记已投递事件（event_uid 口径），供轮询二次去重。""" + if event_uid is None or str(event_uid) == "": + return + uid = str(event_uid) + with self._lock: + lst = self._seen.setdefault(str(scene_id or ""), []) + if uid in lst: + self._counters["duplicated"] += 1 + return + lst.append(uid) + if len(lst) > 4 * SEEN_IDS_IN_SQL: + del lst[:len(lst) - 2 * SEEN_IDS_IN_SQL] + + def seen_uids(self, scene_id: str) -> List[str]: + with self._lock: + return list(self._seen.get(str(scene_id or ""), [])) + + def notify_broadcast(self, message: Any, created_at: Any = None) -> Dict[str, Any]: + """广播正常路径：消费端订阅回调里调用 → 记到点（轮询循环随即停止）。 + + :param message: M11b-3 广播消息 dict（含 ``scene_id`` / ``event_uid``）。 + :param created_at: 可选，写入返回值里的库内口径 ``created_at``。广播消息 + 本身不带该格式（只有 ISO ``commit_ts_text``，见模块 docstring 细节 A）， + 因此推荐消费端把 ``write_event_with_state`` 返回的 ``created_at`` 传进来， + 让水位线随广播精确推进。 + """ + scene_id = str(_field(message, "scene_id") or "") + uid = _field(message, "event_uid") or _field(message, "event_id") + advanced = self.set_watermark(scene_id, created_at) + self.note_event_seen(scene_id, uid) + with self._lock: + st = self._scene_state(scene_id) + st["notified"] = True + st["last_notify_ts"] = time.time() + info = {"scene_id": scene_id, "notified": True, + "watermark": st["watermark"], "watermark_advanced": advanced} + return info + + def was_notified(self, scene_id: str) -> bool: + with self._lock: + return bool(self._scenes.get(str(scene_id or ""), {}).get("notified")) + + def clear_notified(self, scene_id: str) -> None: + """重置到点标记（下一轮事件等待前调用，避免上一轮的 notified 短路本轮）。""" + with self._lock: + self._scene_state(str(scene_id or ""))["notified"] = False + + # ---------------- 查询 ---------------- + def _connect(self): + try: + factory = resolve_conn_factory(self._conn_factory) + except Exception as exc: # fail-closed：没有连接工厂就别假装查过 + raise TxConfigError("polling fallback: conn factory unavailable: %r" % (exc,)) + if not callable(factory): + raise TxConfigError("polling fallback: resolved conn factory is not callable") + return factory() + + def query_since(self, scene_id: str, watermark: str, + seen_uids: Optional[Sequence[Any]] = None) -> List[Dict[str, Any]]: + """执行一次补漏查询（公开方法，供基准/测试直接调用）。返回行 dict 列表。""" + conn = self._connect() + try: + dialect = ps.detect_dialect(conn) + sql, params = build_poll_sql(dialect, watermark, scene_id, seen_uids, + self.limit) + params[0] = str(scene_id) + cur = conn.cursor() + try: + cur.execute(sql, params) + cols = [d[0] for d in (cur.description or [])] + rows: List[Dict[str, Any]] = [] + for raw in cur.fetchall() or []: + if isinstance(raw, dict): + rows.append(dict(raw)) + else: + rows.append(dict(zip(cols, list(raw)))) + finally: + try: + cur.close() + except Exception: + pass + try: + conn.commit() + except Exception: + pass + return rows + finally: + try: + conn.close() + except Exception: + pass + + def explain_poll_query(self, scene_id: str, watermark: str = "") -> List[Dict[str, Any]]: + """对补漏查询执行 ``EXPLAIN``（MySQL/MariaDB 有意义），返回计划行。""" + conn = self._connect() + try: + dialect = ps.detect_dialect(conn) + sql, params = build_poll_sql(dialect, watermark, scene_id, None, self.limit) + cur = conn.cursor() + try: + cur.execute("EXPLAIN " + sql, params) + cols = [d[0] for d in (cur.description or [])] + out: List[Dict[str, Any]] = [] + for raw in cur.fetchall() or []: + out.append(dict(raw) if isinstance(raw, dict) + else dict(zip(cols, list(raw)))) + return out + finally: + try: + cur.close() + except Exception: + pass + finally: + try: + conn.close() + except Exception: + pass + + def poll_once(self, scene_id: str, wait_for: Optional[str] = None) -> Dict[str, Any]: + """一轮补漏：查水位线之后的行，扣除已投递集合 = 广播漏网事件。 + + :param wait_for: 可选，等待特定 ``event_uid``；出现则 ``found=True``。 + """ + scene_id = str(scene_id or "") + with self._lock: + st = self._scene_state(scene_id) + watermark = st["watermark"] + seen = list(self._seen.get(scene_id) or []) + self._counters["polls"] += 1 + rows = self.query_since(scene_id, watermark, seen) + seen_set = set(seen) + recovered: List[Dict[str, Any]] = [] + found = False + max_created = watermark + for row in rows: + uid = _row_uid(row) + if uid and uid in seen_set: + continue + recovered.append(row) + if uid: + seen_set.add(uid) + self.note_event_seen(scene_id, uid) + created = _row_created(row) + if created and created > max_created: + max_created = created + if wait_for and uid == str(wait_for): + found = True + if max_created: + self.set_watermark(scene_id, max_created) + with self._lock: + self._counters["queries"] += 1 + self._counters["recovered"] += len(recovered) + if recovered: + self._counters["missed_by_broadcast"] += len(recovered) + return {"scene_id": scene_id, "watermark": watermark, "rows": rows, + "recovered": recovered, "recovered_count": len(recovered), + "found": found, "notified": self.was_notified(scene_id)} + + # ---------------- 兜底循环 ---------------- + def run(self, scene_id: str, wait_for: Optional[str] = None, + timeout: Optional[float] = None, interval: Optional[float] = None, + on_recovered: Optional[Callable[[Dict[str, Any]], None]] = None, + sleep: Callable[[float], None] = time.sleep) -> Dict[str, Any]: + """3s 轮询兜底循环：**收到广播即停；等不到广播时按周期查库补漏；超时也停**。 + + 消费端通常在「广播等待超时」后把场景交给定时器线程调用本方法。 + ``sleep`` 可注入（测试用假时钟加速，不改变 3s 语义）。 + """ + scene_id = str(scene_id or "") + interval = float(interval if interval is not None else self.interval) + timeout = float(timeout if timeout is not None else self.timeout) + self.clear_notified(scene_id) + deadline = time.monotonic() + timeout + polls = 0 + recovered: List[Dict[str, Any]] = [] + stopped_by = "timeout" + found = False + while time.monotonic() < deadline: + sleep(interval) + polls += 1 + res = self.poll_once(scene_id, wait_for=wait_for) + for row in res["recovered"]: + recovered.append(row) + if on_recovered: + try: + on_recovered(row) + except Exception: + pass + if res["found"]: + found = True + stopped_by = "event_matched" + break + if res["notified"]: + stopped_by = "broadcast_received" + with self._lock: + self._counters["stops_on_notify"] += 1 + break + if stopped_by == "timeout": + with self._lock: + self._counters["timeouts"] += 1 + return {"scene_id": scene_id, "polls": polls, "recovered": recovered, + "recovered_count": len(recovered), "found": found, + "stopped_by": stopped_by, "interval": interval, "timeout": timeout} + + def wait_event(self, scene_id: str, event_uid: str, + timeout: Optional[float] = None, interval: Optional[float] = None, + sleep: Callable[[float], None] = time.sleep) -> Dict[str, Any]: + """等待特定事件被感知（广播或轮询兜底），返回 ``{found, waited_ms, polls}``。""" + start = time.monotonic() + res = self.run(scene_id, wait_for=str(event_uid), timeout=timeout, + interval=interval, sleep=sleep) + waited_ms = (time.monotonic() - start) * 1000.0 + return {"scene_id": str(scene_id or ""), "event_uid": str(event_uid), + "found": bool(res["found"]), "waited_ms": round(waited_ms, 3), + "polls": res["polls"], "stopped_by": res["stopped_by"], + "recovered_count": res["recovered_count"]} + + def stats(self) -> Dict[str, Any]: + with self._lock: + snap = dict(self._counters) + snap["scenes"] = sorted(self._scenes.keys()) + snap["interval"] = self.interval + snap["timeout"] = self.timeout + return snap + + +# ---------------------------------------------------------------- 计划判定 +def plan_uses_scene_index(plan_rows: Sequence[Any]) -> bool: + """判定 EXPLAIN 结果是否命中 ``ix_scene_created``（即不是全表扫描）。 + + MySQL 5.7 经典格式看 ``key`` / ``possible_keys`` + ``type``；``FORMAT=JSON`` 或 + mariadb 扩展格式退化为在序列化文本里找索引名。``type=ALL`` 且未用该索引 → False。 + """ + rows = list(plan_rows or []) + if not rows: + return False + for row in rows: + key = str(_field(row, "key", "") or "") + if key == "ix_scene_created": + return True + possible = str(_field(row, "possible_keys", "") or "") + rtype = str(_field(row, "type", "") or "") + if "ix_scene_created" in possible and rtype != "ALL": + return True + blob = repr(rows) + if "ix_scene_created" not in blob: + return False + return "'type': 'ALL'" not in blob and '"type": "ALL"' not in blob + + +# ---------------------------------------------------------------- 单例门面 +_DEFAULT = PollingFallback() +_LOCK = threading.RLock() + + +def get_fallback(conn_factory: Optional[Callable[[], Any]] = None) -> PollingFallback: + """取进程级兜底器单例（消费端默认用它；需要独立状态时自行 new 一个）。""" + global _DEFAULT + with _LOCK: + if conn_factory is not None and _DEFAULT._conn_factory is not conn_factory: + _DEFAULT._conn_factory = conn_factory + return _DEFAULT + + +def configure_fallback(conn_factory: Optional[Callable[[], Any]] = None, + interval: Optional[float] = None, + timeout: Optional[float] = None) -> PollingFallback: + """配置单例的连接工厂与节奏（interval/timeout），返回单例本身。""" + fb = get_fallback(conn_factory) + with _LOCK: + if interval is not None: + fb.interval = float(interval) + if timeout is not None: + fb.timeout = float(timeout) + return fb + + +def reset_fallback(conn_factory: Optional[Callable[[], Any]] = None, + interval: Optional[float] = None, + timeout: Optional[float] = None) -> PollingFallback: + """重建单例（清空水位线/去重/计数，测试隔离与宿主重配置用）。""" + global _DEFAULT + with _LOCK: + _DEFAULT = PollingFallback( + conn_factory=conn_factory if conn_factory is not None + else _DEFAULT._conn_factory, + interval=interval if interval is not None else _DEFAULT.interval, + timeout=timeout if timeout is not None else _DEFAULT.timeout, + ) + return _DEFAULT + + +def fallback_stats() -> Dict[str, Any]: + """单例兜底指标（补漏次数 / 漏网事件数 / 因广播到点停止次数 / 超时次数）。""" + return get_fallback().stats() + + +def poll_since(scene_id: str, watermark: str = "", conn_factory=None, + seen_uids: Optional[Sequence[Any]] = None) -> List[Dict[str, Any]]: + """门面：一次性查某场景水位线之后的事件（消费端手动补漏，不起定时器）。""" + return get_fallback(conn_factory).query_since(scene_id, watermark, seen_uids) + + +def run_fallback(scene_id: str, conn_factory=None, timeout: Optional[float] = None, + interval: Optional[float] = None, + on_recovered: Optional[Callable[[Dict[str, Any]], None]] = None, + sleep: Callable[[float], None] = time.sleep) -> Dict[str, Any]: + """门面：跑一轮 3s 兜底循环（消费端定时器线程体内调用）。""" + return get_fallback(conn_factory).run(scene_id, timeout=timeout, + interval=interval, + on_recovered=on_recovered, sleep=sleep) + + +def wait_event(scene_id: str, event_uid: str, conn_factory=None, + timeout: Optional[float] = None, interval: Optional[float] = None, + sleep: Callable[[float], None] = time.sleep) -> Dict[str, Any]: + """门面：用单例等待特定事件被感知（广播到点或轮询补漏，≤ timeout）。""" + return get_fallback(conn_factory).wait_event(scene_id, event_uid, timeout=timeout, + interval=interval, sleep=sleep) + + +def explain_poll_query(scene_id: str, watermark: str = "", conn_factory=None + ) -> List[Dict[str, Any]]: + """门面：对补漏查询跑 EXPLAIN（验证走 ``ix_scene_created``，非全表扫描）。""" + return get_fallback(conn_factory).explain_poll_query(scene_id, watermark)