545 lines
24 KiB
Python
545 lines
24 KiB
Python
# -*- 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)
|