---
name: world_sync
description: world_sync 模块技能文档。
---

# world_sync 模块

## 概述
世界数据同步模块，管理世界数据的导入/导出/合并同步记录与执行。依赖 world 模块（world 表）、appbase（appcodes 字典 sync_type/sync_status）。通过 load_world_sync() 挂载。

自 M11b-2/M11b-3 起，本模块同时承载 **scense_runtime 服务端权威写入薄扩展**：单事务事件+状态写入（`pbl_runtime_authority.write_event_with_state`）与**事务提交后广播**（`pbl_runtime_broadcast`）。

自 M11b-4 起，另承载**消费端轮询兜底**（`pbl_runtime_polling`）与**端到端时延基准**（`pbl_runtime_latency`）：广播为主、轮询为兜底，见「M11b-4 轮询兜底与时延基准」节。

## 数据模型
### 表 world_sync（models/world_sync.json）
| 字段 | 类型 | 说明 |
|------|------|------|
| id | str(32) | 主键 |
| world_id | str(32) | 所属世界（→world.id） |
| sync_type | str(16) | 同步类型（0导入/1导出/2合并） |
| source | str(255) | 数据来源 |
| status | str(16) | 状态（appcodes sync_status） |
| result_json | text | 同步结果 JSON |
| sync_date | date | 同步日期（业务日期） |
| created_at | timestamp | 创建时间 |

- 主键：["id"]；索引：idx_sync_world(world_id)
- 字典：world_id→world、sync_type→appcodes_kv(parentid='sync_type')、status→appcodes_kv(parentid='sync_status')

### 运行时权威表（M11b-2，models/）
`pbl_runtime_event`（append-only 事件）、`pbl_entity_state`（实体状态，state_version 服务端权威）、
`pbl_world_state_snapshot`（世界快照）。所有读写强制 `tenant_id` 打头，缺失 fail-closed。
`pbl_runtime_event` 上有 M11b-1 的 **`ix_scene_created (scene_id, created_at)`**，是 M11b-4 轮询补漏查询的唯一索引依赖。

## 关键接口
- list_world_syncs / get_world_sync / create_world_sync / update_world_sync / delete_world_sync / execute_world_sync
- wwwroot/api/world_sync.dspy（同步契约）、create/world_sync_update/world_sync_delete.dspy、get_search_*.dspy
- M11b-2：`write_event_with_state(event_data, state_update, ...)` —— 事件 + 状态在**同一事务**内提交
- M11b-3 广播 API（`world_sync/pbl_runtime_broadcast.py`，已在 `__init__.py` 导出）：
  - `install_broadcast_hook()` / `is_broadcast_hook_installed()`：把 `write_event_with_state` 包成 COMMIT 后广播（幂等，`load_world_sync()` 已自动调用）
  - `emit_post_commit_broadcast(write_result, context=...)`：手动发一条
  - `wrap_with_post_commit_broadcast(fn)`：包装任意返回事务结果的函数
  - `configure_broadcast(channels=..., webhook_urls=..., redis_url=..., queue_size=2000, retry_attempts=2, retry_backoff_s=0.02, enabled=True)`：宿主启动时配置通道
  - `subscribe_broadcast(handler)` / `unsubscribe_broadcast(handler)`：进程内订阅
  - `broadcast_stats()` / `broadcast_last_message()` / `broadcast_latency_percentile_ms(99)` / `drain_broadcast(timeout=2.0)` / `shutdown_broadcast()`：观测与优雅退出
  - 通道类：`InProcessChannel`、`WebhookChannel`、`RedisPubSubChannel`、`NoopChannel`（均继承 `BaseChannel`，`publish(message)`）
- M11b-4 轮询兜底 + 时延 API：见下节「门面清单」

## 事务提交后广播（M11b-3）要点
- **触发点**：仅事务 COMMIT 成功之后（`commit_ts` 即延迟起点）；事务回滚不广播。
- **异步**：单 dispatcher 线程 + 有界 `queue.Queue(2000)`，`submit` 用 `put_nowait` 微秒返回，不阻塞事务返回；队列满降级为当前线程直发一次。
- **SLO**：`BROADCAST_LATENCY_SLO_MS = 300.0`，口径 P99；实测 n=200 p50=2.4ms / p99=4.1ms / max=4.2ms（`scripts/m11b3_selftest_output.txt`）。
- **失败语义**：fire-and-forget + 有限重试；通道/订阅者异常一律吞掉并 `ERROR` 日志 + `stats["failed"]` 计数，**绝不回滚已提交事务**、绝不外抛给业务调用方。
- **广播为主、轮询为兜底**：广播**投递路径自身**不含 `time.sleep` 轮询、不读 DB，等待靠 `Queue.get(block=True)` 条件变量；广播不可靠（丢消息/Redis 断连/订阅者异常）时，由消费端侧的 `pbl_runtime_polling` 做 3s 只读补漏，见「M11b-4 轮询兜底与时延基准」节。**本模块并非「无轮询」**。
- **环境变量**：`PBL_BROADCAST_DISABLED`（全局关闭）、`PBL_BROADCAST_WEBHOOK_URLS`（逗号分隔 webhook）、`PBL_BROADCAST_REDIS_ENABLED` + `PBL_BROADCAST_REDIS_URL`（Redis pub/sub）。

## M11b-4 轮询兜底与时延基准

### 定位与边界
广播是**尽力而为**的加速通道，会丢。M11b-4 在**消费端侧**提供只读补漏能力 + 全链路时延度量，
**不修改 M11b-1/2/3 任何核心逻辑**（selftest AC4.1/4.2/4.3 断言：写入端源码无 polling/fallback 字样、
polling 与写入端事务模块 import 零交集、补漏 SQL 只读 `pbl_runtime_event`）。

### 门面清单（16 个包级导出名，`from world_sync import X` 可用；`init.py` 已 `env.X = X`）
轮询兜底 `world_sync/pbl_runtime_polling.py`（12 个）：
- `poll_since(scene_id, watermark="", conn_factory=None, ...)` —— 单轮补漏查询（不起定时器），返回 `{scene_id, watermark, rows, count}`
- `run_fallback(scene_id, ...)` —— 起一轮兜底循环（同步阻塞，宿主放线程里跑），返回 `{polls, recovered_count, found, stopped_by, ...}`
- `wait_event(scene_id, event_uid, ...)` —— 等指定事件出现（`stopped_by=event_matched`）
- `explain_poll_query(scene_id, watermark, conn_factory=None)` —— 对补漏 SQL 跑 EXPLAIN，返回计划行
- `build_poll_sql(dialect, watermark, scene_id=..., ...)` —— 按方言生成补漏 SQL（谓词 `scene_id` 等值打头 + `created_at >= ?` 范围 + `ORDER BY created_at`）
- `normalize_watermark(value)` —— 水位线格式守卫，返回归一后的字符串；**不合规（含 ISO）返回 `''`**。datetime 对象按 UTC 格式化
- `plan_uses_scene_index(plan_rows)` —— 程序化判定是否命中 `ix_scene_created`（`type=ALL` → False）
- `configure_fallback(...)` / `get_fallback()` / `reset_fallback(...)` —— 全局兜底实例配置/取用/重置
- `fallback_stats()` —— 计数快照 `{polls, queries, recovered, duplicated, missed_by_broadcast, timeouts, stops_on_notify, scenes, interval, timeout}`
- 类 `PollingFallback`：`run(scene_id, wait_for=None, timeout=None, interval=None, on_recovered=None, sleep=time.sleep)` / `notify_broadcast(message, created_at=None)`（收 M11b-3 广播消息 dict）/ `set_watermark(scene_id, created_at)`（bool 返回=是否推进）/ `watermark(scene_id)` / `note_event_seen` / `explain_poll_query` / `stats()`

