deliver: 交付收口（引擎代为提交）

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",

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)

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

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)