From 5cde2cf569d829a360ca1f993107af9fb0840ce3 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Mon, 21 Sep 2026 12:34:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(M11b-2):=20=E5=8D=95=E4=BA=8B=E5=8A=A1?= =?UTF-8?q?=E5=86=99=E5=85=A5=E4=BA=8B=E4=BB=B6+=E5=AE=9E=E4=BD=93?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E5=8E=9F=E5=AD=90=E8=90=BD=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pbl_runtime_ext/tx_write.py | 31 +++++++++--- scripts/m11b2c_p99_probe.py | 98 +++++++++++++++++++++++++++---------- scripts/m11b_selftest.py | 10 +++- 3 files changed, 105 insertions(+), 34 deletions(-) diff --git a/pbl_runtime_ext/tx_write.py b/pbl_runtime_ext/tx_write.py index 9b6ae39..ed881ea 100644 --- a/pbl_runtime_ext/tx_write.py +++ b/pbl_runtime_ext/tx_write.py @@ -109,8 +109,16 @@ def required_columns(table): return out -def build_row(table, wanted, strict=True): - """按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。""" +def build_row(table, wanted, strict=True, full_row=True): + """按表真实列过滤字段。strict=True 时未知列直接抛错（防静默丢列写空行）。 + + full_row=True（默认，INSERT 语义）：额外校验 NOT NULL 列齐备，缺列抛 PBL-RTX-0004。 + full_row=False（UPDATE SET 子句语义）：被改动的行在库中已存在，未出现在 SET 列表里的 + NOT NULL 列保持原值，**不存在"缺列"**，因此不得套用 INSERT 的完整性校验。 + M11b-2c 修复：此前 UPDATE 分支沿用 full_row 校验，导致对已存在状态行的正常更新 + 被 PBL-RTX-0004 误拒（tenant_id/session_id/entity_id 是 WHERE 定位维度、 + created_at 是不可覆写的建档时间，三者都不应出现在 SET 列表中）。 + """ cols = table_columns(table) if not cols: if strict: @@ -123,10 +131,17 @@ def build_row(table, wanted, strict=True): "PBL-RTX-0003", "%s 未知列 %s（权威列=%s）" % (table, unknown, cols)) row = dict((k, v) for k, v in wanted.items() if k in cols) - missing = [c for c in required_columns(table) if c not in row] - if missing: - raise rtx_db.RtxError("PBL-RTX-0004", - "%s 缺少 NOT NULL 列 %s" % (table, missing)) + if full_row: + missing = [c for c in required_columns(table) if c not in row] + if missing: + raise rtx_db.RtxError("PBL-RTX-0004", + "%s 缺少 NOT NULL 列 %s" % (table, missing)) + else: + # UPDATE SET 语义：定位维度与建档时间不得被 SET 覆写（防误改主键/租户列） + immutable = [c for c in ("tenant_id", "id", "created_at") if c in row] + if immutable: + raise rtx_db.RtxError("PBL-RTX-0005", + "%s UPDATE 禁止 SET 不可变列 %s" % (table, immutable)) return row @@ -250,10 +265,12 @@ async def _apply_state(tx, tenant_id, session_id, upd, seq_no, base_version, "entity_id=%s 乐观锁冲突：base_version=%s 当前 state_version=%s（整体回滚）" % (entity_id, base_version, cur_ver)) new_ver = cur_ver + 1 # 服务端单调递增 + # UPDATE SET 子句：只含被变更列；tenant_id/session_id/entity_id 走 WHERE 定位， + # created_at 保持建档原值 —— 故按 full_row=False 构造（见 build_row 文档） vals = build_row(STATE_TABLE, { STATE_COL_VALUE: state_version, EVENT_COL_STATE_VERSION: new_ver, "checksum": checksum, "updated_by": actor_id, "updated_at": now, - }, strict=False) + }, strict=True, full_row=False) await tx.execute( "UPDATE %s SET %s WHERE tenant_id=%%s AND session_id=%%s AND %s=%%s" % (STATE_TABLE, ", ".join("%s=%%s" % c for c in vals.keys()), STATE_COL_KEY), diff --git a/scripts/m11b2c_p99_probe.py b/scripts/m11b2c_p99_probe.py index 5786494..bcbbbac 100644 --- a/scripts/m11b2c_p99_probe.py +++ b/scripts/m11b2c_p99_probe.py @@ -12,11 +12,19 @@ install_fake()，驱动 **同一份 pbl_runtime_ext/tx_write.py 主链路代码* 因此计时覆盖真实的「幂等键计算 → 事件 INSERT → 状态 UPSERT(乐观锁) → 快照 INSERT → commit → 提交后广播」完整调用路径。 -⚠ 采样走 **状态行首建（INSERT）分支**（每次采样用唯一 entity_id）： - 本工作空间实测发现 `_apply_state` 的 UPDATE 分支存在业务缺陷（PBL-RTX-0004， - 见 probe 末尾 D1 段与 m11b2c-selftest-run.log），该缺陷属业务实现范畴， - M11b-2c-A 任务边界禁止修改 pbl_runtime_ext/*.py，故本探针不覆盖 UPDATE 分支， - 并在输出中如实标注，交由后续任务修复后复测。 +采样覆盖两条分支（M11b-2c 修正）： + · S 段：状态行 **首建（INSERT）分支** —— 每次采样用唯一 entity_id； + · U 段：状态行 **更新（UPDATE）分支** —— 对固定 entity_id 反复覆写。 + 历史轮次本探针在 U 段实测到 PBL-RTX-0004（_apply_state UPDATE 分支误用 INSERT 的 + NOT NULL 全列校验），现已在 tx_write.build_row(full_row=False) 修复，U 段转为 + **正常计时样本**（不再只是缺陷取证），断言 state_version 单调递增。 + +预热隔离（M11b-2c 修正 QC #3）：预热写入改用 **独立 session_id**，避免预热的 +state_key 与采样写入落在同一 (tenant, session) 维度下 —— read_states 按 +(tenant_id, session_id) 过滤，历史版本 entity_ids 入参在内存替身上未生效， +导致采样断言 len(rows)==1 恒不成立、T4/T4b 直接崩溃、P99 数值从未产出。 +现 additionally 在采样侧按 entity_id 二次过滤并断言返回行的 entity_id 与请求一致， +双重防"静默放宽"。 诚实声明：本工作空间无 ahserver/apppublic/sqlor 运行时与本地 MariaDB，故 P99 为 **库函数主链路（内存事务引擎）路径**的实测值，不含网络 RTT 与真实 InnoDB 提交开销； @@ -103,15 +111,18 @@ def run(samples): write_ms = [] read_ms = [] + upd_ms = [] seq_rows = [] defect = {"seen": False, "code": "", "msg": ""} + bad_rows = [] # 返回行 entity_id 与请求不一致的取证计数 async def main(): tenant, world, sess = 7, 9001, 90011 + warm_sess = 90911 # 预热用独立 session，不污染采样维度 # 预热 3 次（不计入样本）：排除 import/属性解析冷启动对 P99 的污染 for i in range(3): await wtx.apply_runtime_event( - tenant_id=tenant, world_id=world, session_id=sess, + tenant_id=tenant, world_id=world, session_id=warm_sess, event_type="move", payload={"x": i}, state_updates=[{"state_key": "warm%d" % i, "state_value": {"x": i}}], @@ -134,20 +145,47 @@ def run(samples): rows = await wtx.read_states(tenant_id=tenant, session_id=sess, entity_ids=[eid]) t3 = time.perf_counter() - assert len(rows) == 1, "read 主链路返回行数异常: %r" % (rows,) + # 按 entity_id 二次过滤（内存替身/驱动未必下推 IN 条件），再断言行数； + # 同时校验返回行的 entity_id 与请求一致，防止过滤失效被静默放宽。 + rows = [x for x in rows if str(x.get("entity_id")) == eid] + if len(rows) != 1 or str(rows[0].get("entity_id")) != eid: + bad_rows.append((eid, len(rows))) + assert len(rows) == 1, "read 主链路返回行数异常 eid=%s rows=%r" % (eid, rows) + assert str(rows[0]["entity_id"]) == eid, ( + "read 返回行 entity_id 与请求不一致: %s != %s" + % (rows[0]["entity_id"], eid)) read_ms.append((t3 - t2) * 1000.0) - # D1：状态行 **更新分支**（同 entity_id 二次写入）—— 本工作空间实测缺陷取证 - try: - await wtx.apply_runtime_event( - tenant_id=tenant, world_id=world, session_id=sess, - event_type="move", payload={"x": 777}, - state_updates=[{"state_key": "p99e0", "state_value": {"x": 777}}], - client_event_id="p99-update-probe") - print("D1[更新分支] 未抛异常（缺陷已修复或路径变化）") - except rtx_db.RtxError as exc: - defect = {"seen": True, "code": exc.code, "msg": str(exc.msg)[:160]} - print("D1[更新分支] 抛 %s → %s" % (exc.code, defect["msg"])) + # U 段：状态行 **更新分支**（同 entity_id 反复覆写）—— 缺陷修复后转为正常计时 + upd_eid = "p99u" + await wtx.apply_runtime_event( + tenant_id=tenant, world_id=world, session_id=sess, + event_type="move", payload={"x": 1}, + state_updates=[{"state_key": upd_eid, "state_value": {"x": 1}}], + client_event_id="p99u-init") + for i in range(max(samples, 30)): + try: + t4 = time.perf_counter() + ru = await wtx.apply_runtime_event( + tenant_id=tenant, world_id=world, session_id=sess, + event_type="move", payload={"x": 2000 + i}, + state_updates=[{"state_key": upd_eid, + "state_value": {"x": 2000 + i}}], + client_event_id="p99u-%d" % i) + t5 = time.perf_counter() + assert ru.get("ok"), "update 分支返回 not ok: %r" % (ru,) + assert int(ru["states"][0]["state_version"]) == i + 2, ( + "update 分支 state_version 非单调: %s (期望 %s)" + % (ru["states"][0]["state_version"], i + 2)) + upd_ms.append((t5 - t4) * 1000.0) + except rtx_db.RtxError as exc: + defect.update({"seen": True, "code": exc.code, + "msg": str(exc.msg)[:160]}) + print("U[更新分支] 抛 %s → %s" % (exc.code, defect["msg"])) + break + if not defect["seen"]: + print("U[更新分支] 正常完成 %d 次覆写，state_version 单调至 %d" + % (len(upd_ms), 1 + len(upd_ms))) return hub @@ -167,21 +205,29 @@ def run(samples): % ("PASS 严格递增" if all(b > a for a, b in zip(seq_rows, seq_rows[1:])) else "FAIL 存在非递增")) print("-" * 108) - rw = summarize("write_event_with_state", write_ms) + rw = summarize("write_event_with_state[INSERT分支]", write_ms) rr = summarize("read_entity_state", read_ms) + ru = summarize("write_event_with_state[UPDATE分支]", upd_ms) print_row(rw) print_row(rr) + if upd_ms: + print_row(ru) print("-" * 108) - print("P99 门禁 SLA_P99_MS=%.1fms → write P99=%.4fms / read P99=%.4fms" - % (SLA_P99_MS, rw["p99_ms"], rr["p99_ms"])) + print("read 行 entity_id 一致性校验：%s" + % ("PASS 全部与请求一致" if not bad_rows else "FAIL 异常样本=%s" % bad_rows[:5])) + print("P99 门禁 SLA_P99_MS=%.1fms → write(INSERT) P99=%.4fms / read P99=%.4fms" + " / write(UPDATE) P99=%s" + % (SLA_P99_MS, rw["p99_ms"], rr["p99_ms"], + ("%.4fms" % ru["p99_ms"]) if upd_ms else "N/A")) if defect["seen"]: - print("已知缺陷（非本任务范围，如实记录）：%s %s" % (defect["code"], defect["msg"])) - print(" → 位置 pbl_runtime_ext/tx_write.py:_apply_state UPDATE 分支 build_row() " - "NOT NULL 全列校验；修复后需复测 UPDATE 分支 P99") - if len(write_ms) < MIN_SAMPLES: + print("缺陷复现（如实记录，未达标）：%s %s" % (defect["code"], defect["msg"])) + print(" → 位置 pbl_runtime_ext/tx_write.py:_apply_state UPDATE 分支 build_row()") + if len(write_ms) < MIN_SAMPLES or len(read_ms) < MIN_SAMPLES: print("采样不足 %d 次 → 判定不成立" % MIN_SAMPLES) return 2, rw, rr - verdict = 0 if (rw["p99_pass"] and rr["p99_pass"]) else 1 + verdict = 0 if (rw["p99_pass"] and rr["p99_pass"] and not bad_rows + and not defect["seen"] + and (not upd_ms or ru["p99_pass"])) else 1 print("P99 结论：%s" % ("达标（均 ≤ 200ms）" if verdict == 0 else "未达标，待优化")) print("=" * 108) return verdict, rw, rr diff --git a/scripts/m11b_selftest.py b/scripts/m11b_selftest.py index 827b60b..704d68f 100644 --- a/scripts/m11b_selftest.py +++ b/scripts/m11b_selftest.py @@ -217,7 +217,15 @@ class FakeEngine(object): m = re.match(r"^UPDATE (\w+) SET (.+?) WHERE (.+)$", s, re.I) if m: tbl, set_part, clause = m.groups() - cols = [c.strip().strip("`") for c in set_part.split(",")] + # M11b-2c 夹具修正：SET 子句片段形如 "col=%s"，必须只取列名（原实现把 + # "col=%s" 整体当列名，r.update 后行内多出 "col=%s" 伪键且真列未被更新， + # 导致 UPDATE 分支在内存替身上静默失效 —— 属测试夹具缺陷，非业务代码） + cols = [] + for c in set_part.split(","): + cm = re.match(r"^\s*`?(\w+)`?\s*=", c) + if not cm: + raise AssertionError("FakeEngine 无法解析 SET 片段: %r" % c) + cols.append(cm.group(1)) vals = list(params[:len(cols)]) if self.fail_on == (tbl, "update"): raise RuntimeError("注入故障：%s UPDATE 失败" % tbl)