时延度量 `world_sync/pbl_runtime_latency.py`（4 个）：
- `measure_once(...)` —— 跑一次真实链路并分段计时
- `run_chain(n, ...)` —— 批量采样（n=200 为基准口径）
- `summarize_latency(samples)` —— P50/P90/P99/max + 分段 P99 + 瓶颈段 + 感知路径分桶
- `latency_report(...)` —— 汇总报告（含 `verdict`、`run_at` 时间戳），落盘 `scripts/m11b4_latency_report.json`

### 常量语义（`pbl_runtime_polling.py` / `pbl_runtime_latency.py`）
| 常量 | 值 | 语义 |
|------|----|------|
| `POLL_INTERVAL_S` | `3.0` | 轮询周期（验收口径「广播丢失 3s 内补漏」= 1 个周期） |
| `POLL_TIMEOUT_S` | `30.0` | 兜底循环总超时默认值（超时即停并计 `timeouts`，不造僵尸定时器）；selftest 显式传 `10.0` |
| `POLL_LIMIT` | `500` | 单轮补漏 LIMIT，防一次拉爆；剩余下一轮继续 |
| `FALLBACK_SLO_MS` | `3000.0` | 补漏 SLO = 1 个轮询周期 |
| `SEEN_IDS_IN_SQL` | `200` | 已投递 `event_uid` 参与 SQL `NOT IN` 的最大长度（防参数爆炸） |
| `E2E_LATENCY_SLO_MS` | `1000.0` | 端到端感知时延 SLO（P99 口径，仅统计广播路径样本） |

### 水位线陷阱（务必读）
- 水位线**只接受库内空格格式** `'YYYY-MM-DD HH:MM:SS'`（可带小数秒）。广播消息里的 **ISO 串（带 `T`/`Z`/时区）必须被拒**（`normalize_watermark` 返回 `''`、`set_watermark` 返回 False）—— 否则与 `created_at` 不可比，会造成「永久漏」或「永久重」，且是静默的。
- 水位线**单调不减**：新值小于旧值直接忽略（并发 notify 与轮询推进不会把水位线拉回）。
- 查询用 `created_at >= watermark`（**闭区间**）而非 `>`：`DATETIME` 秒级精度下**同秒事件会被 `>` 永久跳过**。同秒重复投递由「已投递 `event_uid` 的 `NOT IN` 集合 + 内存 seen 集合」双保险消除。
- 每场景独立水位线（`fallback_stats()["scenes"]` 可见），跨场景不串。

### 索引依赖与 EXPLAIN 判定方法
补漏 SQL 谓词顺序与 `ix_scene_created (scene_id, created_at)` 完全对齐（等值列打头、范围列在后，
既走索引过滤又免排序）。复核方式：
- SQLite：`explain_poll_query()` 输出应含 `SEARCH pbl_runtime_event USING INDEX ix_scene_created (scene_id=? AND created_at>?)`，出现 `SCAN` 即全表扫描 = 不合规。
- MySQL：`plan_uses_scene_index(explain_poll_query(...))` → True 才算合规；`type=ALL` 判 False。测试库凭据不可用时该项记 skipped，须在交付件标注环境受限（勿写 PASS）。

### 消费端接线方式
- 消费端订阅广播：`subscribe_broadcast(handler)`；handler 里调 `get_fallback().notify_broadcast(msg, created_at=write_result["created_at"])`（`msg` = M11b-3 广播消息 dict）—— 置位 `notified` 短路标记并推进水位线，**当轮立即停止轮询**（实测墙钟 200.8ms，未走满 3s 周期）。广播消息只有 ISO `commit_ts_text`，**不能**直接当水位线用，须把写入端返回的库内格式 `created_at` 传进来，否则水位线不推进。
- 兜底循环的启停由**宿主/消费端自己掌握**：把 `run_fallback()` / `PollingFallback.run()` 放进宿主自己的定时器或线程。**模块不自建后台线程**，避免污染写入端生命周期。
- 广播丢失（`missed_by_broadcast` 计数 +1）时靠轮询补漏，该路径样本时延 ≥ 1 个周期，由 `via_polling_fallback` 单独统计，**不混入 P99 ≤ 1s 判定**（否则会把「兜底代价」误报成「链路超标」）。

