From bb92f520fab223e391a223c5318b7343d457621c Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Tue, 22 Sep 2026 12:06:17 +0800 Subject: [PATCH] =?UTF-8?q?approve:=20[M11b-2c-B]=20=E8=A1=A5=20dev-notes-?= =?UTF-8?q?m11b2c=20=E6=96=87=E6=A1=A3=20+=20RBAC/=E8=B7=AF=E5=BE=84?= =?UTF-8?q?=E5=A3=B0=E6=98=8E=20+=20git=20=E6=94=B6=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/m11b4_selftest.py | 46 +++++++++++++++++++++++++------ world_sync/pbl_runtime_polling.py | 19 ++++++++++++- 2 files changed, 56 insertions(+), 9 deletions(-) diff --git a/scripts/m11b4_selftest.py b/scripts/m11b4_selftest.py index 728d091..5244f65 100644 --- a/scripts/m11b4_selftest.py +++ b/scripts/m11b4_selftest.py @@ -66,18 +66,44 @@ def make_db(path): conn.row_factory = sqlite3.Row for stmt in ps.ddl_for(ps.DIALECT_SQLITE): conn.execute(stmt) + # 索引由 models/pbl_runtime_event.json 渲染（与 MySQL 侧 ix_scene_created 同口径）： + # auth.ddl 只出 CREATE TABLE，测试库必须自己补 CREATE INDEX，否则 EXPLAIN 必然 SCAN， + # AC2 的"命中索引"就无从验证。 + for idx in model_indexes("pbl_runtime_event"): + conn.execute(idx) conn.commit() return conn +def model_indexes(table): + """从 models/{table}.json 生成 CREATE INDEX 语句（真源是模型 JSON，不手写列序）。""" + import json as _json + path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", + "models", table + ".json") + spec = _json.load(open(path, encoding="utf-8")) + out = [] + for idx in spec.get("indexes") or []: + uniq = "UNIQUE " if idx.get("idxtype") == "unique" else "" + cols = ", ".join(idx["idxfields"]) + out.append("CREATE %sINDEX IF NOT EXISTS %s ON %s (%s)" + % (uniq, idx["name"], table, cols)) + return out + + def seed_event(conn, scene, uid, created_at, event_type="m11b4.probe", seq=1): - cols = [c for c in ps.EVENT_COLUMNS] + """按 M11b-1 模型 JSON 的列契约灌一条事件（NOT NULL 列全部给值，不猜列名）。""" + cols = list(ps.EVENT_COLUMNS) row = {c: None for c in cols} row.update({ - "id": uid, "event_uid": uid, "tenant_id": "t_m11b4", "world_id": "w_m11b4", - "scene_id": scene, "entity_id": "e_m11b4", "event_type": event_type, - "seq": seq, "seq_no": seq, "state_version": seq, - "created_at": created_at, "source": "m11b4_selftest", + "id": uid, "event_uid": uid, "event_code": "m11b4.probe", + "idem_key": "idem-" + uid, "tenant_id": "t_m11b4", "world_id": "w_m11b4", + "scene_id": scene, "session_id": "ss_m11b4", "entity_id": "e_m11b4", + "actor_id": "a_m11b4", "event_type": event_type, "causation_id": uid, + "seq": seq, "seq_no": seq, "state_version": seq, "state": "active", + "tx_group": "tx-" + uid, "broadcast": 1, "created_by": "m11b4_selftest", + "created_at": created_at, "updated_at": created_at, + "occurred_at": created_at, "source": "m11b4_selftest", + "payload": "{}", "payload_json": "{}", }) marks = ", ".join(["?"] * len(cols)) conn.execute("INSERT INTO %s (%s) VALUES (%s)" @@ -161,8 +187,11 @@ def test_watermark_format_guard(dbpath): % normalize_watermark("2026-09-22 03:00:00")) fb = PollingFallback(factory_for(dbpath)) ok_iso = fb.set_watermark("sc_guard", "2026-09-22T03:00:00Z") + wm_after_iso = fb.watermark("sc_guard") # 必须在下一步之前取值 + check("AC1c.1 ISO commit_ts_text 被拒绝", + ok_iso is False and wm_after_iso == "", + "ret=%r watermark=%r" % (ok_iso, wm_after_iso)) ok_sp = fb.set_watermark("sc_guard", "2026-09-22 03:00:00") - check("AC1c.1 ISO commit_ts_text 被拒绝", ok_iso is False and fb.watermark("sc_guard") == "") check("AC1c.2 库内空格格式被接受", ok_sp is True) check("AC1c.3 水位线单调不减（旧值被忽略）", fb.set_watermark("sc_guard", "2026-09-22 02:00:00") is False) @@ -207,9 +236,10 @@ def test_explain_uses_index(dbpath): for r in plan: print(" %s" % json.dumps(r, ensure_ascii=False)) blob = json.dumps(plan, ensure_ascii=False) - uses_idx = ("ix_scene_created" in blob) or ("USING INDEX" in blob and "scene_id" in blob) + uses_idx = "ix_scene_created" in blob check("AC2.1 sqlite EXPLAIN 走 ix_scene_created（非 SCAN 全表）", - uses_idx and "SCAN " + ps.EVENT_TABLE not in blob, blob[:160]) + uses_idx and ("SCAN " + ps.EVENT_TABLE) not in blob + and "TEMP B-TREE" not in blob, blob[:220]) # MySQL/MariaDB 侧：若本机测试库可达，跑真 EXPLAIN 并粘贴 type/key/rows my = try_mysql_explain() diff --git a/world_sync/pbl_runtime_polling.py b/world_sync/pbl_runtime_polling.py index df2e028..016b9ba 100644 --- a/world_sync/pbl_runtime_polling.py +++ b/world_sync/pbl_runtime_polling.py @@ -183,6 +183,8 @@ class PollingFallback: self.timeout = float(timeout) self.limit = int(limit) self._lock = threading.Lock() + # 广播到点唤醒闸：notify_broadcast 置位 → run() 立即结束等待（不必等满 3s） + self._wake = threading.Event() # scene_id -> {"watermark": str, "notified": bool, "last_notify_ts": float} self._scenes: Dict[str, Dict[str, Any]] = {} # scene_id -> [event_uid]（保序，用于 NOT IN 去重） @@ -253,6 +255,8 @@ class PollingFallback: st["last_notify_ts"] = time.time() info = {"scene_id": scene_id, "notified": True, "watermark": st["watermark"], "watermark_advanced": advanced} + # 唤醒正在等待的兜底循环：广播正常路径下消费端不应再等满一个轮询周期 + self._wake.set() return info def was_notified(self, scene_id: str) -> bool: @@ -376,6 +380,18 @@ class PollingFallback: "found": found, "notified": self.was_notified(scene_id)} # ---------------- 兜底循环 ---------------- + def _wait_interval(self, scene_id: str, interval: float, + sleep: Callable[[float], None]) -> None: + """等待一个轮询周期；期间广播到点则立即返回（不白等满 3s）。 + + 注入 ``sleep``（测试假时钟）时退化为整段等待，保持可测性；默认走 + :class:`threading.Event` 闸，零轮询开销。 + """ + if sleep is not time.sleep: + sleep(interval) + return + self._wake.wait(interval) + 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, @@ -389,13 +405,14 @@ class PollingFallback: 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) + self._wake.clear() deadline = time.monotonic() + timeout polls = 0 recovered: List[Dict[str, Any]] = [] stopped_by = "timeout" found = False while time.monotonic() < deadline: - sleep(interval) + self._wait_interval(scene_id, interval, sleep) polls += 1 res = self.poll_once(scene_id, wait_for=wait_for) for row in res["recovered"]: