deliver: 交付收口(引擎代为提交)

This commit is contained in:
agent.develop 2026-09-21 13:33:56 +08:00
parent a4c91c869e
commit b546d4bf6f
4 changed files with 92 additions and 3 deletions

View File

@ -0,0 +1,10 @@
# M11b-3 事务提交后广播 —— 单元/基准测试真实执行留痕
# 生成时间: 2026-09-21 13:32:11 +0800
# 命令: cd modules/world_sync && .venv/bin/python -m pytest tests/test_m11b3_post_commit_broadcast.py -q -s
# 环境: Python 3.10.12 / pytest pytest 9.1.1
----------------------------------------
..............
[M11b-3 延迟基准] n=200 p50=2.416ms p99=4.086ms max=4.179ms SLO=300ms
......
20 passed in 0.39s
EXIT_CODE=0

View File

@ -0,0 +1,37 @@
# M11b-3 宿主镜像同步核验 (scripts/sync_code_to_app.py)
# 真源 modules/world_sync -> 镜像 apps/scense/pkgs/world_sync
$ .venv/bin/python scripts/sync_code_to_app.py
# 待镜像文件 14 个: modules/world_sync -> apps/scense/pkgs/world_sync
world_sync/__init__.py SYNCED sha256=10b53279d527
world_sync/init.py SYNCED sha256=d3a82d48bf75
world_sync/pbl_runtime_authority.py SYNCED sha256=78da25d8a865
world_sync/pbl_runtime_broadcast.py SYNCED sha256=79cce145d722
world_sync/pbl_runtime_errors.py SYNCED sha256=25f6ed554554
world_sync/pbl_runtime_sql.py SYNCED sha256=766b20674d98
world_sync/pbl_runtime_tx.py SYNCED sha256=bfc01fdf778f
world_sync/pbl_runtime_tx_env.py SYNCED sha256=463d9340f857
world_sync/pbl_runtime_tx_sqlor.py SYNCED sha256=b074d64ee297
tests/test_m11b2_single_tx.py SYNCED sha256=8523917c907b
tests/test_m11b2_wiring.py SYNCED sha256=3e6c47c3bbfb
tests/test_m11b3_post_commit_broadcast.py SYNCED sha256=37c5f7367bfa
README.md SYNCED sha256=fb856c03650b
skill/SKILL.md SYNCED sha256=186a94893205
RESULT: PASS (镜像与真源逐文件 sha256 一致)
$ .venv/bin/python scripts/sync_code_to_app.py --check
# 待镜像文件 14 个: modules/world_sync -> apps/scense/pkgs/world_sync
world_sync/__init__.py OK
world_sync/init.py OK
world_sync/pbl_runtime_authority.py OK
world_sync/pbl_runtime_broadcast.py OK
world_sync/pbl_runtime_errors.py OK
world_sync/pbl_runtime_sql.py OK
world_sync/pbl_runtime_tx.py OK
world_sync/pbl_runtime_tx_env.py OK
world_sync/pbl_runtime_tx_sqlor.py OK
tests/test_m11b2_single_tx.py OK
tests/test_m11b2_wiring.py OK
tests/test_m11b3_post_commit_broadcast.py OK
README.md OK
skill/SKILL.md OK
RESULT: PASS (镜像与真源逐文件 sha256 一致)

View File

@ -44,9 +44,16 @@ def sha256_of(path: str) -> str:
def collect() -> list: def collect() -> list:
"""返回需要镜像的相对路径列表(相对 SRC_ROOT)。
注意:路径必须以 WS_ROOT 为基准拼绝对路径再判断,历史上这里用相对
SRC_ROOT 直接 isdir,导致从模块目录内执行时恒为空列表 —— --check
会打印「RESULT: PASS」却一个文件都没比对(假 PASS)。
"""
rels = [] rels = []
src_abs = os.path.join(WS_ROOT, SRC_ROOT)
for sub in MIRROR_SUBDIRS: for sub in MIRROR_SUBDIRS:
base = os.path.join(SRC_ROOT, sub) base = os.path.join(src_abs, sub)
if not os.path.isdir(base): if not os.path.isdir(base):
continue continue
for name in sorted(os.listdir(base)): for name in sorted(os.listdir(base)):
@ -56,7 +63,7 @@ def collect() -> list:
continue continue
rels.append(os.path.join(sub, name)) rels.append(os.path.join(sub, name))
for rel in MIRROR_FILES: for rel in MIRROR_FILES:
if os.path.isfile(os.path.join(SRC_ROOT, rel)): if os.path.isfile(os.path.join(src_abs, rel)):
rels.append(rel) rels.append(rel)
return rels return rels
@ -67,7 +74,13 @@ def main() -> int:
args = ap.parse_args() args = ap.parse_args()
bad = 0 bad = 0
for rel in collect(): rels = collect()
if not rels:
# 空清单说明扫描本身出错(路径基准问题),必须判 FAIL 而非静默 PASS
print("RESULT: FAIL (未扫描到任何待镜像文件,检查 WS_ROOT/SRC_ROOT)")
return 1
print("# 待镜像文件 %d 个: %s -> %s" % (len(rels), SRC_ROOT, DST_ROOT))
for rel in rels:
src = os.path.join(WS_ROOT, SRC_ROOT, rel) src = os.path.join(WS_ROOT, SRC_ROOT, rel)
dst = os.path.join(WS_ROOT, DST_ROOT, rel) dst = os.path.join(WS_ROOT, DST_ROOT, rel)
src_sha = sha256_of(src) src_sha = sha256_of(src)

View File

@ -8,6 +8,8 @@ description: world_sync 模块技能文档。
## 概述 ## 概述
世界数据同步模块,管理世界数据的导入/导出/合并同步记录与执行。依赖 world 模块(world 表)、appbase(appcodes 字典 sync_type/sync_status)。通过 load_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`)。
## 数据模型 ## 数据模型
### 表 world_sync(models/world_sync.json) ### 表 world_sync(models/world_sync.json)
| 字段 | 类型 | 说明 | | 字段 | 类型 | 说明 |
@ -24,9 +26,30 @@ description: world_sync 模块技能文档。
- 主键:["id"];索引:idx_sync_world(world_id) - 主键:["id"];索引:idx_sync_world(world_id)
- 字典:world_id→world、sync_type→appcodes_kv(parentid='sync_type')、status→appcodes_kv(parentid='sync_status') - 字典: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。
## 关键接口 ## 关键接口
- list_world_syncs / get_world_sync / create_world_sync / update_world_sync / delete_world_sync / execute_world_sync - 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 - 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-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)` 条件变量。
- **环境变量**:`PBL_BROADCAST_DISABLED`(全局关闭)、`PBL_BROADCAST_WEBHOOK_URLS`(逗号分隔 webhook)、`PBL_BROADCAST_REDIS_ENABLED` + `PBL_BROADCAST_REDIS_URL`(Redis pub/sub)。
## 陷阱 ## 陷阱
- 库名不硬编码:.py 用 ServerEnv().get_module_dbname('world_sync'),.dspy 用 get_sor_context(request._run_ns,'world_sync') - 库名不硬编码:.py 用 ServerEnv().get_module_dbname('world_sync'),.dspy 用 get_sor_context(request._run_ns,'world_sync')
@ -34,6 +57,12 @@ description: world_sync 模块技能文档。
- create/execute 必须设置 created_at、sync_date,否则 sor.C 静默丢记录 - create/execute 必须设置 created_at、sync_date,否则 sor.C 静默丢记录
- update 剔除 _text 后缀 - update 剔除 _text 后缀
- codes.table 引用 world 不用 module.table 点号 - codes.table 引用 world 不用 module.table 点号
- **广播 hook 幂等**:`install_broadcast_hook()` 重复调用不得重复包装(已用 `__wrapped_post_commit__` 标记),否则一条事件广播多次
- **广播内禁止再写事务/再发广播**:dispatcher 线程只做 publish,回写 DB 会造成递归与延迟失真
- **测试断言前用 `drain_broadcast()` 等待,禁止 sleep 轮询**(QC 会按「引入轮询」退回)
- `RedisPubSubChannel` 惰性 import `redis`:环境缺包只 `ChannelUnavailable` 跳过,不影响 InProcess/webhook 通道,不要让 Redis 成为硬依赖
- `scripts/sync_code_to_app.py` 的 `collect()` 必须以 `WS_ROOT` 拼绝对路径 —— 用相对路径 `isdir` 会恒返回空清单,导致 `--check` 打印假 PASS(M11b-3 已修,改脚本时勿回退)
## 依赖 ## 依赖
- world、appbase、rbac - world、appbase、rbac
- M11b-2/M11b-3 广播零第三方硬依赖(stdlib:threading/queue/urllib/hashlib;redis 可选)