## 陷阱
- 库名不硬编码：.py 用 ServerEnv().get_module_dbname('world_sync')，.dspy 用 get_sor_context(request._run_ns,'world_sync')
- dspy 无 import
- create/execute 必须设置 created_at、sync_date，否则 sor.C 静默丢记录
- update 剔除 _text 后缀
- codes.table 引用 world 不用 module.table 点号
- **广播 hook 幂等**：`install_broadcast_hook()` 重复调用不得重复包装（已用 `__wrapped_post_commit__` 标记），否则一条事件广播多次
- **广播内禁止再写事务/再发广播**：dispatcher 线程只做 publish，回写 DB 会造成递归与延迟失真
- **广播链路测试断言前用 `drain_broadcast()` 等待，不要在广播测试里 sleep 轮询**（QC 会按「引入轮询」退回）。注意：这与 M11b-4 的**消费端补漏轮询**是两件事——后者是产品能力，位于 `pbl_runtime_polling.py`，不受本条限制。
- **M11b-4 三处同步注册**：新增/删除 polling、latency 门面函数必须同步改 ①实现 ②`__init__.py` 导入 + `__all__` ③`init.py` `env.xxx`。漏 ② → 宿主 `from world_sync import explain_poll_query` AttributeError（本任务 QC #2 实际踩过）。selftest AC5.1 逐名核验 16 个导出名。
- **文档数值必须与 `scripts/m11b4_latency_report.json` 一致**：该 JSON 是唯一数值源且自带 `run_at` 时间戳；引用分段 P99/广播分位数/墙钟数字时逐字段照抄并标注时间戳，禁止凭记忆或引用另一次未落盘的运行。
- **AC1.4 判定口径含容差**：`wall_ms <= FALLBACK_SLO_MS + SLO_TOLERANCE_MS`（3000+500ms，来源为真实 3s sleep 的唤醒抖动与周期边界），文档不得写成「3000.5ms ≤ 3000ms」。
- **`SLO_TOLERANCE_MS` 是模块源码常量**（QC #17）：定义在 `pbl_runtime_polling.py`（与 `FALLBACK_SLO_MS` 同段），并已按三处同步导出（`__init__.py` import + `__all__`）与登记（`init.py` `env.SLO_TOLERANCE_MS`）。selftest 只 `from world_sync.pbl_runtime_polling import SLO_TOLERANCE_MS` 引用，**禁止在测试脚本里重复本地定义**——否则 dev-notes 常量表「与源码逐一对齐」的声称不成立。同理 `env.SEEN_IDS_IN_SQL` 也已登记。
- **`poll_since` 门面返回键是 `count`**（不是 `recovered_count`）：`{scene_id, watermark, rows, count}`。`recovered_count` 只存在于 `PollingFallback.poll_once()` / `run()` 的返回值。写文档/断言前先看源码返回字典，别把两套口径混用（QC #15）。
- `RedisPubSubChannel` 惰性 import `redis`：环境缺包只 `ChannelUnavailable` 跳过，不影响 InProcess/webhook 通道，不要让 Redis 成为硬依赖
- `scripts/sync_code_to_app.py` 的 `collect()` 必须以 `WS_ROOT` 拼绝对路径 —— 用相对路径 `isdir` 会恒返回空清单，导致 `--check` 打印假 PASS（M11b-3 已修，改脚本时勿回退）

## 依赖
- world、appbase、rbac
- M11b-2/M11b-3 广播零第三方硬依赖（stdlib：threading/queue/urllib/hashlib；redis 可选）
- M11b-4 轮询兜底/时延基准同样零第三方依赖（stdlib：re/threading/time/json），只依赖 `pbl_runtime_sql` / `pbl_runtime_errors` / `pbl_runtime_tx_env`