diff --git a/.m11b2d_probe.py b/.m11b2d_probe.py new file mode 100644 index 0000000..4189487 --- /dev/null +++ b/.m11b2d_probe.py @@ -0,0 +1,66 @@ +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "tests")) +import _m11b2_compat as H # noqa: E402 + +e = H.RuntimeEngine() +H.install(e) +try: + r1 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move", + payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 2}}], + client_event_id="ev-1", actor_id="u1") + print("R1", sorted(r1.keys())) + print("R1 vals", r1["ok"], r1["dedup"], r1["event_id"], r1["seq"], r1["state_rows"], r1["stmt_count"], r1["channel"]) + print("states", r1["states"]) + print("trace", H.trace_kinds(e)) + print("counts", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE)) + print("state row", H.read(tenant_id=7, session_id=11)) + # dedup + r2 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="unit.move", + payload={"dx": 1}, state_updates=[{"entity_id": "e1", "state": {"hp": 9}}], + client_event_id="ev-1") + print("R2", r2["dedup"], r2["event_id"], r2["seq"], r2["state_rows"], sorted(r2.keys())) + print("counts after dedup", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE)) + # conflict + try: + H.apply(tenant_id=7, world_id=1, session_id=11, event_type="x", + payload={}, state_updates=[{"entity_id": "e1", "state": {"hp": 0}}], + client_event_id="ev-3", base_version=999) + print("NO CONFLICT !!") + except H.RtxError as exc: + print("conflict code", exc.code, "trace", H.trace_kinds(e)[-3:]) + print("counts after conflict", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE)) + # snapshot + r3 = H.apply(tenant_id=7, world_id=1, session_id=11, event_type="snap", + payload={}, state_updates=[{"entity_id": "e1", "state": {"hp": 5}}], + client_event_id="ev-4", snapshot={"reason": "join", "hp": 5}) + print("snapshot rows", e.count(H.SNAPSHOT_TABLE), "broadcasts", len(e.broadcasts)) + print("broadcast payload keys", sorted(r3.keys())) + print("UPDATE stmts", H.update_statements(e)) + print("errors:") + for kw in ({"tenant_id": None}, {"world_id": None}, {"session_id": None}, {"event_type": None}): + base = dict(tenant_id=7, world_id=1, session_id=11, event_type="x", payload={}) + base.update(kw) + try: + H.apply(**base) + print(" no raise for", kw) + except H.RtxError as exc: + print(" ", kw, "->", exc.code) + # bad state_updates + for su in ([{"state": {"a": 1}}], [{"entity_id": "e1"}], "notalist"): + try: + H.apply(tenant_id=7, world_id=1, session_id=11, event_type="x", payload={}, + state_updates=su, client_event_id="ev-b%d" % id(su) % 100) + print(" no raise for", su) + except H.RtxError as exc: + print(" su", su, "->", exc.code) + # missing tenant + try: + H.apply(tenant_id=None, world_id=1, session_id=11, event_type="x", payload={}) + except H.RtxError as exc: + print("tenant->", exc.code) + print("final counts", e.count(H.EVENT_TABLE), e.count(H.STATE_TABLE), e.count(H.SNAPSHOT_TABLE)) + print("trace tail", H.trace_kinds(e)[-6:]) +finally: + H.uninstall() diff --git a/tests/_m11b2_compat.py b/tests/_m11b2_compat.py index b855bbd..ff436c7 100644 --- a/tests/_m11b2_compat.py +++ b/tests/_m11b2_compat.py @@ -1,165 +1,449 @@ # -*- coding: utf-8 -*- -"""M11b-2c 测试夹具层兼容垫片（**仅测试使用，不属于模块运行时 API**）。 +"""M11b-2d-A 测试夹具层：**函数式 API 直连真源**的运行时事务引擎（仅测试使用）。 -背景（QC #3 允许路径：只改夹具/断言，不改业务实现、不改表定义 JSON） -===================================================================== -``tests/test_m11b2_single_tx.py`` 是在 M11b-2a「世界同步真源迁移」之前写成的，它 -绑定的是当时的类式写入器 API： +本文件的历史与本轮收敛（响应 QC #2 硬门禁） +========================================================== +M11b-2a「世界同步真源迁移」把写入实现迁到 **函数式 API** +``pbl_runtime_ext.tx_write.apply_runtime_event`` / ``read_states`` +（见 ``modules/pbl_runtime_ext/pbl_runtime_ext/tx_write.py:344`` / ``:466``）。 - tx.RuntimeStateEventWriter / tx.build_plan / tx.percentile / tx.configure_writer +本文件在 M11b-2c 时的形态是「类式垫片」：定义 ``RuntimeStateEventWriter`` 类， +持有 ``conn_factory`` / ``dialect`` / ``close_connection``，再把方法调用转发到 +``world_sync/pbl_runtime_tx.write_event_with_state``。那是**第二条夹具路径**—— +它把测试钉在「旧类式调用形态 + world_sync 本地同步实现」上，既没验证现行真源 +``apply_runtime_event``，又与真源返回键（``action`` / ``states[]`` / ``stmt_count`` +/ ``latency_ms``）不一致，垫片只能靠 ``_augment`` 补假键（``state_inserted`` 等） +维持旧断言，实测 22 failed（``Cannot operate on a closed database.`` / +``KeyError: 'expected_version'``）。 -M11b-2a 把真源实现改为**函数式 API**（``pbl_runtime_tx.write_event_with_state`` / -``read_entity_state``，见 ``docs/02-develop/dev-notes-m11b2a-world-sync-migration.md`` -与 ``world_sync/init.py`` 第 17 行的导出说明），类式写入器与模块级全局 writer 单例 -被一并移除——``git log -S RuntimeStateEventWriter -- world_sync/pbl_runtime_tx.py`` -为空即证据：该类从未存在于当前真源。 +**本轮已删除该类别垫片**，本文件只保留一条夹具路径： -因此本垫片**不新增业务能力**，只做三件事： -1. ``RuntimeStateEventWriter``：把「持有 conn_factory + dialect + close_connection」 - 的旧调用形态适配到当前 ``write_event_with_state`` 函数（同一条代码路径）。 -2. ``build_plan``：转发到 ``world_sync.init.build_runtime_write_plan``（已存在的 - 不落库预演函数），保持旧名字可用。 -3. ``percentile``：转发到 ``pbl_runtime_sql.percentile``（真源唯一实现，避免第二份 - 分位数算法口径漂移）。 -4. ``configure_writer``：旧「模块级默认 writer」语义在真源已不存在；这里只在**测试 - 进程内**保存一个默认连接工厂，供 ``write_event_with_state`` 的兼容包装取用， - 不写 ServerEnv、不影响生产路径。 + 1. ``RuntimeEngine``：以 **真实 sqlite3** 为存储的事务引擎。表结构（列名/类型/ + NOT NULL/DEFAULT/唯一索引）**从 ``pbl_runtime_ext/models/*.json`` 读取生成** + ——不另立第二份 schema 定义，避免列名再次漂移（这正是 28 failed 的根因之一： + 旧夹具里手写 DDL 用了 ``event_id`` / ``state`` 等真源不存在的列）。 + 2. ``install(engine)`` / ``uninstall()``：把 ``rtx_db.transaction`` / + ``rtx_db.dbname`` / ``rtx_db.q_all`` 换成 sqlite 引擎实现，使 + ``tx_write.apply_runtime_event`` / ``read_states`` **走自身真实代码路径** + （幂等预检 → INSERT 事件 → UPSERT 状态 → 快照 → COMMIT → 提交后广播）。 + 与 ``modules/pbl_runtime_ext/scripts/m11b_selftest.py::install_fake`` 同一 + 思路，差别只在存储换成真 sqlite（可验证真实 ROLLBACK 与 DB 唯一索引兜底）。 + 3. ``apply(**kw)`` / ``read(**kw)``：函数式 API 的同步入口（``asyncio.run``）。 + **不做任何返回键增补**——测试直接断言真源返回键。 + 4. ``percentile``：转发到 ``world_sync.pbl_runtime_sql.percentile``（真源唯一实现）。 -返回键差异也在垫片里收敛：真源返回 ``action``（``inserted``/``updated``/ -``event_only``），旧断言读 ``state_inserted``（bool）。垫片补 ``state_inserted`` -= ``action == "inserted"``，不改动真源返回值。 +边界（QC #3 允许路径）：只改测试夹具层，**不改业务实现、不改 models/*.json 表定义、 +不改任何 load_path.py RBAC 注册**。本文件不含任何生产能力：只有被 pytest 夹具调用时 +才安装引擎补丁，``uninstall()`` 逐样还原。 """ -import importlib.util +import asyncio +import json import os +import sqlite3 import sys -import types +import threading +import time HERE = os.path.dirname(os.path.abspath(__file__)) -PKG_DIR = os.path.dirname(HERE) -_PKG = "_ws_m11b2_compat" +WS_ROOT = os.path.dirname(HERE) # modules/world_sync +RTX_ROOT = os.path.join(os.path.dirname(WS_ROOT), "pbl_runtime_ext") # modules/pbl_runtime_ext -# 与 test_m11b2_single_tx.py 使用同一个合成包名，确保加载的是同一批模块实例 -# （否则 configure_writer 保存的默认工厂与被测函数看到的不是同一份状态）。 -_SHARED_PKG = "_ws_m11b2" +for _p in (WS_ROOT, RTX_ROOT): + if _p not in sys.path: + sys.path.insert(0, _p) +# --- 真源导入（被测对象本体，不是副本）--------------------------------- +from pbl_runtime_ext import broadcast as _bcast # noqa: E402 +from pbl_runtime_ext import m11b_config as _cfg # noqa: E402 +from pbl_runtime_ext import rtx_db as _rtx_db # noqa: E402 +from pbl_runtime_ext import tx_write as _tx_write # noqa: E402 +from world_sync import pbl_runtime_sql as ps # noqa: E402 -def _load(name, pkg=None): - pkg = pkg or _SHARED_PKG - if pkg not in sys.modules: - mod = types.ModuleType(pkg) - mod.__path__ = [PKG_DIR] - sys.modules[pkg] = mod - full = "%s.%s" % (pkg, name) - if full in sys.modules: - return sys.modules[full] - spec = importlib.util.spec_from_file_location( - full, os.path.join(PKG_DIR, name + ".py")) - mod = importlib.util.module_from_spec(spec) - sys.modules[full] = mod - spec.loader.exec_module(mod) - return mod - - -tx = _load("pbl_runtime_tx") -ps = _load("pbl_runtime_sql") - -# --- 旧名字 -> 真源实现 ------------------------------------------------- +# 旧夹具里以名字引用的常量：统一取真源，避免第二份字面量 SQLITE = ps.DIALECT_SQLITE MYSQL = ps.DIALECT_MYSQL -percentile = ps.percentile +percentile = ps.percentile # 真源唯一实现（world_sync.pbl_runtime_sql） +apply_runtime_event = _tx_write.apply_runtime_event # 现行函数式真源 API +read_states = _tx_write.read_states +build_row = _tx_write.build_row +assert_append_only = _tx_write.assert_append_only +RtxError = _rtx_db.RtxError -_write_event_with_state = tx.write_event_with_state -_read_entity_state = tx.read_entity_state +EVENT_TABLE = _cfg.EVENT_TABLE +STATE_TABLE = _cfg.STATE_TABLE +SNAPSHOT_TABLE = _cfg.SNAPSHOT_TABLE +PROTECTED_BASE_TABLES = _cfg.PROTECTED_BASE_TABLES +ERR_TENANT_MISSING = _cfg.ERR_TENANT_MISSING +ERR_BAD_PARAM = _cfg.ERR_BAD_PARAM +ERR_STATE_CONFLICT = _cfg.ERR_STATE_CONFLICT +ERR_APPEND_ONLY = _cfg.ERR_APPEND_ONLY +ERR_BROADCAST_AFTER_COMMIT = _cfg.ERR_BROADCAST_AFTER_COMMIT +# build_row 的三道门禁码（真源 tx_write.py 内以字面量抛出） +ERR_UNKNOWN_COLUMN = "PBL-RTX-0003" +ERR_MISSING_NOT_NULL = "PBL-RTX-0004" +ERR_IMMUTABLE_COLUMN = "PBL-RTX-0006" -# 测试进程内的「默认连接工厂」（仅服务于旧 configure_writer 语义） -_default = {"factory": None, "dialect": None, "close_connection": True} +_MODELS_DIR = os.path.join(RTX_ROOT, "models") -def configure_writer(conn_factory=None, dialect=None, close_connection=True, - **_ignored): - """旧模块级默认 writer 语义的测试内等价物：登记默认连接工厂。 - - 真源已无全局 writer 单例（见 ``world_sync/init.py:356`` 的 - ``configure_runtime_writer`` —— 它登记的是 ServerEnv 契约，不是类实例）。 - 本函数只在测试进程内记一个 factory，供同名兼容包装使用。 - """ - _default["factory"] = conn_factory - _default["dialect"] = dialect - _default["close_connection"] = close_connection - return conn_factory +# ---------------------------------------------------------------- schema +def _sql_type(t): + """抽象类型 → sqlite 存储类（sqlite 动态类型，映射只为可读性/亲和性）。""" + t = str(t or "").lower() + if t in ("long", "int", "bigint", "smallint", "short"): + return "INTEGER" + if t in ("float", "double", "ddouble", "decimal"): + return "REAL" + if t in ("text", "json"): + return "TEXT" + if t in ("date", "datetime", "time", "timestamp"): + return "TEXT" + return "TEXT" -def write_event_with_state(event_data=None, state_update=None, conn_factory=None, - conn=None, event_uid=None, source=None, clock=None): - """兼容包装：未显式给 conn/conn_factory 时取 ``configure_writer`` 登记的工厂。""" - if conn is None and conn_factory is None: - conn_factory = _default["factory"] - res = _write_event_with_state(event_data, state_update, - conn_factory=conn_factory, conn=conn, - event_uid=event_uid, source=source, clock=clock) - return _augment(res) +def _not_null(fd): + if fd.get("null") is False: + return True + return str(fd.get("nullable") or "").strip().lower() in ("no", "false", "0") -def _augment(result): - """补旧断言用到的键，不改动真源已有键的语义。""" - if isinstance(result, dict): - action = result.get("action") - result.setdefault("state_inserted", action == "inserted") - result.setdefault("state_updated", action == "updated") - result.setdefault("event_only", action == "event_only") - return result +def _default_clause(fd): + dv = fd.get("default") + if dv is None: + return "" + s = str(dv).strip() + if s.upper() in ("CURRENT_TIMESTAMP", "NULL", "NONE", ""): + return "" + if isinstance(dv, bool): + return " DEFAULT %d" % int(dv) + if isinstance(dv, (int, float)): + return " DEFAULT %s" % dv + return " DEFAULT '%s'" % s.replace("'", "''") -class RuntimeStateEventWriter(object): - """旧类式写入器的测试适配：持有 conn_factory/dialect/close_connection， - 方法体直接调用真源函数式 API（同一条代码路径，不复制业务逻辑）。 +def model_fields(table): + """表列定义（权威源 = pbl_runtime_ext/models/{table}.json）。""" + path = os.path.join(_MODELS_DIR, "%s.json" % table) + with open(path, "r", encoding="utf-8") as fh: + return json.load(fh) + + +def ddl_statements(table): + """按 models/*.json 生成 sqlite DDL（列/NOT NULL/DEFAULT/唯一索引）。""" + spec = model_fields(table) + cols = [] + for fd in spec.get("fields") or []: + name = fd.get("name") + if not name: + continue + if fd.get("auto_increment"): + cols.append('"%s" INTEGER PRIMARY KEY AUTOINCREMENT' % name) + continue + d = '"%s" %s' % (name, _sql_type(fd.get("type"))) + if _not_null(fd): + d += " NOT NULL" + d += _default_clause(fd) + cols.append(d) + out = ['CREATE TABLE IF NOT EXISTS "%s" (\n %s\n)' % (table, ",\n ".join(cols))] + for idx in spec.get("indexes") or []: + fields = idx.get("idxfields") or [] + if not fields: + continue + kind = "UNIQUE INDEX" if str(idx.get("idxtype")) == "unique" else "INDEX" + cols_sql = ", ".join('"%s"' % f for f in fields) + out.append('CREATE %s IF NOT EXISTS "%s" ON "%s" (%s)' + % (kind, idx.get("name") or ("%s_%d" % (table, len(out))), table, cols_sql)) + return out + + +def _to_qmark(sql): + """真源 SQL 用 sqlor 的 ``%s`` 占位；sqlite 用 ``?``。""" + return sql.replace("%s", "?") + + +# ---------------------------------------------------------------- 引擎 +class RuntimeEngine(object): + """真实 sqlite 事务引擎（测试专用），适配 ``rtx_db.Transaction`` 接口。 + + 记录三条可断言的轨迹： + * ``trace`` 有序事务轨迹：``BEGIN`` / ``SQL`` / ``COMMIT`` / ``ROLLBACK`` / + ``BROADCAST`` —— 用于「广播只在 commit 之后」「BEGIN/COMMIT 配平」 + * ``sql_log`` 实际发出的 SQL（含参数），用于「SET 子句不含不可变列」「无广播/ + 轮询/睡眠语句」这类语句级断言 + * ``broadcasts`` 提交后广播消息（hub.publish 录制） """ - def __init__(self, conn_factory=None, dialect=None, close_connection=True, - source=None, clock=None, hooks=None, **extra): - self.conn_factory = conn_factory - self.dialect = dialect - self.close_connection = close_connection - self.source = source - self.clock = clock - self.hooks = hooks or {} - self._extra = dict(extra) - self.calls = [] + def __init__(self, path=":memory:", tables=(EVENT_TABLE, STATE_TABLE, SNAPSHOT_TABLE)): + self.path = path + self.tables = tuple(tables) + self.trace = [] + self.sql_log = [] + self.broadcasts = [] + self.txn_depth = 0 + self._lock = threading.RLock() + self._mem = None + if path == ":memory:": + self._mem = sqlite3.connect(":memory:", isolation_level=None, + check_same_thread=False) + self._mem.execute("PRAGMA busy_timeout=15000") + self._create(self._mem) + else: + c = sqlite3.connect(path, isolation_level=None) + c.execute("PRAGMA busy_timeout=15000") + self._create(c) + c.close() - def write_event_with_state(self, event_data=None, state_update=None, **kw): - conn = kw.pop("conn", None) - event_uid = kw.pop("event_uid", self._extra.get("event_uid")) - source = kw.pop("source", self.source) - clock = kw.pop("clock", self.clock) - conn_factory = kw.pop("conn_factory", None) or self.conn_factory - pre = self.hooks.get("before_write") - if callable(pre): - pre(event_data, state_update) - if conn is None and conn_factory is None: - conn_factory = _default["factory"] - post = self.hooks.get("after_write") - result = _write_event_with_state(event_data, state_update, - conn_factory=conn_factory, conn=conn, - event_uid=event_uid, source=source, - clock=clock) - self.calls.append(result.get("event_uid")) - if callable(post): - post(result) - return _augment(result) + # -- 建表 ------------------------------------------------------------- + def _create(self, conn): + for t in self.tables: + for stmt in ddl_statements(t): + conn.execute(stmt) + conn.commit() - # 旧代码里偶有以 read_state 名称取当前状态的用法 - def read_entity_state(self, tenant_id=None, session_id=None, entity_id=None, - world_id=None): - res = _read_entity_state(conn_factory=self.conn_factory, - tenant_id=tenant_id, session_id=session_id, - entity_id=entity_id, world_id=world_id) - return res + # -- 连接 ------------------------------------------------------------- + def admin_conn(self): + """给测试直接查库用的连接（文件库返回新连接，调用方负责 close）。""" + if self._mem is not None: + return self._mem, False + c = sqlite3.connect(self.path, isolation_level=None) + c.execute("PRAGMA busy_timeout=15000") + return c, True + + def query(self, sql, params=()): + conn, own = self.admin_conn() + try: + cur = conn.execute(_to_qmark(sql), tuple(params or ())) + cols = [d[0] for d in cur.description] if cur.description else [] + return [dict(zip(cols, r)) for r in cur.fetchall()] + finally: + if own: + conn.close() + + def scalar(self, sql, params=(), default=None): + rows = self.query(sql, params) + if not rows: + return default + vals = list(rows[0].values()) + return vals[0] if vals else default + + def count(self, table): + return int(self.scalar("SELECT COUNT(*) AS c FROM \"%s\"" % table, (), 0) or 0) + + def seed_state(self, tenant_id, session_id, entity_id, state_json, + state_version=0, checksum="", updated_by="seed"): + """预置状态行（走真实 INSERT，不绕过列契约）。""" + now = time.strftime("%Y-%m-%d %H:%M:%S") + conn, own = self.admin_conn() + try: + conn.execute("DELETE FROM \"%s\" WHERE tenant_id=? AND session_id=? AND entity_id=?" + % STATE_TABLE, (str(tenant_id), str(session_id), str(entity_id))) + conn.execute( + "INSERT INTO \"%s\" (tenant_id, session_id, entity_id, state_json," + " state_version, checksum, updated_by, created_at, updated_at)" + " VALUES (?,?,?,?,?,?,?,?,?)" % STATE_TABLE, + (str(tenant_id), str(session_id), str(entity_id), state_json, + int(state_version), checksum, updated_by, now, now)) + conn.commit() + finally: + if own: + conn.close() + + # -- 事务轨迹 --------------------------------------------------------- + def _note(self, kind, detail=None): + with self._lock: + self.trace.append((kind, detail)) + + # -- 只读查询（供 rtx_db.q_all 补丁）---------------------------------- + async def q_all(self, sql, params=None, db=None): + return _rtx_db._norm_rows(self.query(sql, params)) -def build_plan(event_data=None, state_update=None, event_uid=None, source=None, - now=None): - """旧 build_plan 名字 -> 真源 init.build_runtime_write_plan（不落库预演）。""" - init_mod = _load("init", pkg=_PKG) - return init_mod.build_runtime_write_plan( - event_data, state_update, event_uid=event_uid, source=source, now=now) +class _Tx(object): + """sqlite 事务句柄：与 ``rtx_db.Transaction`` 同接口（execute/insert/update/ + select/commit/rollback/stmt_count），**stmt_count 只计 DML**（与真源一致： + 真源 ``execute`` 不计数、``insert``/``update`` 计数）。 + """ + + def __init__(self, engine): + self.e = engine + self.conn = None + self.committed = False + self.rolled_back = False + self.stmt_count = 0 + + async def __aenter__(self): + self.conn = self.e._mem if self.e._mem is not None else sqlite3.connect( + self.e.path, isolation_level=None) + if self.e._mem is None: + self.conn.execute("PRAGMA busy_timeout=15000") + with self.e._lock: + self.conn.execute("BEGIN IMMEDIATE") + self.e.txn_depth += 1 + self.e._note("BEGIN") + return self + + async def __aexit__(self, exc_type, exc, tb): + if exc_type is not None or not self.committed: + await self.rollback() + self._close() + return False + + def _close(self): + if self.e._mem is None and self.conn is not None: + try: + self.conn.close() + except Exception: # noqa: BLE001 + pass + self.conn = None + + def _run(self, sql, params): + self.e.sql_log.append((" ".join(str(sql).split()), tuple(params or ()))) + cur = self.conn.cursor() + try: + cur.execute(_to_qmark(sql), tuple(params or ())) + cols = [d[0] for d in cur.description] if cur.description else [] + rows = [dict(zip(cols, r)) for r in cur.fetchall()] if cols else [] + return rows + finally: + try: + cur.close() + except Exception: # noqa: BLE001 + pass + + # -- DML / 查询 ------------------------------------------------------- + async def execute(self, sql, params=None): + return self._run(sql, params) + + async def insert(self, table, row): + row = dict(row) + if "tenant_id" in row and row.get("tenant_id") in (None, "", 0): + raise RtxError(ERR_TENANT_MISSING, "%s 写入缺少 tenant_id" % table) + cols = list(row.keys()) + sql = 'INSERT INTO "%s" (%s) VALUES (%s)' % ( + table, ", ".join('"%s"' % c for c in cols), ", ".join(["?"] * len(cols))) + self.e.sql_log.append((" ".join(sql.split()), tuple(row[c] for c in cols))) + cur = self.conn.cursor() + try: + cur.execute(sql, tuple(row[c] for c in cols)) + rid = cur.lastrowid + finally: + try: + cur.close() + except Exception: # noqa: BLE001 + pass + self.stmt_count += 1 + return rid + + async def update(self, table, values, where): + cols = list(dict(values).keys()) + sql = 'UPDATE "%s" SET %s' % (table, ", ".join('"%s" = ?' % c for c in cols)) + self.e.sql_log.append((" ".join(sql.split()), tuple(dict(values)[c] for c in cols))) + cur = self.conn.cursor() + try: + cur.execute(sql, tuple(dict(values)[c] for c in cols)) + n = cur.rowcount + finally: + try: + cur.close() + except Exception: # noqa: BLE001 + pass + self.stmt_count += 1 + return n + + async def select(self, table, where="", fields="*", order="", limit=200): + sql = 'SELECT %s FROM "%s"' % (fields or "*", table) + if where: + sql += " WHERE %s" % where + if order: + sql += " ORDER BY %s" % order + sql += " LIMIT %d" % int(limit or 200) + return self._run(sql, ()) + + # -- 事务控制 --------------------------------------------------------- + async def commit(self): + if self.committed or self.rolled_back: + return + self.conn.commit() + self.committed = True + with self.e._lock: + self.e.txn_depth -= 1 + self.e._note("COMMIT") + + async def rollback(self): + if self.committed or self.rolled_back: + return + try: + self.conn.rollback() + except Exception: # noqa: BLE001 + pass + self.rolled_back = True + with self.e._lock: + self.e.txn_depth -= 1 + self.e._note("ROLLBACK") + + +# ---------------------------------------------------------------- 安装 / 还原 +_ORIG = {} +_INSTALLED = False + + +def install(engine): + """把 ``rtx_db`` 的事务/查询入口换成 sqlite 引擎，并录制提交后广播。""" + global _INSTALLED + if _INSTALLED: + uninstall() + _ORIG["transaction"] = _rtx_db.transaction + _ORIG["dbname"] = _rtx_db.dbname + _ORIG["q_all"] = _rtx_db.q_all + _rtx_db.dbname = lambda env=None: "m11b2d_test_sqlite" + _rtx_db.transaction = lambda db=None: _Tx(engine) + _rtx_db.q_all = engine.q_all + + _bcast.reset_hub() + hub = _bcast.get_hub() + original_publish = hub.publish + + def _recording_publish(channel, event): + engine.broadcasts.append({"channel": channel, "event": event}) + engine._note("BROADCAST", channel) + return original_publish(channel, event) + + hub.publish = _recording_publish + _ORIG["hub"] = (hub, original_publish) + _INSTALLED = True + return engine + + +def uninstall(): + global _INSTALLED + if not _INSTALLED: + return + for key in ("transaction", "dbname", "q_all"): + if key in _ORIG: + setattr(_rtx_db, key, _ORIG[key]) + _INSTALLED = False + _ORIG.clear() + + +def apply(**kwargs): + """函数式真源 API 的同步入口（不做返回键增补）。""" + return asyncio.run(apply_runtime_event(**kwargs)) + + +def read(**kwargs): + return asyncio.run(read_states(**kwargs)) + + +def trace_kinds(engine): + return [k for k, _d in engine.trace] + + +def update_statements(engine): + """引擎实际发出的 UPDATE 语句（大写、压空格），用于不可变列/乐观锁断言。""" + return [(s, p) for s, p in engine.sql_log if s.upper().startswith("UPDATE ")] + + +def insert_statements(engine, table): + return [(s, p) for s, p in engine.sql_log + if s.upper().startswith('INSERT INTO "%s"' % table.upper())]