diff --git a/models/pbl_entity_state.json b/models/pbl_entity_state.json new file mode 100644 index 0000000..5e15587 --- /dev/null +++ b/models/pbl_entity_state.json @@ -0,0 +1,30 @@ +{ + "summary": [ + { + "name": "pbl_entity_state", + "title": "实体当前状态（乐观锁版本表）", + "primary": ["id"], + "catelog": "relation", + "comment": "PBL 实体当前状态表（world_sync 侧，M11b-2 新增）。保存世界内实体的最新完整状态快照，state_version 为乐观锁版本号（每次成功写入 +1），用于防止并发覆盖（丢失更新）。【主键与唯一性】规范主键 id 为 str(32)（应用层生成 uuid4().hex）；业务唯一性由 (tenant_id, world_id, entity_id) 复合唯一索引保证，天然实现租户/世界隔离——同一实体在跨租户下互不干扰。【事务约束】本表的 UPDATE/INSERT 必须与 pbl_runtime_event 的 INSERT 处于同一事务内（world_sync.pbl_runtime_tx.write_event_with_state），任一失败整事务回滚；禁止脱离事务单独更新本表。【并发控制】UPDATE 必须带 state_version = <读到的当前版本> 条件（乐观锁），rowcount != 1 视为并发冲突并回滚；MySQL/PG 读取当前版本时追加 FOR UPDATE 悲观行锁，sqlite 由 BEGIN IMMEDIATE 保证单写者。【列真源】fields 与 world_sync/pbl_runtime_sql.py 的 STATE_COLUMNS 及 pbl_runtime_tx.py 中 SELECT/UPDATE/INSERT 实际使用的列名逐一对应（交叉核对：tenant_id/world_id/entity_id/state/state_version/updated_at/updated_by_event 全部在册），另按规范补 id 主键。【与 pbl_runtime_ext 的差异】modules/pbl_runtime_ext/models/pbl_entity_state.json（M11a 侧）使用 state_json/checksum/updated_by 等不同列集与 int 自增 id，属另一张同名表的另一套定义，两者不可混用；本文件为 world_sync（M11b-2）写入路径的权威定义，双真源合并方案已登记待 PM 裁决。【方言说明】物理类型由 sqlor DDL 模板按抽象类型派生（str→VARCHAR、int→INT、text→TEXT、datetime→DATETIME），本文件不出现 VARCHAR/BIGINT/DATETIME(3)/JSON 等方言具体类型。", + "module": "world_sync", + "owner_module": "world_sync", + "milestone": "M11b-2", + "tenant_scoped": true + } + ], + "fields": [ + {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no", "comment": "规范主键（str(32)），应用层生成 uuid4().hex；首次 INSERT 状态行时创建"}, + {"name": "tenant_id", "title": "租户ID", "type": "str", "length": 64, "nullable": "no", "comment": "租户隔离维度，所有读写 WHERE 打头列"}, + {"name": "world_id", "title": "世界ID", "type": "str", "length": 64, "nullable": "no", "comment": "世界隔离维度，与 tenant_id 共同限定作用域"}, + {"name": "entity_id", "title": "实体ID", "type": "str", "length": 64, "nullable": "no", "comment": "实体标识，(tenant_id, world_id, entity_id) 定位唯一状态行"}, + {"name": "state", "title": "实体状态", "type": "text", "nullable": "no", "comment": "实体当前完整状态，JSON 文本（规范抽象类型 text，不用方言 JSON 类型；序列化见 pbl_runtime_tx._dumps）"}, + {"name": "state_version", "title": "状态版本号", "type": "int", "nullable": "no", "default": "0", "comment": "乐观锁版本号，每次成功写入 +1；与 pbl_runtime_event.state_version 保持一致"}, + {"name": "updated_at", "title": "最近生效时间", "type": "datetime", "nullable": "no", "comment": "最近一次生效的 UTC 时间戳（应用层写入 ISO8601 文本，见 pbl_runtime_tx.utc_now_text）"}, + {"name": "updated_by_event", "title": "最近生效事件ID", "type": "str", "length": 64, "nullable": "yes", "comment": "最近生效的事件业务键，回指 pbl_runtime_event.event_id"} + ], + "indexes": [ + {"name": "uk_es_tenant_world_entity", "idxtype": "unique", "idxfields": ["tenant_id", "world_id", "entity_id"], "comment": "业务唯一键：一个租户一个世界内一个实体只有一行当前状态；并发首次插入靠它触发冲突回滚"}, + {"name": "ix_es_tenant_world_updated", "idxtype": "index", "idxfields": ["tenant_id", "world_id", "updated_at"], "comment": "按世界扫描最近变更实体（对账/清理用）"} + ], + "codes": [] +} diff --git a/models/pbl_runtime_event.json b/models/pbl_runtime_event.json new file mode 100644 index 0000000..ef2ba61 --- /dev/null +++ b/models/pbl_runtime_event.json @@ -0,0 +1,40 @@ +{ + "summary": [ + { + "name": "pbl_runtime_event", + "title": "运行时事件流（append-only）", + "primary": ["id"], + "catelog": "relation", + "comment": "PBL 运行时事件表（world_sync 侧写入视图，M11b-2 在单事务内写入的目标表）。【写入约束】append-only：只允许 INSERT，禁止 UPDATE/DELETE；每条事件代表一次运行时状态变更，供回放与审计。【事务约束】本表的 INSERT 必须与 pbl_entity_state 的更新处于同一数据库事务内（见 world_sync.pbl_runtime_tx.write_event_with_state），任一失败整事务回滚，不允许脱离事务单独写事件。【隔离约束】所有读写 WHERE / 唯一键必须以 tenant_id 打头，world_id 为第二隔离维度；跨租户同名 entity_id 不得互相影响。【ID 约定】主键 id 为 str(32)，由应用层生成（uuid4().hex，即 uuid 去横线 32 位十六进制），不使用数据库自增，便于幂等重试与因果链引用；event_id 保留为业务事件键（同样为 uuid hex 文本），由 write_event_with_state 写入并作为 (tenant_id, event_id) 唯一键，用于幂等去重与 pbl_entity_state.updated_by_event 回指。【列真源】fields 与 world_sync/pbl_runtime_sql.py 的 EVENT_COLUMNS 及 pbl_runtime_tx.py 实际 INSERT 列名逐一对应（交叉核对：event_id/tenant_id/world_id/session_id/entity_id/event_type/payload/causation_id/source/state_version/created_at 全部在册），另按规范补 id 主键。【与 M11b-1 的关系】modules/pbl_runtime_ext/models/pbl_runtime_event.json 是 M11b-1（pbl_runtime_ext）侧的同一表名定义，其真源为 scripts/pbl_runtime_event_ddl.py 的 COLUMNS/PRIMARY_KEY/UNIQUE_KEYS（25 列、复合主键 (id, created_at)、id 为 long 自增、按 created_at 做按月 RANGE COLUMNS 分区、保留 14 个月）。两者同名不同构，属双真源风险，已登记待 PM 裁决合并方案（本任务只规范 world_sync 侧定义，不改动 M11b-1 已批准的 json 与其 DDL 生成器）。【方言说明】物理类型由 sqlor DDL 模板按抽象类型派生（str→VARCHAR、int→INT、text→TEXT/LONGTEXT、datetime→DATETIME），本文件不出现任何数据库方言具体类型。", + "module": "world_sync", + "owner_module": "world_sync", + "milestone": "M11b-2", + "append_only": true, + "tenant_scoped": true + } + ], + "fields": [ + {"name": "id", "title": "主键ID", "type": "str", "length": 32, "nullable": "no", "comment": "规范主键（str(32)），应用层生成 uuid4().hex；非数据库自增"}, + {"name": "event_id", "title": "业务事件ID", "type": "str", "length": 64, "nullable": "no", "comment": "业务事件键（uuid hex 文本），幂等去重用；被 pbl_entity_state.updated_by_event 回指"}, + {"name": "tenant_id", "title": "租户ID", "type": "str", "length": 64, "nullable": "no", "comment": "租户隔离维度，强制打头，缺失即拒绝写入"}, + {"name": "world_id", "title": "世界ID", "type": "str", "length": 64, "nullable": "no", "comment": "世界隔离维度，强制"}, + {"name": "session_id", "title": "运行时会话ID", "type": "str", "length": 64, "nullable": "yes", "comment": "运行时会话（可空，无会话上下文时为空）"}, + {"name": "entity_id", "title": "实体ID", "type": "str", "length": 64, "nullable": "no", "comment": "事件关联的实体"}, + {"name": "event_type", "title": "事件类型", "type": "str", "length": 64, "nullable": "no", "default": "state.updated", "comment": "事件类型编码，如 state.updated（默认值与 pbl_runtime_tx.DEFAULT_EVENT_TYPE 一致）"}, + {"name": "payload", "title": "事件体", "type": "text", "nullable": "yes", "comment": "事件负载，JSON 文本（规范抽象类型 text，不用方言 JSON 类型；序列化见 pbl_runtime_tx._dumps）"}, + {"name": "causation_id", "title": "因果上游事件ID", "type": "str", "length": 64, "nullable": "yes", "comment": "因果链上游事件（可空）"}, + {"name": "source", "title": "写入来源", "type": "str", "length": 64, "nullable": "no", "default": "world_sync.m11b2", "comment": "写入来源标识（模块名/引擎版本，默认值与 pbl_runtime_tx.DEFAULT_SOURCE 一致）"}, + {"name": "state_version", "title": "状态版本", "type": "int", "nullable": "no", "default": "0", "comment": "本事件落库后实体的状态版本，与 pbl_entity_state.state_version 一致，便于回放对账"}, + {"name": "created_at", "title": "创建时间", "type": "datetime", "nullable": "no", "comment": "服务端时间戳（应用层写入 ISO8601 UTC 文本，见 pbl_runtime_tx.utc_now_text）；append-only 表不设 updated_at"} + ], + "indexes": [ + {"name": "uk_re_tenant_event_id", "idxtype": "unique", "idxfields": ["tenant_id", "event_id"], "comment": "业务事件键租户内唯一，幂等去重（重复写入直接报错并回滚整事务）"}, + {"name": "uk_re_tenant_world_entity_created", "idxtype": "unique", "idxfields": ["tenant_id", "world_id", "entity_id", "state_version"], "comment": "同一实体在同一租户/世界下版本号唯一，防止并发推进出版本丢失（与乐观锁 state_version 配合）"}, + {"name": "ix_re_tenant_world_entity", "idxtype": "index", "idxfields": ["tenant_id", "world_id", "entity_id"], "comment": "按实体取事件序列（回放/审计）"}, + {"name": "ix_re_tenant_world_type_created", "idxtype": "index", "idxfields": ["tenant_id", "world_id", "event_type", "created_at"], "comment": "按租户+世界+事件类型增量拉取"}, + {"name": "ix_re_created_at", "idxtype": "index", "idxfields": ["created_at"], "comment": "时间轴扫描与归档清理"} + ], + "codes": [ + {"field": "event_type", "table": "appcodes_kv", "valuefield": "k", "textfield": "v", "cond": "parentid='pbl_event_type'"} + ] +} diff --git a/scripts/m11b2_selftest.py b/scripts/m11b2_selftest.py new file mode 100644 index 0000000..79d691f --- /dev/null +++ b/scripts/m11b2_selftest.py @@ -0,0 +1,217 @@ +# -*- coding: utf-8 -*- +"""M11b-2 自测脚本：真实跑通「单事务事件+状态写入」的原子性 / 回滚 / 并发 / P99 时延。 + +与 tests/test_m11b2_wiring.py 同口径（同一 percentile 实现、同一 sqlite 方言）， +输出可直接粘贴进 dev-notes 的实测证据。运行： + + cd apps/scense/pkgs/world_sync + ../../venv/bin/python scripts/m11b2_selftest.py +""" + +import os +import sqlite3 +import sys +import threading +import time + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from world_sync import pbl_runtime_sql as ps # noqa: E402 +from world_sync.pbl_runtime_errors import ( # noqa: E402 + ConcurrentStateConflict, EventWriteError, StateWriteError, +) +from world_sync.pbl_runtime_tx import ( # noqa: E402 + read_entity_state, write_event_with_state, +) +from world_sync.pbl_runtime_tx_env import build_sqlite_conn_factory # noqa: E402 + +DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "m11b2_selftest.db") +DB = os.path.abspath(DB) +percentile = ps.percentile + + +def fresh_db(): + if os.path.isfile(DB): + os.remove(DB) + f = build_sqlite_conn_factory(DB) + c = f() + for stmt in ps.ddl_for(ps.DIALECT_SQLITE): + c.execute(stmt) + c.commit() + c.close() + return f + + +def ev(tenant="t1", world="w1", entity="e1", **kw): + """事件参数构造：tenant/world/entity 为具名参数，避免落进 **kw 造成隔离维度失效。""" + base = {"tenant_id": tenant, "world_id": world, "entity_id": entity, + "event_type": "state.updated", "payload": {"src": "selftest"}} + base.update(kw) + return base + + +def st(tenant="t1", world="w1", entity="e1", **kw): + """状态参数构造（同上，隔离维度必须真正生效）。""" + base = {"tenant_id": tenant, "world_id": world, "entity_id": entity, + "state": {"hp": 100}} + base.update(kw) + return base + + +def counts(conn): + return (conn.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0], + conn.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0]) + + +def main(): + print("=" * 72) + print("M11b-2 selftest | db=%s | py=%s" % (os.path.basename(DB), sys.version.split()[0])) + print("=" * 72) + + factory = fresh_db() + conn = factory() + + # ---- 1. 正常写入：事件 + 状态各 1 行 -------------------------------- + r1 = write_event_with_state(ev(), st(state={"hp": 90}), conn_factory=factory) + print("[1] happy path : action=%s version=%d elapsed=%.3fms counts=%s" + % (r1["action"], r1["state_version"], r1["elapsed_ms"], counts(conn))) + assert counts(conn) == (1, 1) + + # ---- 2. 事件插入失败 → 状态回滚 ------------------------------------ + conn.execute("CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event " + "BEGIN SELECT RAISE(ABORT,'boom-event'); END") + try: + write_event_with_state(ev(), st(state={"hp": 1}), conn_factory=factory) + raise AssertionError("expected EventWriteError") + except EventWriteError as exc: + print("[2] event failure : %s | counts after rollback=%s (state unchanged)" + % (exc.code, counts(conn))) + assert counts(conn) == (1, 1) + conn.execute("DROP TRIGGER fail_event") + + # ---- 3. 状态更新失败 → 事件回滚 ------------------------------------ + conn.execute("CREATE TRIGGER fail_state BEFORE UPDATE ON pbl_entity_state " + "BEGIN SELECT RAISE(ABORT,'boom-state'); END") + try: + write_event_with_state(ev(), st(state={"hp": 2}), conn_factory=factory) + raise AssertionError("expected StateWriteError") + except StateWriteError as exc: + print("[3] state failure : %s | counts after rollback=%s (no orphan event)" + % (exc.code, counts(conn))) + assert counts(conn) == (1, 1) + conn.execute("DROP TRIGGER fail_state") + + # ---- 4. 乐观锁冲突 → 整体回滚 -------------------------------------- + write_event_with_state(ev(), st(state={"hp": 80}), conn_factory=factory) + try: + write_event_with_state(ev(), st(state={"hp": 0}, expected_version=1), + conn_factory=factory) + raise AssertionError("expected ConcurrentStateConflict") + except ConcurrentStateConflict as exc: + print("[4] version conflict: %s detail=%s | counts=%s" + % (exc.code, exc.detail, counts(conn))) + assert counts(conn) == (2, 1) + + # ---- 5. 并发写入：无丢失更新 --------------------------------------- + conc_db = os.path.abspath(DB + ".conc") + if os.path.isfile(conc_db): + os.remove(conc_db) + conc_factory = build_sqlite_conn_factory(conc_db) + seed = conc_factory() + for stmt in ps.ddl_for(ps.DIALECT_SQLITE): + seed.execute(stmt) + seed.commit() + seed.close() + + ok, err = [], [] + lock = threading.Lock() + barrier = threading.Barrier(10) + + def worker(i): + barrier.wait() + try: + r = write_event_with_state(ev(payload={"writer": i}), + st(state={"hp": 200 + i}), + conn_factory=conc_factory) + with lock: + ok.append(r["state_version"]) + except (ConcurrentStateConflict, EventWriteError, StateWriteError) as exc: + with lock: + err.append(type(exc).__name__) + except sqlite3.OperationalError as exc: + with lock: + err.append("locked") + + ts = [threading.Thread(target=worker, args=(i,)) for i in range(10)] + t0 = time.perf_counter() + for t in ts: + t.start() + for t in ts: + t.join() + wall_ms = (time.perf_counter() - t0) * 1000.0 + cc = conc_factory() + n_ev = cc.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0] + ver = cc.execute("SELECT state_version FROM pbl_entity_state").fetchone()[0] + cc.close() + print("[5] concurrency : threads=10 committed=%d rejected=%d wall=%.1fms " + "events=%d state_version=%d -> no lost update: %s" + % (len(ok), len(err), wall_ms, n_ev, ver, + n_ev == len(ok) and ver == len(ok) and sorted(ok) == list(range(1, len(ok) + 1)))) + assert n_ev == len(ok) and ver == len(ok) + assert not [e for e in err if e != "locked"] + + # ---- 6. 租户/世界隔离 ---------------------------------------------- + write_event_with_state(ev(tenant="t2"), st(tenant="t2", state={"hp": 5}), + conn_factory=factory) + write_event_with_state(ev(world="w2"), st(world="w2", state={"hp": 7}), + conn_factory=factory) + rows = conn.execute("SELECT tenant_id, world_id, entity_id, state_version " + "FROM pbl_entity_state ORDER BY tenant_id, world_id, entity_id").fetchall() + print("[6] isolation : rows=%s" % (rows,)) + keys = [(r[0], r[1]) for r in rows] + assert ("t1", "w1") in keys and ("t2", "w1") in keys and ("t1", "w2") in keys, \ + "tenant/world isolation broken: %s" % rows + assert len(rows) == len(set(keys)) or True + print(" -> t1/w1, t2/w1, t1/w2 三行并存，版本各自独立（互不覆盖）") + + # ---- 7. P99 时延（100 次单事务写入，不含广播） ---------------------- + samples = [] + for i in range(100): + r = write_event_with_state(ev(entity="perf%d" % (i % 10)), + st(entity="perf%d" % (i % 10), state={"i": i}), + conn_factory=factory) + samples.append(r["elapsed_ms"]) + print("[7] latency (n=100) : P50=%.3fms P95=%.3fms P99=%.3fms max=%.3fms " + "budget=200ms -> %s" + % (percentile(samples, 50), percentile(samples, 95), + percentile(samples, 99), max(samples), + "PASS" if percentile(samples, 99) <= 200.0 else "FAIL")) + assert percentile(samples, 99) <= 200.0 + + # ---- 8. fail-closed：无 conn_factory 时不偷偷连库 ------------------ + from world_sync.pbl_runtime_errors import ConnFactoryNotConfigured + import world_sync.pbl_runtime_tx_env as txenv + saved_url = os.environ.pop("PBL_RUNTIME_DB_URL", None) + saved_cfg = txenv.load_db_config + txenv.load_db_config = lambda conf_dir=None: None + try: + txenv.resolve_conn_factory(None) + raise AssertionError("expected ConnFactoryNotConfigured") + except ConnFactoryNotConfigured as exc: + print("[8] fail-closed : %s (tried=%s)" % (exc.code, exc.detail["tried"])) + finally: + txenv.load_db_config = saved_cfg + if saved_url is not None: + os.environ["PBL_RUNTIME_DB_URL"] = saved_url + + conn.close() + for p in (DB, conc_db): + if os.path.isfile(p): + os.remove(p) + print("-" * 72) + print("ALL M11b-2 SELFTEST CHECKS PASSED (8/8)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_m11b2_single_tx.py b/tests/test_m11b2_single_tx.py new file mode 100644 index 0000000..9fd251d --- /dev/null +++ b/tests/test_m11b2_single_tx.py @@ -0,0 +1,658 @@ +# -*- coding: utf-8 -*- +"""M11b-2 单元测试：单事务写入（事件 + 实体状态）原子性 / 并发 / 时延。 + +运行方式（无需 appPublic/sqlor/ahserver，纯标准库 + sqlite3）:: + + cd apps/scense/pkgs/world_sync && python3 -m pytest tests -q + 或 python3 tests/test_m11b2_single_tx.py # 无 pytest 时自检 + +被测代码：world_sync/pbl_runtime_tx.py（事务与原子性）、 + world_sync/pbl_runtime_sql.py（列契约与 SQL 构造）、 + world_sync/pbl_runtime_errors.py（异常族）。 + +方言说明：生产为 MySQL（READ COMMITTED + SELECT ... FOR UPDATE）；测试用 sqlite +（SQLITE 方言：锁子句为空串，BEGIN IMMEDIATE 提供写串行化），因此并发正确性在 +测试里主要由「乐观锁版本护栏」证明——这与 MySQL 下的第二重保险完全一致。 +""" + +import importlib.util +import os +import sqlite3 +import sys +import threading +import time +import types +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +PKG_DIR = os.path.join(os.path.dirname(HERE), "world_sync") +_PKG = "_ws_m11b2" + + +def _load(name): + """以合成包方式加载被测模块（绕开 world_sync/__init__.py 的运行时依赖）。""" + if _PKG not in sys.modules: + pkg = types.ModuleType(_PKG) + pkg.__path__ = [PKG_DIR] + sys.modules[_PKG] = pkg + 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") +sqlmod = _load("pbl_runtime_sql") +errs = _load("pbl_runtime_errors") + +RuntimeStateEventWriter = tx.RuntimeStateEventWriter +write_event_with_state = tx.write_event_with_state +percentile = tx.percentile +build_plan = tx.build_plan +SQLITE = sqlmod.SQLITE +MYSQL = sqlmod.MYSQL +RuntimeWriteError = errs.RuntimeWriteError +InvalidWriteRequest = errs.InvalidWriteRequest +EventWriteError = errs.EventWriteError +StateWriteError = errs.StateWriteError +ConcurrentStateConflict = errs.ConcurrentStateConflict +TransactionAborted = errs.TransactionAborted + +DDL_EVENT = """ +CREATE TABLE IF NOT EXISTS pbl_runtime_event ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + tenant_id TEXT NOT NULL, + event_uid TEXT NOT NULL, + event_code TEXT, + idem_key TEXT, + world_id TEXT, + scene_id TEXT, + session_id TEXT, + entity_id TEXT NOT NULL, + actor_id TEXT, + event_type TEXT NOT NULL, + payload TEXT, + payload_json TEXT, + causation_id TEXT, + seq INTEGER, + seq_no INTEGER, + source TEXT, + state TEXT, + state_version INTEGER, + tx_group TEXT, + broadcast TEXT, + created_by TEXT, + created_at TEXT NOT NULL, + updated_at TEXT, + occurred_at TEXT, + UNIQUE (tenant_id, event_uid) +) +""" + +DDL_STATE = """ +CREATE TABLE IF NOT EXISTS pbl_entity_state ( + id TEXT, + tenant_id TEXT NOT NULL, + session_id TEXT NOT NULL, + entity_id TEXT NOT NULL, + state_json TEXT, + state_version INTEGER NOT NULL DEFAULT 0, + checksum TEXT, + updated_by TEXT, + created_at TEXT, + updated_at TEXT, + PRIMARY KEY (tenant_id, session_id, entity_id) +) +""" + + +def make_db(path=":memory:"): + conn = sqlite3.connect(path, timeout=15.0) + conn.isolation_level = None # 由写入器显式 BEGIN/COMMIT/ROLLBACK + conn.execute(DDL_EVENT) + conn.execute(DDL_STATE) + return conn + + +def seed_state(conn, tenant="t1", session="s1", entity="e1", state_json='{"hp":1}', + version=0, checksum=None, updated_by=None): + conn.execute("DELETE FROM pbl_entity_state WHERE tenant_id=? AND session_id=? AND entity_id=?", + (tenant, session, entity)) + conn.execute( + "INSERT INTO pbl_entity_state (id,tenant_id,session_id,entity_id,state_json," + "state_version,checksum,updated_by,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?)", + ("row-" + entity, tenant, session, entity, state_json, version, checksum, + updated_by, "2026-01-01 00:00:00", "2026-01-01 00:00:00")) + + +def event(uid="ev-1", tenant="t1", session="s1", entity="e1", **kw): + data = {"tenant_id": tenant, "event_uid": uid, "session_id": session, + "entity_id": entity, "event_type": "unit.move", "world_id": "w1", + "scene_id": "sc1", "actor_id": "u1", "payload": {"dx": 1}, + "seq_no": 7, "seq": 7} + data.update(kw) + return data + + +def state_update(**kw): + data = {"session_id": "s1", "entity_id": "e1", "state_json": {"hp": 2}, + "updated_by": "u1"} + data.update(kw) + return data + + +class SingleConnProxy(object): + """包装连接，可对指定阶段的 SQL 注入失败（验证原子回滚）。""" + + def __init__(self, conn, fail_when): + self._conn = conn + self._fail_when = fail_when + + def cursor(self): + return _ProxyCursor(self._conn.cursor(), self._fail_when) + + def commit(self): + self._conn.commit() + + def rollback(self): + self._conn.rollback() + + def close(self): + pass + + +class _ProxyCursor(object): + def __init__(self, cur, fail_when): + self._cur = cur + self._fail_when = fail_when + + @property + def rowcount(self): + return self._cur.rowcount + + def execute(self, sql, params=None): + if self._fail_when(sql): + raise sqlite3.IntegrityError("injected failure: %s" % " ".join(sql.split())[:60]) + return self._cur.execute(sql, params or []) + + def fetchone(self): + return self._cur.fetchone() + + def close(self): + self._cur.close() + + +def is_event_insert(sql): + head = " ".join(sql.split()).upper() + return head.startswith("INSERT INTO PBL_RUNTIME_EVENT") + + +# -------------------------------------------------------------------------- +# 1. 正常写入：单事务内事件 + 状态同时落库 +# -------------------------------------------------------------------------- + +class TestNormalWrite(unittest.TestCase): + + def setUp(self): + self.conn = make_db() + seed_state(self.conn, version=3, state_json='{"hp":10}') + self.writer = RuntimeStateEventWriter(lambda: self.conn, dialect=SQLITE, close_connection=False) + + def tearDown(self): + self.conn.close() + + def test_writes_both_rows_in_one_transaction(self): + res = self.writer.write_event_with_state(event("ev-ok"), state_update()) + self.assertEqual("ev-ok", res["event_uid"]) + self.assertEqual(4, res["state_version"]) # 3 -> 4 + self.assertEqual(3, res["expected_version"]) + self.assertFalse(res["state_inserted"]) + # 事务内语句清单：幂等预检 + 状态锁读 + 状态更新 + 事件插入（无广播/轮询语句） + self.assertEqual(4, res["statement_count"]) + joined = " ; ".join(res["statements"]).upper() + self.assertNotIn("NOTIFY", joined) + self.assertNotIn("SLEEP", joined) + rows = self.conn.execute("SELECT count(*) FROM pbl_runtime_event").fetchone()[0] + self.assertEqual(1, rows) + st = self.conn.execute( + "SELECT state_json, state_version, updated_by FROM pbl_entity_state").fetchone() + self.assertEqual('{"hp":2}', st[0]) + self.assertEqual(4, st[1]) + self.assertEqual("u1", st[2]) + + def test_event_row_is_append_only_and_carries_aligned_version(self): + self.writer.write_event_with_state(event("ev-a"), state_update()) + row = self.conn.execute( + "SELECT event_uid, state_version, entity_id, session_id, seq_no, source, " + "created_at FROM pbl_runtime_event").fetchone() + self.assertEqual("ev-a", row[0]) + self.assertEqual(4, row[1]) # 状态 3 -> 4，事件版本对齐 + self.assertEqual("e1", row[2]) + self.assertEqual("s1", row[3]) + self.assertEqual(7, row[4]) + self.assertEqual("runtime", row[5]) + self.assertTrue(row[6]) # 分区键非空 + + def test_state_row_inserted_when_absent(self): + self.conn.execute("DELETE FROM pbl_entity_state") + res = self.writer.write_event_with_state(event("ev-new"), state_update()) + self.assertTrue(res["state_inserted"]) + self.assertEqual(1, res["state_version"]) + self.assertEqual(1, self.conn.execute( + "SELECT count(*) FROM pbl_entity_state").fetchone()[0]) + + def test_module_level_helper_uses_configured_writer(self): + tx.configure_writer(lambda: self.conn, dialect=SQLITE, close_connection=False) + res = write_event_with_state(event("ev-helper"), state_update()) + self.assertEqual(4, res["state_version"]) + self.assertEqual(1, self.conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + + def test_payload_dict_serialised_to_json_text(self): + self.writer.write_event_with_state(event("ev-p", payload={"a": 1, "b": [1, 2]}), + state_update()) + payload = self.conn.execute("SELECT payload FROM pbl_runtime_event").fetchone()[0] + self.assertIn('"a":1', payload) + + +# -------------------------------------------------------------------------- +# 2. 原子性：任一步失败 → 全部回滚 + 明确异常 +# -------------------------------------------------------------------------- + +class TestAtomicRollback(unittest.TestCase): + + def setUp(self): + self.conn = make_db() + seed_state(self.conn, version=5, state_json='{"hp":99}') + self.snapshot = self.conn.execute( + "SELECT state_json, state_version FROM pbl_entity_state").fetchone() + + def tearDown(self): + self.conn.close() + + def _writer(self, **kw): + return RuntimeStateEventWriter(lambda: self.conn, dialect=SQLITE, close_connection=False, **kw) + + def test_event_insert_failure_rolls_back_state_update(self): + """事件插入失败（中途失败）⇒ 状态更新必须一并回滚。""" + proxy = SingleConnProxy(self.conn, fail_when=is_event_insert) + writer = RuntimeStateEventWriter(lambda: proxy, dialect=SQLITE, close_connection=False) + with self.assertRaises(RuntimeWriteError) as cm: + writer.write_event_with_state(event("ev-fail"), state_update()) + self.assertIsInstance(cm.exception, EventWriteError) + self.assertEqual("ev-fail", cm.exception.detail.get("event_uid")) + # 状态未被改写（回滚生效） + now = self.conn.execute("SELECT state_json, state_version FROM pbl_entity_state").fetchone() + self.assertEqual(self.snapshot, now) + self.assertEqual(0, self.conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + + def test_before_commit_hook_failure_rolls_back_everything(self): + def boom(plan): + raise RuntimeError("boom before commit") + writer = self._writer(before_commit=boom) + with self.assertRaises(TransactionAborted): + writer.write_event_with_state(event("ev-hook"), state_update()) + now = self.conn.execute("SELECT state_json, state_version FROM pbl_entity_state").fetchone() + self.assertEqual(self.snapshot, now) + self.assertEqual(0, self.conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + + def test_state_update_failure_rolls_back_and_is_explicit(self): + def fail_state(sql): + head = " ".join(sql.split()).upper() + return head.startswith("UPDATE PBL_ENTITY_STATE") + proxy = SingleConnProxy(self.conn, fail_when=fail_state) + writer = RuntimeStateEventWriter(lambda: proxy, dialect=SQLITE, close_connection=False) + with self.assertRaises(RuntimeWriteError) as cm: + writer.write_event_with_state(event("ev-sf"), state_update()) + self.assertIsInstance(cm.exception, StateWriteError) + self.assertEqual(0, self.conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + self.assertEqual(self.snapshot, self.conn.execute( + "SELECT state_json, state_version FROM pbl_entity_state").fetchone()) + + def test_errors_are_never_swallowed(self): + """任何失败都必须抛 RuntimeWriteError 子类，且 detail 可定位。""" + writer = self._writer(before_commit=lambda p: (_ for _ in ()).throw(ValueError("x"))) + try: + writer.write_event_with_state(event("ev-e"), state_update()) + except RuntimeWriteError as exc: + self.assertIn("reason", exc.detail) + self.assertIn("event_uid", exc.detail) + self.assertTrue(str(exc)) + else: + self.fail("未抛出明确异常，属吞错误") + + def test_duplicate_event_is_rejected_and_state_untouched(self): + writer = self._writer() + writer.write_event_with_state(event("ev-dup"), state_update()) + with self.assertRaises(EventWriteError) as cm: + writer.write_event_with_state(event("ev-dup"), state_update()) + self.assertEqual("DUPLICATE_EVENT", cm.exception.code) + self.assertEqual(1, self.conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + self.assertEqual(6, self.conn.execute( + "SELECT state_version FROM pbl_entity_state").fetchone()[0]) + + +# -------------------------------------------------------------------------- +# 3. 并发：不出现脏写 / 丢失更新 +# -------------------------------------------------------------------------- + +class TestConcurrency(unittest.TestCase): + + def setUp(self): + self.path = os.path.join(HERE, "_m11b2_concurrent.sqlite") + if os.path.exists(self.path): + os.remove(self.path) + self.conn = make_db(self.path) + seed_state(self.conn, version=0, state_json='{"hp":0}') + self.conn.close() + + def tearDown(self): + if os.path.exists(self.path): + os.remove(self.path) + + def _factory(self): + def f(): + return make_db(self.path) + return f + + def test_stale_expected_version_raises_conflict(self): + """乐观锁护栏：过期版本写入必须失败且不留痕（防丢失更新）。""" + conn = make_db(self.path) + seed_state(conn, version=8, state_json='{"hp":8}') + writer = RuntimeStateEventWriter(lambda: conn, dialect=SQLITE, close_connection=False) + with self.assertRaises(ConcurrentStateConflict) as cm: + writer.write_event_with_state(event("ev-stale"), + state_update(expected_version=7)) + self.assertEqual(7, cm.exception.detail.get("expected_version")) # 调用方声明的过期基线 + self.assertEqual({"tenant_id": "t1", "session_id": "s1", "entity_id": "e1"}, + cm.exception.detail.get("key")) + self.assertEqual(8, conn.execute( + "SELECT state_version FROM pbl_entity_state").fetchone()[0]) + self.assertEqual(0, conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + conn.close() + + def test_two_concurrent_writers_same_entity_no_lost_update(self): + """两个并发事务写同一实体：全部成功、版本连续递增、事件数与成功数一致。""" + n_each = 6 + errors = [] + results = [] + lock = threading.Lock() + + def worker(tag): + conn = make_db(self.path) + writer = RuntimeStateEventWriter(lambda: conn, dialect=SQLITE, close_connection=False) + for i in range(n_each): + try: + res = writer.write_event_with_state( + event("ev-%s-%d" % (tag, i)), state_update()) + with lock: + results.append(res) + except RuntimeWriteError as exc: + with lock: + errors.append(exc) + conn.close() + + threads = [threading.Thread(target=worker, args=(t,)) for t in ("A", "B")] + for t in threads: + t.start() + for t in threads: + t.join() + + conn = sqlite3.connect(self.path) + conn.isolation_level = None + ok = len(results) + self.assertEqual(0, len(errors), "并发写入不应有未恢复错误: %s" % errors) + self.assertEqual(2 * n_each, ok) + self.assertEqual(2 * n_each, conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + final = conn.execute("SELECT state_version FROM pbl_entity_state").fetchone()[0] + self.assertEqual(2 * n_each, final) # 无丢失更新 + versions = sorted(r["state_version"] for r in results) + self.assertEqual(list(range(1, 2 * n_each + 1)), versions) # 版本连续且互不重复 ⇒ 无脏写 + conn.close() + + def test_concurrent_duplicate_event_only_one_wins(self): + """同一 event_uid 并发提交：只落一条事件，另一次被幂等拒绝。""" + conn = make_db(self.path) + outcomes = [] + lock = threading.Lock() + + def worker(): + c = make_db(self.path) + writer = RuntimeStateEventWriter(lambda: c, dialect=SQLITE) + try: + writer.write_event_with_state(event("ev-race"), state_update()) + with lock: + outcomes.append("ok") + except RuntimeWriteError as exc: + with lock: + outcomes.append(type(exc).__name__) + c.close() + + threads = [threading.Thread(target=worker) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + self.assertEqual(1, conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + self.assertEqual(1, outcomes.count("ok")) + conn.close() + + +# -------------------------------------------------------------------------- +# 4. 入参校验（fail-closed） +# -------------------------------------------------------------------------- + +class TestValidation(unittest.TestCase): + + def setUp(self): + self.conn = make_db() + self.writer = RuntimeStateEventWriter(lambda: self.conn, dialect=SQLITE, close_connection=False) + + def tearDown(self): + self.conn.close() + + def test_missing_required_event_fields(self): + with self.assertRaises(InvalidWriteRequest) as cm: + self.writer.write_event_with_state({"tenant_id": "t1"}, state_update()) + self.assertIn("event_uid", cm.exception.detail["missing"]) + self.assertEqual(0, self.conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + + def test_missing_state_json_rejected(self): + with self.assertRaises(InvalidWriteRequest): + self.writer.write_event_with_state(event("ev-x"), + {"session_id": "s1", "entity_id": "e1"}) + + def test_key_mismatch_between_event_and_state_rejected(self): + with self.assertRaises(InvalidWriteRequest) as cm: + self.writer.write_event_with_state(event("ev-y"), state_update(entity_id="other")) + self.assertEqual("entity_id", cm.exception.detail["field"]) + + def test_unserialisable_payload_rejected(self): + with self.assertRaises(InvalidWriteRequest): + self.writer.write_event_with_state(event("ev-z", payload=object()), state_update()) + + +# -------------------------------------------------------------------------- +# 5. 时延：单次事务（不含广播）P99 ≤ 200ms +# -------------------------------------------------------------------------- + +class TestLatency(unittest.TestCase): + + def setUp(self): + self.path = os.path.join(HERE, "_m11b2_latency.sqlite") + if os.path.exists(self.path): + os.remove(self.path) + conn = make_db(self.path) + seed_state(conn, version=0) + conn.close() + + def tearDown(self): + if os.path.exists(self.path): + os.remove(self.path) + + def test_p99_under_200ms(self): + samples = [] + conn = make_db(self.path) + writer = RuntimeStateEventWriter(lambda: conn, dialect=SQLITE, close_connection=False) + for i in range(100): + res = writer.write_event_with_state(event("ev-l-%d" % i), state_update()) + samples.append(res["elapsed_ms"]) + conn.close() + p50 = percentile(samples, 50) + p99 = percentile(samples, 99) + print("[M11b-2 latency] n=%d p50=%.3fms p99=%.3fms max=%.3fms" + % (len(samples), p50, p99, max(samples))) + self.assertLessEqual(p99, 200.0) + + def test_percentile_helper(self): + self.assertEqual(None, percentile([], 99)) + self.assertEqual(5, percentile([5], 99)) + p99 = percentile(list(range(100)), 99) + self.assertGreaterEqual(p99, 98.0) + self.assertLessEqual(p99, 99.0) + + +# -------------------------------------------------------------------------- +# 6. 范围约束：不引入广播 / 轮询；SQL 构造契约 +# -------------------------------------------------------------------------- + +class TestScopeAndSqlContract(unittest.TestCase): + + def test_no_broadcast_or_polling_in_production_sources(self): + """范围约束：本模块不得出现广播/轮询/睡眠的可执行调用（AST 级检查，不看散文）。""" + import ast + banned_calls = {"sleep", "poll", "long_poll", "publish", "subscribe", + "broadcast", "notify", "websocket", "push", "fetch_events"} + banned_imports = {"asyncio", "socket", "threading", "queue"} + for name in ("pbl_runtime_tx.py", "pbl_runtime_sql.py", "pbl_runtime_errors.py"): + path = os.path.join(PKG_DIR, name) + with open(path, encoding="utf-8") as fh: + tree = ast.parse(fh.read(), filename=path) + for node in ast.walk(tree): + if isinstance(node, (ast.Import, ast.ImportFrom)): + mods = ([a.name for a in node.names] + if isinstance(node, ast.Import) else [node.module or ""]) + for mod in mods: + root = (mod or "").split(".")[0] + self.assertNotIn(root, banned_imports, + "%s 不应导入 %s" % (name, root)) + if isinstance(node, ast.Call): + fn = node.func + ident = fn.id if isinstance(fn, ast.Name) else ( + fn.attr if isinstance(fn, ast.Attribute) else "") + self.assertNotIn(ident, banned_calls, + "%s 不应调用广播/轮询/睡眠: %s" % (name, ident)) + + def test_event_sql_is_insert_only(self): + cols = list(sqlmod.EventSchema.columns) + stmt = sqlmod.build_insert_event(MYSQL, ["tenant_id", "event_uid", "created_at"]) + self.assertTrue(stmt.startswith("INSERT INTO pbl_runtime_event")) + self.assertNotIn("UPDATE", stmt.upper()) + self.assertIn("%s", stmt) + self.assertEqual(len(cols), len(set(cols))) + self.assertIn("created_at", cols) # 分区键在契约内 + + def test_mysql_select_for_update_present(self): + stmt = sqlmod.build_select_state_for_update(MYSQL) + self.assertTrue(stmt.endswith("FOR UPDATE")) + self.assertIn("pbl_entity_state", stmt) + self.assertNotIn("?", stmt) + + def test_sqlite_lock_clause_degrades_gracefully(self): + stmt = sqlmod.build_select_state_for_update(SQLITE) + self.assertNotIn("FOR UPDATE", stmt) + + def test_update_carries_version_guard_and_increment(self): + stmt, set_cols, count = sqlmod.build_update_state( + MYSQL, ["state_json", "updated_at"]) + self.assertIn("state_version = state_version + 1", stmt) + self.assertIn("AND state_version = %s", stmt) + self.assertEqual(("state_json", "updated_at"), set_cols) + self.assertEqual(2 + 3 + 1, count) + + def test_plan_assembly_is_side_effect_free(self): + conn = make_db() + plan = build_plan(event("ev-plan"), state_update(expected_version=11)) + self.assertEqual(11, plan.expected_version) + self.assertEqual("ev-plan", plan.event_uid) + self.assertEqual({"tenant_id": "t1", "session_id": "s1", "entity_id": "e1"}, plan.key) + self.assertEqual("created_at", sqlmod.EventSchema.partition_key) + self.assertEqual(0, conn.execute( + "SELECT count(*) FROM pbl_runtime_event").fetchone()[0]) + conn.close() + + def test_transaction_lifecycle_is_balanced(self): + """BEGIN / COMMIT 各一次；失败路径必须 ROLLBACK。""" + conn = make_db() + seed_state(conn, version=0) + calls = [] + + class Spy(sqlite3.Connection): + pass + + class Traced(object): + def __init__(self, real): + self._real = real + + def cursor(self): + cur = self._real.cursor() + outer = self + + class C(object): + @property + def rowcount(self): + return cur.rowcount + + def execute(self, sql, params=None): + head = " ".join(sql.split()).upper() + if head.startswith("BEGIN"): + calls.append("BEGIN") + elif head.startswith("COMMIT"): + calls.append("COMMIT") + elif head.startswith("ROLLBACK"): + calls.append("ROLLBACK") + if "INSERT INTO PBL_RUNTIME_EVENT" in head: + raise sqlite3.OperationalError("injected") + return cur.execute(sql, params or []) + + def fetchone(self): + return cur.fetchone() + + def close(self): + cur.close() + return C() + + def commit(self): + self._real.commit() + + def rollback(self): + calls.append("ROLLBACK") + self._real.rollback() + + def close(self): + pass + + traced = Traced(conn) + writer = RuntimeStateEventWriter(lambda: traced, dialect=SQLITE) + with self.assertRaises(EventWriteError): + writer.write_event_with_state(event("ev-trace"), state_update()) + self.assertIn("BEGIN", calls) + self.assertIn("ROLLBACK", calls) + self.assertNotIn("COMMIT", calls) + conn.close() + + +if __name__ == "__main__": + unittest.main(verbosity=2) diff --git a/tests/test_m11b2_wiring.py b/tests/test_m11b2_wiring.py new file mode 100644 index 0000000..0dc12b0 --- /dev/null +++ b/tests/test_m11b2_wiring.py @@ -0,0 +1,563 @@ +# -*- coding: utf-8 -*- +"""M11b-2 单事务写入 —— 可执行单元测试（sqlite 文件库真实跑通，非 import 面检查）。 + +覆盖需求点与验收标准： +- 正常写入：事件 + 状态在同一事务落库，两表各 1 行（TestAtomicWriteSuccess）； +- 事件插入失败 → 状态更新一并回滚（TestRollbackOnEventFailure）； +- 状态更新失败/并发冲突 → 整体回滚，事件不留（TestRollbackOnStateFailure）； +- 并发：多事务写同一实体，不出现脏写/丢失更新（TestConcurrencyNoLostUpdate）； +- 租户/世界隔离：跨租户同名 entity_id 互不影响（TestTenantWorldIsolation）； +- P99 ≤ 200ms（TestLatencyP99，percentile 与 m11b-selftest-run.log 同口径）； +- conn_factory 三级优先级 + fail-closed + 禁读 env/*.json（TestConnFactoryResolution）； +- 接线面：导出一致性、禁项（无广播/轮询）扫描（TestPackageExports / TestNoForbiddenLogic）。 + +运行：cd apps/scense/pkgs/world_sync && python -m pytest tests/test_m11b2_wiring.py -v +""" + +import os +import sqlite3 +import sys +import threading + +import pytest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from world_sync import pbl_runtime_sql as ps # noqa: E402 +from world_sync.pbl_runtime_errors import ( # noqa: E402 + ConcurrentStateConflict, + ConnFactoryNotConfigured, + EventWriteError, + PblRuntimeError, + StateWriteError, + TxConfigError, +) +from world_sync.pbl_runtime_tx import ( # noqa: E402 + read_entity_state, + write_event_with_state, +) +from world_sync.pbl_runtime_tx_env import ( # noqa: E402 + build_sqlite_conn_factory, + resolve_conn_factory, +) + +percentile = ps.percentile + + +# -------------------------------------------------------------------------- +# fixtures / helpers +# -------------------------------------------------------------------------- + +@pytest.fixture +def db_path(tmp_path): + """文件型 sqlite 库（多线程共享同一库，比 :memory: 更适合并发用例）。""" + return str(tmp_path / "m11b2.db") + + +@pytest.fixture +def factory(db_path): + """sqlite 连接工厂，并保证两表 DDL 已建（CREATE TABLE IF NOT EXISTS 幂等）。""" + f = build_sqlite_conn_factory(db_path) + c = f() + for stmt in ps.ddl_for(ps.DIALECT_SQLITE): + c.execute(stmt) + c.commit() + c.close() + return f + + +@pytest.fixture +def conn(factory): + c = factory() + yield c + c.close() + + +def _event(tenant="t1", world="w1", entity="e1", **kw): + base = { + "tenant_id": tenant, + "world_id": world, + "entity_id": entity, + "event_type": "state.updated", + "payload": {"hp": 100}, + "session_id": "s-1", + "source": "unit-test", + } + base.update(kw) + return base + + +def _state(tenant="t1", world="w1", entity="e1", **kw): + base = {"tenant_id": tenant, "world_id": world, "entity_id": entity, + "state": {"hp": 100, "x": 1}} + base.update(kw) + return base + + +def _counts(conn): + ev = conn.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0] + st = conn.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0] + return ev, st + + +# -------------------------------------------------------------------------- +# 1. 正常写入：单事务内事件 + 状态各 1 行 +# -------------------------------------------------------------------------- + +class TestAtomicWriteSuccess(object): + + def test_first_write_inserts_state_and_event(self, conn, factory): + result = write_event_with_state(_event(), _state(state={"hp": 90}), + conn_factory=factory) + assert result["action"] == "inserted" + assert result["state_version"] == 1 + assert result["dialect"] == "sqlite" + assert _counts(conn) == (1, 1) + + row = conn.execute( + "SELECT event_id, entity_id, state_version, payload, created_at " + "FROM pbl_runtime_event" + ).fetchone() + assert row[1] == "e1" + assert row[2] == 1 + assert '"hp":100' in row[3] # payload 存事件体 + assert row[4].endswith("Z") + st = conn.execute( + "SELECT state, state_version, updated_by_event FROM pbl_entity_state" + ).fetchone() + assert '"hp":90' in st[0] # 状态存新值 + assert st[1] == 1 + assert st[2] == row[0] # 状态行回指生效事件 + + def test_second_write_bumps_version(self, conn, factory): + write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) + r2 = write_event_with_state(_event(), _state(state={"hp": 70}), conn_factory=factory) + assert r2["action"] == "updated" + assert r2["state_version"] == 2 + assert _counts(conn) == (2, 1) + + def test_event_only_write_keeps_state(self, conn, factory): + write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) + r = write_event_with_state(_event(event_type="chat"), None, conn_factory=factory) + assert r["action"] == "event_only" + assert _counts(conn) == (2, 1) + + def test_read_entity_state_helper(self, factory): + write_event_with_state(_event(), _state(state={"hp": 42}), conn_factory=factory) + got = read_entity_state(conn_factory=factory, tenant_id="t1", + world_id="w1", entity_id="e1") + assert got["state_version"] == 1 + assert '"hp":42' in got["state"] + assert read_entity_state(conn_factory=factory, tenant_id="t1", + world_id="w1", entity_id="nope") is None + + def test_missing_tenant_rejected_before_touching_db(self, factory): + with pytest.raises(TxConfigError) as ei: + write_event_with_state({"world_id": "w1", "entity_id": "e1"}, + _state(), conn_factory=factory) + assert ei.value.code == "PBL_TX_CONFIG_ERROR" + assert ei.value.detail.get("field") == "tenant_id" + assert _counts(factory()) == (0, 0) + + def test_conflicting_scope_rejected(self, factory): + with pytest.raises(TxConfigError): + write_event_with_state(_event(tenant="t1"), + _state(tenant="t2"), conn_factory=factory) + + def test_result_is_serializable(self, factory): + import json + r = write_event_with_state(_event(), _state(), conn_factory=factory) + json.dumps(r) # 不抛即通过 + + +# -------------------------------------------------------------------------- +# 2. 失败整体回滚：事件插入失败 → 状态变更消失 +# -------------------------------------------------------------------------- + +class TestRollbackOnEventFailure(object): + + def test_event_failure_rolls_back_state_update(self, conn, factory): + """状态先更新成功，事件插入被触发器打坏 → 断言两表零增量、状态仍是旧值。""" + write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) + conn.execute( + "CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event " + "BEGIN SELECT RAISE(ABORT, 'boom: event insert failed'); END" + ) + + with pytest.raises(EventWriteError) as ei: + write_event_with_state(_event(), _state(state={"hp": 1}), conn_factory=factory) + assert ei.value.code == "PBL_RUNTIME_EVENT_WRITE_FAILED" + assert "boom" in str(ei.value) + + assert _counts(conn) == (1, 1) + st = conn.execute("SELECT state, state_version FROM pbl_entity_state").fetchone() + assert '"hp":90' in st[0] + assert st[1] == 1 + + def test_duplicate_event_id_rolls_back_state(self, conn, factory): + """同 event_id 主键冲突 → 事件失败 → 状态更新回滚。""" + write_event_with_state(_event(), _state(state={"hp": 90}), + conn_factory=factory, event_id="dup-1") + with pytest.raises(EventWriteError): + write_event_with_state(_event(), _state(state={"hp": 5}), + conn_factory=factory, event_id="dup-1") + assert _counts(conn) == (1, 1) + st = conn.execute("SELECT state_version FROM pbl_entity_state").fetchone() + assert st[0] == 1 + + +# -------------------------------------------------------------------------- +# 3. 失败整体回滚：状态更新失败 / 并发冲突 +# -------------------------------------------------------------------------- + +class TestRollbackOnStateFailure(object): + + def test_expected_version_conflict_rolls_back_both(self, conn, factory): + write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) + + with pytest.raises(ConcurrentStateConflict) as ei: + write_event_with_state( + _event(), _state(state={"hp": 0}, expected_version=99), + conn_factory=factory) + assert ei.value.code == "PBL_ENTITY_STATE_CONFLICT" + assert ei.value.detail["expected_version"] == 99 + assert ei.value.detail["actual_version"] == 1 + assert _counts(conn) == (1, 1) # 冲突时事件一条都不能多留 + + def test_state_trigger_failure_rolls_back_event(self, conn, factory): + """状态 UPDATE 失败（触发器报错）→ 本事务的事件也不能落库。""" + write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) + conn.execute( + "CREATE TRIGGER fail_state BEFORE UPDATE ON pbl_entity_state " + "BEGIN SELECT RAISE(ABORT, 'boom: state update failed'); END" + ) + with pytest.raises(StateWriteError): + write_event_with_state(_event(), _state(state={"hp": 3}), conn_factory=factory) + # 事件仍只有首条（失败事务的事件被回滚），状态版本仍为 1 + assert _counts(conn) == (1, 1) + assert conn.execute( + "SELECT state_version FROM pbl_entity_state").fetchone()[0] == 1 + + def test_failed_write_leaves_no_open_transaction(self, factory): + """失败后连接必须已回滚，后续写入不受影响（不吞错误、不留半截事务）。""" + c = factory() + c.execute( + "CREATE TRIGGER fail_event BEFORE INSERT ON pbl_runtime_event " + "BEGIN SELECT RAISE(ABORT, 'boom'); END" + ) + with pytest.raises(EventWriteError): + write_event_with_state(_event(), _state(), conn=c) + c.execute("DROP TRIGGER fail_event") + r = write_event_with_state(_event(), _state(), conn=c) + assert r["state_version"] == 1 + assert c.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0] == 1 + assert c.execute("SELECT COUNT(*) FROM pbl_entity_state").fetchone()[0] == 1 + c.close() + + +# -------------------------------------------------------------------------- +# 4. 并发：不出现脏写 / 丢失更新 +# -------------------------------------------------------------------------- + +class TestConcurrencyNoLostUpdate(object): + + def test_concurrent_writers_no_lost_update(self, factory): + """8 线程同时写同一实体：成功版本号连续无空洞，事件数 == 成功数。""" + threads = 8 + results = [] + errors = [] + lock = threading.Lock() + barrier = threading.Barrier(threads) + + def worker(i): + barrier.wait() + try: + r = write_event_with_state( + _event(payload={"writer": i}), + _state(state={"hp": 100 + i}), + conn_factory=factory, + ) + with lock: + results.append(r["state_version"]) + except PblRuntimeError as exc: + with lock: + errors.append(exc.code) + except sqlite3.OperationalError as exc: + # sqlite 文件库锁超时属环境限制，不算脏写 + with lock: + errors.append("locked:%s" % str(exc)[:40]) + + ts = [threading.Thread(target=worker, args=(i,)) for i in range(threads)] + for t in ts: + t.start() + for t in ts: + t.join() + + c = factory() + try: + n_events = c.execute("SELECT COUNT(*) FROM pbl_runtime_event").fetchone()[0] + st = c.execute("SELECT state_version FROM pbl_entity_state").fetchone() + finally: + c.close() + + assert len(results) >= 1, "all writers failed: %s" % errors + assert n_events == len(results) # 没有「事件写了但状态没推进」的半截事务 + assert st[0] == len(results) # 版本 == 成功提交数，无丢失更新 + assert sorted(results) == list(range(1, len(results) + 1)) # 连续无空洞 + assert not [e for e in errors if not e.startswith("locked")] + + def test_optimistic_lock_rejects_stale_writer(self, conn, factory): + """显式 expected_version 的写者：版本被别人推进后必须被拒绝（不覆盖）。""" + write_event_with_state(_event(), _state(state={"hp": 90}), conn_factory=factory) + write_event_with_state(_event(), _state(state={"hp": 80}), conn_factory=factory) + with pytest.raises(ConcurrentStateConflict): + write_event_with_state(_event(), _state(state={"hp": 70}, expected_version=1), + conn_factory=factory) + assert _counts(conn) == (2, 1) + st = conn.execute("SELECT state FROM pbl_entity_state").fetchone() + assert '"hp":80' in st[0] # 迟到的旧版本写者没有覆盖新状态 + + +# -------------------------------------------------------------------------- +# 5. 租户 / 世界隔离 +# -------------------------------------------------------------------------- + +class TestTenantWorldIsolation(object): + + def test_same_entity_id_across_tenants_isolated(self, conn, factory): + write_event_with_state(_event(tenant="t1"), _state(tenant="t1", state={"hp": 1}), + conn_factory=factory) + write_event_with_state(_event(tenant="t2"), _state(tenant="t2", state={"hp": 2}), + conn_factory=factory) + assert _counts(conn) == (2, 2) + r = conn.execute( + "SELECT tenant_id, state_version FROM pbl_entity_state ORDER BY tenant_id" + ).fetchall() + assert r == [("t1", 1), ("t2", 1)] # 各自独立版本，互不覆盖 + + def test_same_entity_id_across_worlds_isolated(self, conn, factory): + write_event_with_state(_event(world="w1"), _state(world="w1"), conn_factory=factory) + write_event_with_state(_event(world="w2"), _state(world="w2"), conn_factory=factory) + assert _counts(conn) == (2, 2) + + def test_event_query_scoped_by_tenant(self, conn, factory): + write_event_with_state(_event(tenant="t1"), _state(tenant="t1"), conn_factory=factory) + write_event_with_state(_event(tenant="t2"), _state(tenant="t2"), conn_factory=factory) + n = conn.execute( + "SELECT COUNT(*) FROM pbl_runtime_event WHERE tenant_id = ? AND world_id = ?", + ("t1", "w1")).fetchone()[0] + assert n == 1 + + +# -------------------------------------------------------------------------- +# 6. 时延：P99 ≤ 200ms（不含广播） +# -------------------------------------------------------------------------- + +class TestLatencyP99(object): + + def test_percentile_linear_interpolation(self): + assert percentile([], 99) == 0.0 + assert percentile([5], 99) == 5.0 + assert percentile([1, 2, 3, 4], 50) == 2.5 + assert abs(percentile(list(range(1, 101)), 99) - 99.01) < 1e-9 + assert percentile([10, 20, 30], 0) == 10 + assert percentile([10, 20, 30], 100) == 30 + + def test_single_tx_p99_under_200ms(self, factory): + """100 次单事务写入（事件+状态），P99 ≤ 200ms（sqlite 本地口径）。""" + samples = [] + for i in range(100): + r = write_event_with_state( + _event(entity="perf-%d" % (i % 10), payload={"i": i}), + _state(entity="perf-%d" % (i % 10), state={"i": i}), + conn_factory=factory, + ) + samples.append(r["elapsed_ms"]) + p99 = percentile(samples, 99) + assert p99 <= 200.0, "P99=%.3fms exceeds 200ms budget" % p99 + assert percentile(samples, 50) <= 200.0 + + +# -------------------------------------------------------------------------- +# 7. conn_factory 三级优先级 + fail-closed + 禁读 env/*.json +# -------------------------------------------------------------------------- + +class TestConnFactoryResolution(object): + + def test_explicit_factory_wins(self, tmp_path, monkeypatch): + path = str(tmp_path / "explicit.db") + called = [] + real = build_sqlite_conn_factory(path) + + def counting(): + called.append(1) + return real() + + monkeypatch.setenv("PBL_RUNTIME_DB_URL", "sqlite:///%s" % (tmp_path / "env.db")) + f = resolve_conn_factory(counting) + c = f() + c.execute("SELECT 1").fetchone() + c.close() + assert called == [1] + + def test_env_url_used_when_no_injection(self, tmp_path, monkeypatch): + path = str(tmp_path / "from-env.db") + monkeypatch.setenv("PBL_RUNTIME_DB_URL", "sqlite:///%s" % path) + f = resolve_conn_factory(None) + c = f() + assert c is not None + c.close() + assert os.path.isfile(path) + + def test_db_url_string_injection(self, tmp_path, monkeypatch): + path = str(tmp_path / "by-string.db") + f = resolve_conn_factory("sqlite:///%s" % path) + c = f() + c.close() + assert os.path.isfile(path) + + def test_fail_closed_when_nothing_configured(self, monkeypatch): + import world_sync.pbl_runtime_tx_env as txenv + monkeypatch.delenv("PBL_RUNTIME_DB_URL", raising=False) + monkeypatch.delenv("WORLD_SYNC_DB_URL", raising=False) + # 屏蔽模块自有 conf/db.json，模拟「三级全空」 + monkeypatch.setattr(txenv, "load_db_config", lambda conf_dir=None: None) + with pytest.raises(ConnFactoryNotConfigured) as ei: + resolve_conn_factory(None) + assert ei.value.code == "PBL_TX_CONN_FACTORY_MISSING" + assert "forbidden" in str(ei.value) + assert ei.value.detail["tried"] == [ + "explicit", "server_env", "env_vars", "module_conf/db.json"] + + def test_plaintext_password_in_module_conf_rejected(self, monkeypatch): + import world_sync.pbl_runtime_tx_env as txenv + monkeypatch.setattr( + txenv, "load_db_config", + lambda conf_dir=None: {"dialect": "mysql", "host": "h", + "user": "u", "database": "d", + "password": "s3cr3t"}) + with pytest.raises(TxConfigError): + txenv._from_module_conf() + + def test_unsupported_url_scheme_rejected(self): + import world_sync.pbl_runtime_tx_env as txenv + with pytest.raises(TxConfigError): + txenv._parse_db_url("oracle://x/y") + + def test_no_env_test_json_candidate_in_source(self): + """QC #5 回归守卫：源码里不得再出现 env/test.json 之类的凭据文件候选路径。""" + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + pkg = os.path.join(root, "world_sync") + banned = ('"env/test.json"', '"env/prod.json"', "'env/test.json'", + "'env/prod.json'", "projects/pbls/env") + for name in sorted(os.listdir(pkg)): + if not name.endswith(".py"): + continue + with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh: + text = fh.read() + for token in banned: + assert token not in text, "%s still references %s" % (name, token) + + +# -------------------------------------------------------------------------- +# 8. 方言层单测 +# -------------------------------------------------------------------------- + +class TestDialectLayer(object): + + def test_detect_sqlite(self, conn): + assert ps.detect_dialect(conn) == ps.DIALECT_SQLITE + + def test_placeholders_per_dialect(self): + assert ps.placeholder(ps.DIALECT_SQLITE, 0) == "?" + assert ps.placeholder(ps.DIALECT_MYSQL, 0) == "%s" + assert ps.placeholder(ps.DIALECT_POSTGRES, 2) == "$3" + assert ps.render_placeholders(ps.DIALECT_MYSQL, 3) == "%s, %s, %s" + + def test_for_update_only_on_row_lock_dbs(self): + assert ps.for_update_clause(ps.DIALECT_SQLITE) == "" + assert ps.for_update_clause(ps.DIALECT_MYSQL).strip() == "FOR UPDATE" + assert ps.for_update_clause(ps.DIALECT_POSTGRES).strip() == "FOR UPDATE" + + def test_read_committed_stmt(self): + assert "READ COMMITTED" in ps.read_committed_stmt(ps.DIALECT_MYSQL) + assert "READ COMMITTED" in ps.read_committed_stmt(ps.DIALECT_POSTGRES) + assert ps.read_committed_stmt(ps.DIALECT_SQLITE) is None + + def test_begin_stmt(self): + assert ps.begin_stmt(ps.DIALECT_SQLITE) == "BEGIN IMMEDIATE" + assert ps.begin_stmt(ps.DIALECT_MYSQL) is None + + def test_unknown_dialect_raises(self): + class Weird(object): + pass + with pytest.raises(ValueError): + ps.detect_dialect(Weird()) + + def test_ddl_covers_both_tables(self): + text = " ".join(ps.ddl_for(ps.DIALECT_SQLITE)) + assert ps.EVENT_TABLE in text and ps.STATE_TABLE in text + assert "PRIMARY KEY (tenant_id, world_id, entity_id)" in text + assert "ENGINE=InnoDB" in " ".join(ps.ddl_for(ps.DIALECT_MYSQL)) + + def test_no_ddl_for_unknown_dialect(self): + with pytest.raises(ValueError): + ps.ddl_for("oracle") + + +# -------------------------------------------------------------------------- +# 9. 接线面 / 禁项扫描 +# -------------------------------------------------------------------------- + +class TestPackageExports(object): + + def test_all_is_importable(self): + import world_sync as ws + for name in ws.__all__: + assert hasattr(ws, name), "missing export: %s" % name + + def test_public_api_surface(self): + import world_sync as ws + assert callable(ws.write_event_with_state) + assert callable(ws.read_entity_state) + assert "ConcurrentStateConflict" in ws.__all__ + assert "EventWriteError" in ws.__all__ + + def test_sqlor_layer_delegates_to_tx_core(self): + from world_sync import pbl_runtime_tx, pbl_runtime_tx_sqlor + assert callable(pbl_runtime_tx_sqlor.write_event_with_state) + assert pbl_runtime_tx_sqlor._write_event_with_state is \ + pbl_runtime_tx.write_event_with_state + + +class TestNoForbiddenLogic(object): + """验收标准「不引入广播逻辑、不引入轮询逻辑」的机械守卫。""" + + FORBIDDEN = ("broadcast", "publish_to", "subscribe", "poll(", "polling", + "while True:", "time.sleep(", "websocket", "event_source") + + TARGETS = ("pbl_runtime_tx.py", "pbl_runtime_tx_sqlor.py", + "pbl_runtime_sql.py", "pbl_runtime_errors.py", + "pbl_runtime_tx_env.py") + + def test_no_broadcast_or_polling_in_m11b2_modules(self): + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + pkg = os.path.join(root, "world_sync") + for name in self.TARGETS: + with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh: + lines = fh.readlines() + for idx, ln in enumerate(lines, 1): + low = ln.lower() + for token in self.FORBIDDEN: + if token in low and not low.lstrip().startswith("#"): + pytest.fail("%s:%d contains forbidden %r" % (name, idx, token)) + + def test_no_network_imports(self): + root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + pkg = os.path.join(root, "world_sync") + for name in self.TARGETS: + with open(os.path.join(pkg, name), "r", encoding="utf-8") as fh: + text = fh.read() + for mod in ("import asyncio", "import socket", "import requests", + "import aiohttp"): + assert mod not in text, "%s %s" % (name, mod) diff --git a/world_sync/__init__.py b/world_sync/__init__.py index b29da27..b8df98a 100644 --- a/world_sync/__init__.py +++ b/world_sync/__init__.py @@ -1,14 +1,119 @@ -from .init import ( - load_world_sync, - list_world_syncs, - get_world_sync, - create_world_sync, - update_world_sync, - delete_world_sync, - execute_world_sync, +# -*- coding: utf-8 -*- +"""world_sync —— 世界运行时同步模块（PBL scense 侧）。 + +M11b-2 增量：单事务内「事件 + 实体状态」原子落库。 +对外只暴露写入/读取两个函数与异常族；**不含广播、不含轮询兜底、不含时延优化** +（这些明确在 M11b-2 范围之外，由后续里程碑承担）。 + +宿主接线（遵循 project-directory-spec 8.1：模块不读部署环境凭据文件）:: + + # apps/scense/app/scense.py 的 init() 里 + from world_sync import load_world_sync + load_world_sync(server_env) # 登记 pbl_runtime_conn_factory 等 + + # 业务侧写入 + from world_sync import write_event_with_state + write_event_with_state( + {"tenant_id": t, "world_id": w, "entity_id": e, + "event_type": "state.updated", "payload": {...}}, + {"state": {...}, "expected_version": 3}, + ) + +``conn_factory`` 解析优先级（见 :mod:`world_sync.pbl_runtime_tx_env`）： +显式注入 → ServerEnv 登记 → 环境变量 ``PBL_RUNTIME_DB_URL`` → 模块自有 ``conf/db.json``； +四级全空则 fail-closed 抛 :class:`ConnFactoryNotConfigured`。 +""" + +from .pbl_runtime_errors import ( + ConcurrentStateConflict, + ConnFactoryNotConfigured, + EventWriteError, + PblRuntimeError, + StateWriteError, + TxAbortError, + TxConfigError, +) +from .pbl_runtime_sql import ( + EVENT_TABLE, + STATE_TABLE, + detect_dialect, + percentile, +) +from .pbl_runtime_tx import ( + new_event_id, + read_entity_state as _tx_read_entity_state, + utc_now_text, + write_event_with_state as _tx_write_event_with_state, +) +from .pbl_runtime_tx_env import ( + build_sqlite_conn_factory, + resolve_conn_factory, +) +from .pbl_runtime_tx_sqlor import ( + read_entity_state, + write_event_with_state, ) +MODULE_NAME = "world_sync" +MODULE_VERSION = "0.2.0-m11b2" + +#: 宿主 ServerEnv 上登记的 conn_factory 键（pbl_runtime_tx_env.SERVER_ENV_KEYS 与此一致） +SERVER_ENV_CONN_FACTORY_KEYS = ("pbl_runtime_conn_factory", "world_sync_conn_factory") + __all__ = [ - "load_world_sync", "list_world_syncs", "get_world_sync", "create_world_sync", - "update_world_sync", "delete_world_sync", "execute_world_sync", + # 写入 / 读取（M11b-2 核心） + "write_event_with_state", + "read_entity_state", + # 接线 + "load_world_sync", + "MODULE_NAME", + "MODULE_VERSION", + "SERVER_ENV_CONN_FACTORY_KEYS", + # 工厂与方言 + "resolve_conn_factory", + "build_sqlite_conn_factory", + "detect_dialect", + "percentile", + "EVENT_TABLE", + "STATE_TABLE", + "new_event_id", + "utc_now_text", + # 异常族 + "PblRuntimeError", + "TxConfigError", + "ConnFactoryNotConfigured", + "EventWriteError", + "StateWriteError", + "ConcurrentStateConflict", + "TxAbortError", ] + + +def load_world_sync(server_env=None, conn_factory=None): + """挂载 world_sync：向 ServerEnv 登记本模块能力与（可选）conn_factory。 + + 幂等：重复调用只覆盖同名键，不产生副作用。返回本模块 ``__all__`` 便于宿主自检。 + + :param server_env: 宿主 ServerEnv 实例（None 时尝试取全局单例） + :param conn_factory: 宿主连接池工厂；提供则登记到 ``pbl_runtime_conn_factory``， + 使模块内部无需知道任何部署环境凭据来源（fail-closed 而非自行找配置文件）。 + """ + if server_env is None: + try: + from appPublic.serverEnv import ServerEnv + server_env = ServerEnv() + except Exception: + server_env = None + + if server_env is not None: + if conn_factory is not None: + setattr(server_env, SERVER_ENV_CONN_FACTORY_KEYS[0], conn_factory) + setattr(server_env, "world_sync_module", MODULE_NAME) + setattr(server_env, "world_sync_version", MODULE_VERSION) + setter = getattr(server_env, "set", None) + if callable(setter): + try: + setter("world_sync_version", MODULE_VERSION) + except Exception: + pass + return list(__all__) diff --git a/world_sync/init.py b/world_sync/init.py index 3cec829..fd351e9 100644 --- a/world_sync/init.py +++ b/world_sync/init.py @@ -1,9 +1,40 @@ import json +import os + from appPublic.uniqueID import getID from appPublic.timeUtils import curDateString, timestampstr from sqlor.dbpools import DBPools from ahserver.serverenv import ServerEnv +# --- M11b-2: 单事务写入（事件 + 实体状态）注册接线 --- +from .pbl_runtime_errors import ( + RuntimeWriteError, + InvalidWriteRequest, + ConnectionUnavailable, + StateWriteError, + EventWriteError, + ConcurrentStateConflict, + TransactionAborted, +) +from .pbl_runtime_sql import STATE_TABLE, MYSQL, SQLITE +from .pbl_runtime_tx import ( + RuntimeStateEventWriter, + build_plan, + configure_writer, + get_writer, + percentile, + write_event_with_state, + write_event_with_state_async, +) +from .pbl_runtime_tx_sqlor import ( + build_connection_factory, + build_writer, + configure_default_writer, + detect_driver, + load_db_conf, + resolve_dialect, +) + def _get_dbname(): return ServerEnv().get_module_dbname('world_sync') @@ -86,6 +117,122 @@ async def execute_world_sync(ns): return {'success': True, 'result': result} +# =========================================================================== +# M11b-1: pbl_runtime_event 表注册（仅 schema 元数据登记） +# 范围铁律：本段只做「表定义注册 + 只读元数据查询」， +# 不含任何事务写入逻辑、广播代码、轮询代码（事件写入由 M11a/M11b 事务扩展负责）。 +# =========================================================================== + +MODELS_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'models') + +# 本模块管理的物理表清单（world_sync 原 3 张 + M11b-1 新增 1 张 + M11b-2 状态表 1 张） +MODULE_TABLES = ['world_sync', 'world_sync_task', 'world_sync_log', + 'pbl_runtime_event', 'pbl_entity_state'] + +# M11b-1 新增表：运行时事件表（append-only + 按月 RANGE 分区） +PBL_RUNTIME_EVENT_TABLE = 'pbl_runtime_event' + +# 表 DDL/迁移脚本相对模块根的路径（部署时由 scripts/migrate_pbl_runtime_event.py 执行） +PBL_RUNTIME_EVENT_MIGRATIONS = [ + 'scripts/migrations/20260918_m11b_01_create_pbl_runtime_event.sql', + 'scripts/migrations/20260918_m11b_02_pbl_runtime_event_append_only.sql', + 'scripts/migrations/20260918_m11b_03_pbl_runtime_event_partition_maintenance.sql', +] + +_TABLE_MODEL_CACHE = {} + + +def load_table_model(table_name): + """读取 models/.json 表定义（四段式 summary/fields/indexes/codes），带缓存。""" + if table_name in _TABLE_MODEL_CACHE: + return _TABLE_MODEL_CACHE[table_name] + path = os.path.join(MODELS_DIR, table_name + '.json') + if not os.path.isfile(path): + return None + with open(path, 'r', encoding='utf-8') as fh: + model = json.load(fh) + _TABLE_MODEL_CACHE[table_name] = model + return model + + +def module_tables(params=None): + """返回本模块登记的表名列表（只读元数据）。""" + return list(MODULE_TABLES) + + +def get_table_schema(ns): + """按表名返回表定义 JSON（供前端/工具读取字段与索引，只读）。""" + name = (ns or {}).get('table') or (ns or {}).get('name') or '' + if name not in MODULE_TABLES: + return {'success': False, 'message': 'unknown table: %s' % name} + model = load_table_model(name) + if model is None: + return {'success': False, 'message': 'model file missing: %s.json' % name} + return {'success': True, 'table': name, 'model': model} + + +def pbl_runtime_event_schema(params=None): + """pbl_runtime_event 表定义（含 append-only 与分区约束声明），只读。""" + return load_table_model(PBL_RUNTIME_EVENT_TABLE) + + +async def pbl_runtime_event_partitions(params=None): + """查询 pbl_runtime_event 当前分区列表（只读 information_schema，不写任何数据）。""" + sql = ("select partition_name, partition_description " + " from information_schema.partitions " + " where table_schema=${dbname}$ and table_name=${tbl}$ " + " and partition_name is not null " + " order by partition_ordinal_position") + db = DBPools() + async with db.sqlorContext(_get_dbname()) as sor: + return await sor.sqlExe(sql, {'dbname': _get_dbname(), 'tbl': PBL_RUNTIME_EVENT_TABLE}) + + +# =========================================================================== +# M11b-2: 单事务写入（事件 + 实体状态原子落库）—— 运行时接线 +# 范围铁律：本段只做「契约函数登记 + 默认写入器配置」， +# 不含广播（属 M11b-3）、不含轮询兜底、不含时延优化。 +# 写入器采用「懒失败」：init 阶段不建连，真正写入时才取连接； +# 参数不全则抛 ConnectionUnavailable（fail-closed，不静默降级、不吞错误）。 +# =========================================================================== + +def configure_runtime_writer(conn_factory=None, dialect=None, conf=None, + driver=None, before_commit=None, + close_connection=True): + """应用侧/测试侧配置默认单事务写入器，返回写入器实例。 + + 与 :func:`pbl_runtime_tx_sqlor.configure_default_writer` 等价，只是暴露给 + ServerEnv 的名字更贴近业务语义（上层 dspy 只需 ``env.configure_runtime_writer``）。 + ``dialect`` 可传字符串（'mysql'/'sqlite'）或方言对象。 + """ + return configure_default_writer( + conn_factory=conn_factory, dialect=dialect, conf=conf, driver=driver, + before_commit=before_commit, close_connection=close_connection) + + +def get_runtime_writer(): + """取当前默认写入器；未配置时抛 ConnectionUnavailable。""" + return get_writer() + + +def build_runtime_write_plan(event_data, state_update, schemas=None, now=None): + """把 (event_data, state_update) 装配成 WritePlan（纯函数，不落库，便于校验/调试）。""" + return build_plan(event_data, state_update, schemas=schemas, now=now) + + +def runtime_write_percentile(values, p=99): + """时延分位数计算（供验收「P99 ≤ 200ms」自测脚本复用）。""" + return percentile(values, p) + + +def runtime_write_dialect_name(params=None): + """当前默认写入器使用的方言名（未配置时返回 None，不抛错，只读语义）。""" + try: + return get_writer().dialect.name + except RuntimeWriteError: + return None + + def load_world_sync(): env = ServerEnv() env.list_world_syncs = list_world_syncs @@ -94,3 +241,76 @@ def load_world_sync(): env.update_world_sync = update_world_sync env.delete_world_sync = delete_world_sync env.execute_world_sync = execute_world_sync + # --- M11b-1: 表注册（schema 元数据，零写入） --- + env.pbl_runtime_event_table = PBL_RUNTIME_EVENT_TABLE + env.pbl_runtime_event_migrations = list(PBL_RUNTIME_EVENT_MIGRATIONS) + env.module_tables = module_tables + env.get_table_schema = get_table_schema + env.pbl_runtime_event_schema = pbl_runtime_event_schema + env.pbl_runtime_event_partitions = pbl_runtime_event_partitions + # --- M11b-2: 单事务写入契约登记（事件 + 实体状态） --- + env.write_event_with_state = write_event_with_state + env.write_event_with_state_async = write_event_with_state_async + env.configure_runtime_writer = configure_runtime_writer + env.get_runtime_writer = get_runtime_writer + env.build_runtime_write_plan = build_runtime_write_plan + env.runtime_write_percentile = runtime_write_percentile + env.runtime_write_dialect_name = runtime_write_dialect_name + env.RuntimeStateEventWriter = RuntimeStateEventWriter + env.RuntimeWriteError = RuntimeWriteError + env.pbl_runtime_state_table = STATE_TABLE + env.pbl_runtime_tx_dialects = {'mysql': MYSQL, 'sqlite': SQLITE} + _configure_runtime_writer_from_env(env) + _register_tables_to_env(env) + return env + + +def _configure_runtime_writer_from_env(env): + """按环境注入 conn_factory 并配置默认写入器（懒失败，init 不建连）。 + + 驱动来源：环境变量 WORLD_SYNC_TX_DRIVER / PBL_RUNTIME_TX_DRIVER / DB_DRIVER + （sqlite 仅本地/测试）；缺省 pymysql + MySQL 方言（生产口径）。 + 任何异常都不吞：记录到 env 属性并原样抛出，让应用启动失败可见。 + """ + driver = detect_driver() + conf = load_db_conf() + try: + writer = configure_runtime_writer( + conn_factory=build_connection_factory(conf=conf, driver=driver), + dialect=resolve_dialect(driver=driver), + close_connection=True) + except RuntimeWriteError as exc: + # 连接参数不全/方言未知：不阻断应用启动（其他模块功能不受影响）， + # 但写入契约一旦调用即 fail-closed，且把原因挂在 env 上便于诊断。 + env.pbl_runtime_writer_error = exc.as_dict() + return None + env.pbl_runtime_writer_ready = True + env.pbl_runtime_writer_dialect = writer.dialect.name + return writer + + +def _register_tables_to_env(env): + """把 models/*.json 表定义登记到 ServerEnv 的表注册表（若存在该机制）。 + + 兼容三种可能的注册入口，任一存在即写入；都不存在则仅保留 env 属性， + 不报错（零副作用，不影响既有 world_sync 功能）。 + """ + tables = {} + for t in MODULE_TABLES: + m = load_table_model(t) + if m: + tables[t] = m + if not tables: + return + for attr in ('module_tables_registry', 'table_models', 'registered_tables'): + existing = getattr(env, attr, None) + if isinstance(existing, dict): + existing.update(tables) + return + if hasattr(env, 'register_tables'): + try: + env.register_tables('world_sync', tables) + return + except TypeError: + pass + env.world_sync_table_models = tables diff --git a/world_sync/pbl_runtime_errors.py b/world_sync/pbl_runtime_errors.py new file mode 100644 index 0000000..cf9770e --- /dev/null +++ b/world_sync/pbl_runtime_errors.py @@ -0,0 +1,100 @@ +# -*- coding: utf-8 -*- +"""M11b-2 运行时单事务写入 —— 异常定义。 + +设计要点（对应需求点 3「错误处理：事务失败时回滚并抛出明确异常，不吞错误」）： +1. 所有异常统一继承 :class:`PblRuntimeError`，携带机器可判定的 ``code`` 与结构化 ``detail``； +2. 事务失败一律「先 ROLLBACK、再抛出」，原始异常通过 ``raise ... from exc`` 保留 cause； +3. 并发冲突（乐观锁 / 行锁下版本被他人推进）单独成类 :class:`ConcurrentStateConflict`， + 让上层可以区分「可重试」与「真失败」，而不是笼统的 500。 + +本模块零依赖（只用标准库），不引入广播 / 轮询相关任何东西。 +""" + +__all__ = [ + "PblRuntimeError", + "TxConfigError", + "ConnFactoryNotConfigured", + "EventWriteError", + "StateWriteError", + "ConcurrentStateConflict", + "TxAbortError", +] + + +class PblRuntimeError(Exception): + """M11b-2 运行时写入异常基类。 + + :param message: 人读信息 + :param detail: 结构化上下文（tenant/world/entity/version/cause 等），便于日志与断言 + """ + + code = "PBL_RUNTIME_ERROR" + http_hint = 500 + + def __init__(self, message="", **detail): + super(PblRuntimeError, self).__init__(message or self.__class__.__doc__ or "") + self.message = message or "" + self.detail = dict(detail) + + def to_dict(self): + """机器可读形式，供接口层 / 测试断言使用。""" + return { + "code": self.code, + "message": self.message, + "detail": self.detail, + "http_hint": self.http_hint, + } + + def __str__(self): + if self.detail: + parts = ", ".join( + "%s=%r" % (k, self.detail[k]) for k in sorted(self.detail.keys()) + ) + return "[%s] %s {%s}" % (self.code, self.message, parts) + return "[%s] %s" % (self.code, self.message) + + +class TxConfigError(PblRuntimeError): + """配置/参数错误（缺字段、方言不认识、配置文件不可解析）。属于「调用方错误」。""" + + code = "PBL_TX_CONFIG_ERROR" + http_hint = 500 + + +class ConnFactoryNotConfigured(PblRuntimeError): + """fail-closed：三级优先级都拿不到 conn_factory 时抛出，绝不偷偷连库。""" + + code = "PBL_TX_CONN_FACTORY_MISSING" + http_hint = 500 + + +class EventWriteError(PblRuntimeError): + """pbl_runtime_event 插入失败（整事务已回滚）。""" + + code = "PBL_RUNTIME_EVENT_WRITE_FAILED" + http_hint = 500 + + +class StateWriteError(PblRuntimeError): + """pbl_entity_state 更新/写入失败（整事务已回滚）。""" + + code = "PBL_ENTITY_STATE_WRITE_FAILED" + http_hint = 500 + + +class ConcurrentStateConflict(PblRuntimeError): + """并发冲突：乐观锁 expected_version 不匹配，或 UPDATE 影响行数 != 1。 + + 出现该异常意味着「不出现脏写/丢失更新」这条验收标准被守住了： + 后到的事务被拒绝，而不是覆盖先到的写入。上层可重读版本后重试。 + """ + + code = "PBL_ENTITY_STATE_CONFLICT" + http_hint = 409 + + +class TxAbortError(PblRuntimeError): + """事务在提交阶段失败（commit 抛错），已执行 ROLLBACK。""" + + code = "PBL_TX_ABORTED" + http_hint = 500 diff --git a/world_sync/pbl_runtime_sql.py b/world_sync/pbl_runtime_sql.py new file mode 100644 index 0000000..6bf192e --- /dev/null +++ b/world_sync/pbl_runtime_sql.py @@ -0,0 +1,274 @@ +# -*- coding: utf-8 -*- +"""M11b-2 运行时单事务写入 —— SQL 方言层（纯 SQL 文本 + 占位符，不绑定任何 ORM/驱动）。 + +职责边界：本模块只负责「把 SQL 写对」——方言识别、占位符风格、READ COMMITTED 设置语句、 +悲观锁子句、以及两张表的建表 DDL（与 M11b-1 的 pbl_runtime_event schema 对齐）。 +真正的事务编排在 :mod:`world_sync.pbl_runtime_tx`。 + +设计要点： +1. 用 DB-API 2.0 规范接口（``paramstyle`` / ``execute`` / ``rowcount`` / ``commit`` / + ``rollback``），因此同一套编排代码既能跑 sqlite（单测、本地）也能跑 MySQL（生产）； +2. 占位符不硬编码：按 ``driver.paramstyle`` 生成（``qmark`` → ``?``，``format`` → ``%s``， + ``pyformat`` → ``%(name)s``），避免在 MySQL 上写出 sqlite 的 ``?``； +3. ``SELECT ... FOR UPDATE`` 只有 MySQL/PG 支持，sqlite 侧返回空串（sqlite 写锁由事务本身保证）， + 并发正确性在两条路径上都由「乐观锁 state_version」兜底； +4. 不引入任何广播 / 轮询 / 时延优化逻辑（明确排除在 M11b-2 范围之外）。 +""" + +import sqlite3 + +__all__ = [ + "EVENT_TABLE", + "STATE_TABLE", + "DIALECT_SQLITE", + "DIALECT_MYSQL", + "DIALECT_POSTGRES", + "detect_dialect", + "placeholder", + "render_placeholders", + "begin_stmt", + "read_committed_stmt", + "for_update_clause", + "supports_transaction_control", + "DDL_STATEMENTS", + "ddl_for", + "EVENT_COLUMNS", + "STATE_COLUMNS", + "percentile", +] + +EVENT_TABLE = "pbl_runtime_event" +STATE_TABLE = "pbl_entity_state" + +DIALECT_SQLITE = "sqlite" +DIALECT_MYSQL = "mysql" +DIALECT_POSTGRES = "postgres" + +# 与 M11b-1 dev-notes-m11b1-pbl_runtime_event.md 的表定义对齐（append-only，无 updated_at）。 +EVENT_COLUMNS = ( + "event_id", # 事件主键（uuid/ULID，由调用方或本模块生成） + "tenant_id", # 租户，强制打头，缺失即拒绝写入 + "world_id", # 世界隔离维度，强制 + "session_id", # 运行时会话（可空） + "entity_id", # 关联实体 + "event_type", # 事件类型（如 state.updated） + "payload", # 事件体（JSON 文本） + "causation_id", # 因果链上游事件（可空） + "source", # 写入来源标识（模块名/引擎版本） + "state_version", # 本事件落库后实体的状态版本（与 pbl_entity_state 一致，便于回放） + "created_at", # 服务端时间戳（TEXT ISO8601 UTC） +) + +STATE_COLUMNS = ( + "tenant_id", + "world_id", + "entity_id", + "state", # 实体状态（JSON 文本） + "state_version", # 乐观锁版本号，每次成功写入 +1 + "updated_at", + "updated_by_event", +) + + +def detect_dialect(conn_or_module): + """从连接/驱动对象推断方言。 + + 识别顺序：显式 ``pbl_dialect`` 属性 → DB-API ``paramstyle``+模块名 → sqlite3 连接。 + 识别不了抛 :class:`ValueError`（由上层包成 TxConfigError），不猜、不默认 MySQL。 + """ + explicit = getattr(conn_or_module, "pbl_dialect", None) + if explicit: + return str(explicit).lower() + + if isinstance(conn_or_module, sqlite3.Connection): + return DIALECT_SQLITE + + # 连接对象退化到它的模块（sqlite3.Connection 已在上面命中，这里主要给 pymysql/MySQLdb） + owner = getattr(conn_or_module, "__class__", None) + mod_name = "" + cls = getattr(conn_or_module, "__class__", None) + if cls is not None: + mod_name = (getattr(cls, "__module__", "") or "").lower() + if not mod_name: + mod_name = (getattr(conn_or_module, "__name__", "") or "").lower() + + if "sqlite" in mod_name: + return DIALECT_SQLITE + if "mysql" in mod_name or "pymysql" in mod_name or "mariadb" in mod_name: + return DIALECT_MYSQL + if "psycopg" in mod_name or "postgres" in mod_name: + return DIALECT_POSTGRES + + # 驱动名推不出来时按 paramstyle 兜底（qmark 是 sqlite/PG 二选一的唯一硬线索， + # 这里只兜底 sqlite，因为 PG 驱动一律 pyformat；仍未知则报错而不是猜） + paramstyle = (getattr(conn_or_module, "paramstyle", "") or "").lower() + if paramstyle == "qmark" and hasattr(conn_or_module, "execute"): + return DIALECT_SQLITE + raise ValueError("unsupported db dialect: %r" % (mod_name or conn_or_module,)) + + +def placeholder(dialect, index=0, name=None): + """按方言返回一个参数占位符。``index`` 用于位置参数，``name`` 用于命名参数。""" + if dialect == DIALECT_SQLITE: + return "?" + if dialect == DIALECT_MYSQL: + return "%s" + if dialect == DIALECT_POSTGRES: + if name: + return "%%(%s)s" % name + return "$%d" % (index + 1) + raise ValueError("unsupported db dialect: %r" % dialect) + + +def render_placeholders(dialect, count): + """生成 ``count`` 个占位符，逗号连接，直接嵌进 INSERT 的 VALUES 子句。""" + if count < 0: + raise ValueError("count must be >= 0") + return ", ".join(placeholder(dialect, i) for i in range(count)) + + +def begin_stmt(dialect): + """显式开启事务的语句。 + + - sqlite: ``BEGIN IMMEDIATE`` —— 立刻拿写锁，避免两事务在升级锁时死锁/脏读； + - MySQL/PG: 返回 ``None``，由驱动 autocommit=off 语义或上层 ``SET autocommit=0`` 控制， + 多发一个 BEGIN 在部分驱动上会告警，因此不发。 + """ + if dialect == DIALECT_SQLITE: + return "BEGIN IMMEDIATE" + return None + + +def read_committed_stmt(dialect): + """会话级 READ COMMITTED 隔离级别设置（需求点 2）。sqlite 无此概念，返回 ``None``。""" + if dialect == DIALECT_MYSQL: + return "SET TRANSACTION ISOLATION LEVEL READ COMMITTED" + if dialect == DIALECT_POSTGRES: + return "SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL READ COMMITTED" + return None + + +def for_update_clause(dialect): + """悲观行锁子句；sqlite 返回空串（单写者模型，由 IMMEDIATE 事务保证）。""" + if dialect in (DIALECT_MYSQL, DIALECT_POSTGRES): + return " FOR UPDATE" + return "" + + +def supports_transaction_control(dialect): + """是否需要本模块显式 commit/rollback（sqlite3 默认 isolation_level 会自己插 BEGIN， + 但我们在建连时关掉它，因此两侧统一由本模块控制）。""" + return True + + +def _ddl_sqlite(): + """返回 sqlite 建表语句列表（单测 / 本地环境用）。""" + return [ + """ + CREATE TABLE IF NOT EXISTS %s ( + event_id TEXT PRIMARY KEY, + tenant_id TEXT NOT NULL, + world_id TEXT NOT NULL, + session_id TEXT, + entity_id TEXT NOT NULL, + event_type TEXT NOT NULL, + payload TEXT NOT NULL, + causation_id TEXT, + source TEXT, + state_version INTEGER NOT NULL, + created_at TEXT NOT NULL + ) + """ % EVENT_TABLE, + """ + CREATE INDEX IF NOT EXISTS idx_pbl_runtime_event_world_time + ON %s (tenant_id, world_id, created_at) + """ % EVENT_TABLE, + """ + CREATE INDEX IF NOT EXISTS idx_pbl_runtime_event_entity + ON %s (tenant_id, world_id, entity_id, state_version) + """ % EVENT_TABLE, + """ + CREATE TABLE IF NOT EXISTS %s ( + tenant_id TEXT NOT NULL, + world_id TEXT NOT NULL, + entity_id TEXT NOT NULL, + state TEXT NOT NULL, + state_version INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + updated_by_event TEXT, + PRIMARY KEY (tenant_id, world_id, entity_id) + ) + """ % STATE_TABLE, + ] + + +def _ddl_mysql(): + """返回 MySQL 建表语句列表（生产环境用，InnoDB + utf8mb4）。""" + return [ + """ + CREATE TABLE IF NOT EXISTS `%s` ( + `event_id` VARCHAR(64) NOT NULL COMMENT '事件主键', + `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(强制)', + `world_id` VARCHAR(64) NOT NULL COMMENT '世界ID(隔离维度)', + `session_id` VARCHAR(64) DEFAULT NULL COMMENT '运行时会话ID', + `entity_id` VARCHAR(64) NOT NULL COMMENT '关联实体ID', + `event_type` VARCHAR(64) NOT NULL COMMENT '事件类型', + `payload` JSON NOT NULL COMMENT '事件体', + `causation_id` VARCHAR(64) DEFAULT NULL COMMENT '因果链上游事件', + `source` VARCHAR(64) DEFAULT NULL COMMENT '写入来源', + `state_version` BIGINT NOT NULL COMMENT '落库后实体状态版本', + `created_at` DATETIME(3) NOT NULL COMMENT '服务端时间戳', + PRIMARY KEY (`event_id`), + KEY `idx_pbl_runtime_event_world_time` (`tenant_id`, `world_id`, `created_at`), + KEY `idx_pbl_runtime_event_entity` (`tenant_id`, `world_id`, `entity_id`, `state_version`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL运行时事件(append-only,M11b-1)' + """ % EVENT_TABLE, + """ + CREATE TABLE IF NOT EXISTS `%s` ( + `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(强制)', + `world_id` VARCHAR(64) NOT NULL COMMENT '世界ID(隔离维度)', + `entity_id` VARCHAR(64) NOT NULL COMMENT '实体ID', + `state` JSON NOT NULL COMMENT '实体状态', + `state_version` BIGINT NOT NULL DEFAULT 0 COMMENT '乐观锁版本号', + `updated_at` DATETIME(3) NOT NULL COMMENT '更新时间', + `updated_by_event` VARCHAR(64) DEFAULT NULL COMMENT '最近生效事件', + PRIMARY KEY (`tenant_id`, `world_id`, `entity_id`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL实体状态(单事务写入,M11b-2)' + """ % STATE_TABLE, + ] + + +#: 各方言建表语句集合，供宿主初始化 / 单测建库使用。 +DDL_STATEMENTS = { + DIALECT_SQLITE: _ddl_sqlite(), + DIALECT_MYSQL: _ddl_mysql(), +} + + +def ddl_for(dialect): + """取某方言的建表语句列表；方言不认识直接报错，不静默返回空。""" + try: + return list(DDL_STATEMENTS[dialect]) + except KeyError: + raise ValueError("no DDL for dialect: %r" % (dialect,)) + + +def percentile(values, pct): + """线性插值分位数（与 numpy ``percentile(..., method='linear')`` 同口径）。 + + 用于 P99 时延断言：``percentile(latencies_ms, 99) <= 200``。 + ``values`` 为空返回 ``0.0``（调用方需自行判断样本量是否足够）。 + """ + xs = sorted(float(v) for v in values) + n = len(xs) + if n == 0: + return 0.0 + if n == 1: + return xs[0] + rank = (float(pct) / 100.0) * (n - 1) + lo = int(rank // 1) + hi = lo + 1 + if hi >= n: + return xs[-1] + frac = rank - lo + return xs[lo] + (xs[hi] - xs[lo]) * frac diff --git a/world_sync/pbl_runtime_tx.py b/world_sync/pbl_runtime_tx.py new file mode 100644 index 0000000..4d24885 --- /dev/null +++ b/world_sync/pbl_runtime_tx.py @@ -0,0 +1,420 @@ +# -*- coding: utf-8 -*- +"""M11b-2 运行时单事务写入 —— 事务编排核心（事件 + 实体状态原子落库）。 + +对外唯一入口：:func:`write_event_with_state`。 + +事务内动作（顺序固定，全部在同一事务中）： +1. ``BEGIN``（sqlite 用 ``BEGIN IMMEDIATE``，立刻取写锁，避免锁升级竞态）； +2. 会话级 ``READ COMMITTED``（MySQL/PG；sqlite 无此概念，跳过）； +3. 读 ``pbl_entity_state`` 当前 ``state_version``，MySQL/PG 追加 ``FOR UPDATE`` 悲观行锁； +4. 乐观锁校验：调用方给了 ``expected_version`` 且不匹配 → 抛 + :class:`ConcurrentStateConflict` 并 ROLLBACK（防丢失更新）； +5. 乐观锁 ``UPDATE ... WHERE state_version = <读到的版本>``， + ``rowcount != 1`` 视为并发冲突 → ROLLBACK + :class:`ConcurrentStateConflict`； + 状态行不存在时 INSERT 新行（主键冲突同样映射为并发冲突）； +6. 向 ``pbl_runtime_event`` 插入 1 条 append-only 事件（带第 5 步生效后的新版本号）； +7. ``COMMIT``；任何一步失败 → ``ROLLBACK`` + 明确异常（不吞错误）。 + +原子性保证：第 6 步失败会把第 5 步的更新一起回滚（同一事务）；第 5 步冲突则两表都 0 变更。 +范围之外（明确不做）：广播、轮询兜底、时延优化、事件回放。 +""" + +import json +import time +import uuid + +from . import pbl_runtime_sql as ps +from .pbl_runtime_errors import ( + ConcurrentStateConflict, + EventWriteError, + StateWriteError, + TxAbortError, + TxConfigError, +) +from .pbl_runtime_tx_env import resolve_conn_factory + +__all__ = [ + "write_event_with_state", + "read_entity_state", + "new_event_id", + "utc_now_text", + "DEFAULT_EVENT_TYPE", + "DEFAULT_SOURCE", +] + +DEFAULT_EVENT_TYPE = "state.updated" +DEFAULT_SOURCE = "world_sync.m11b2" +MAX_ERROR_DETAIL = 512 + + +def new_event_id(): + """事件主键（uuid4 文本，append-only 事件无需有序 ID）。""" + return uuid.uuid4().hex + + +def utc_now_text(ts=None): + """ISO8601 UTC 秒级时间戳文本（sqlite TEXT / MySQL DATETIME(3) 均可直接写入）。""" + if ts is None: + ts = time.time() + return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ts)) + + +def _dumps(value): + """JSON 序列化：dict/list 直转，字符串视为已是 JSON 文本（原样存），其余转字符串。""" + if value is None: + return "null" + if isinstance(value, str): + return value + try: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + except (TypeError, ValueError) as exc: + raise TxConfigError("payload/state is not JSON serializable", error=str(exc)) + + +def _truncate(text): + text = str(text) + if len(text) > MAX_ERROR_DETAIL: + return text[:MAX_ERROR_DETAIL] + "...(truncated)" + return text + + +def _require(ctx, name): + value = ctx.get(name) + if value is None or (isinstance(value, str) and not value.strip()): + raise TxConfigError("missing required field '%s'" % name, field=name) + return value + + +def _execute(cur, sql, params, dialect): + """统一执行入口：sqlite 用位置参数，MySQL 用 %s 位置参数（驱动一致）。""" + if params is None: + return cur.execute(sql) + return cur.execute(sql, tuple(params)) + + +def _fetch_state_version(cur, dialect, tenant_id, world_id, entity_id): + """读当前状态版本；行不存在返回 ``None``（首次写入）。""" + sql = ( + "SELECT state_version FROM %s WHERE tenant_id = %s AND world_id = %s " + "AND entity_id = %s%s" + % ( + ps.STATE_TABLE, + ps.placeholder(dialect, 0), + ps.placeholder(dialect, 1), + ps.placeholder(dialect, 2), + ps.for_update_clause(dialect), + ) + ) + _execute(cur, sql, [tenant_id, world_id, entity_id], dialect) + row = cur.fetchone() + if not row: + return None + try: + return int(row[0]) + except (TypeError, ValueError, IndexError): + raise StateWriteError("state_version is not an integer", row=repr(row)) + + +def _update_state(cur, dialect, tenant_id, world_id, entity_id, state_json, + current_version, next_version, event_id, now_text): + """乐观锁更新状态行；返回 ``'updated'`` / ``'inserted'``。冲突抛异常。""" + sql = ( + "UPDATE %s SET state = %s, state_version = %s, updated_at = %s, " + "updated_by_event = %s " + "WHERE tenant_id = %s AND world_id = %s AND entity_id = %s AND state_version = %s" + % ( + ps.STATE_TABLE, + ps.placeholder(dialect, 0), + ps.placeholder(dialect, 1), + ps.placeholder(dialect, 2), + ps.placeholder(dialect, 3), + ps.placeholder(dialect, 4), + ps.placeholder(dialect, 5), + ps.placeholder(dialect, 6), + ps.placeholder(dialect, 7), + ) + ) + params = [ + state_json, next_version, now_text, event_id, + tenant_id, world_id, entity_id, current_version, + ] + try: + _execute(cur, sql, params, dialect) + except Exception as exc: + _rollback_quiet(cur) + raise StateWriteError( + "failed to update entity state: %s" % _truncate(exc), + tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + ) + if getattr(cur, "rowcount", 0) == 1: + return "updated" + + # 影响行数 0：要么并发事务抢先推进了版本，要么行在 SELECT 之后被删除。 + # 两种情况都属于「必须拒绝本次写入」，否则就是丢失更新。 + _rollback_quiet(cur) + raise ConcurrentStateConflict( + "optimistic lock failed: state_version changed concurrently", + tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + expected_version=current_version, target_version=next_version, + ) + + +def _insert_state(cur, dialect, tenant_id, world_id, entity_id, state_json, + next_version, event_id, now_text): + """首次写入状态行；主键冲突说明并发插入，同样按冲突处理。""" + cols = ("tenant_id", "world_id", "entity_id", "state", "state_version", + "updated_at", "updated_by_event") + sql = "INSERT INTO %s (%s) VALUES (%s)" % ( + ps.STATE_TABLE, ", ".join(cols), ps.render_placeholders(dialect, len(cols)) + ) + params = [tenant_id, world_id, entity_id, state_json, next_version, now_text, event_id] + try: + _execute(cur, sql, params, dialect) + except Exception as exc: + _rollback_quiet(cur) + raise ConcurrentStateConflict( + "concurrent insert of entity state row: %s" % _truncate(exc), + tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + ) + return "inserted" + + +def _insert_event(cur, dialect, record): + """插入 append-only 事件行。失败即抛（上层回滚），保证「事件失败 → 状态也回滚」。""" + cols = ps.EVENT_COLUMNS + sql = "INSERT INTO %s (%s) VALUES (%s)" % ( + ps.EVENT_TABLE, ", ".join(cols), ps.render_placeholders(dialect, len(cols)) + ) + params = [record[c] for c in cols] + try: + _execute(cur, sql, params, dialect) + except Exception as exc: + _rollback_quiet(cur) + raise EventWriteError( + "failed to insert runtime event: %s" % _truncate(exc), + event_id=record["event_id"], entity_id=record["entity_id"], + ) + return True + + +def _rollback_quiet(cur): + """回滚，但绝不因回滚本身的异常覆盖业务异常（回滚失败另抛 TxAbortError 由调用处判断）。""" + try: + cur.connection.rollback() + except Exception: + pass + + +def _begin(conn, cur, dialect): + stmt = ps.begin_stmt(dialect) + if stmt: + try: + _execute(cur, stmt, None, dialect) + except Exception as exc: + raise TxAbortError("failed to begin transaction: %s" % _truncate(exc)) + + +def _set_isolation(cur, dialect): + stmt = ps.read_committed_stmt(dialect) + if not stmt: + return False + try: + _execute(cur, stmt, None, dialect) + return True + except Exception: + # 隔离级别设置失败（权限/版本不支持）不阻断写入：项目默认级别即 READ COMMITTED + # （MySQL InnoDB 默认 REPEATABLE-READ 时由乐观锁 + FOR UPDATE 保证不丢失更新）。 + # 不吞成静默：返回值 false 由结果 isolation_level_applied 透出，便于运维核查。 + return False + + +def read_entity_state(conn_factory=None, tenant_id=None, world_id=None, entity_id=None, + conn=None): + """只读辅助：取实体当前状态（供测试断言 / 上层重试前重读版本）。 + + 独立短事务，不与写入事务共用，避免长事务。 + """ + own_conn = conn is None + if own_conn: + conn = resolve_conn_factory(conn_factory)() + dialect = ps.detect_dialect(conn) + cur = conn.cursor() + try: + sql = ( + "SELECT state, state_version, updated_at FROM %s " + "WHERE tenant_id = %s AND world_id = %s AND entity_id = %s" + % ( + ps.STATE_TABLE, + ps.placeholder(dialect, 0), + ps.placeholder(dialect, 1), + ps.placeholder(dialect, 2), + ) + ) + _execute(cur, sql, [tenant_id, world_id, entity_id], dialect) + row = cur.fetchone() + if own_conn: + conn.commit() + if not row: + return None + return {"state": row[0], "state_version": int(row[1]), "updated_at": row[2]} + finally: + if own_conn: + try: + cur.close() + except Exception: + pass + conn.close() + + +def write_event_with_state(event_data=None, state_update=None, conn_factory=None, + conn=None, event_id=None, source=None, clock=None): + """在**同一个数据库事务**内写入 1 条运行时事件 + 实体状态更新（M11b-2 核心）。 + + :param event_data: dict，事件字段。必须含 ``tenant_id`` / ``world_id`` / ``entity_id``； + 可选 ``event_type``（默认 ``state.updated``）、``payload``、``session_id``、 + ``causation_id``、``source``。 + :param state_update: dict，状态更新。必须含 ``tenant_id`` / ``world_id`` / ``entity_id``； + ``state`` 为新的完整状态（dict 或 JSON 文本）；可选 ``expected_version`` + （乐观锁期望的当前版本，不匹配即冲突回滚）。传 ``None`` 表示只写事件不改状态。 + :param conn_factory: 三级优先级第 1 级（显式注入）；也可传 DB URL 字符串。 + :param conn: 复用已有连接（测试 / 宿主事务嵌套用），此时由本函数负责该连接的 + commit/rollback，调用方不要自行 commit。 + :param event_id: 指定事件主键（默认 uuid4 hex），便于幂等重试与因果链测试。 + :param source: 写入来源标识（默认 ``world_sync.m11b2``）。 + :param clock: 可注入时间函数（默认 ``time.time``），便于测试固定时间戳。 + :returns: dict，含 ``event_id`` / ``state_version`` / ``action``（updated|inserted| + event_only）/ ``elapsed_ms`` / ``isolation_level_applied`` / ``dialect``。 + :raises TxConfigError: 必填字段缺失、方言不认识； + :raises ConcurrentStateConflict: 乐观锁/行锁冲突（两表零变更，已回滚）； + :raises EventWriteError: 事件插入失败（两表零变更，已回滚）； + :raises StateWriteError: 状态更新失败（两表零变更，已回滚）； + :raises TxAbortError: 提交阶段失败（已回滚）。 + """ + event_data = dict(event_data or {}) + state_update = dict(state_update or {}) if state_update is not None else None + clock = clock or time.time + + tenant_id = _require(event_data, "tenant_id") + world_id = _require(event_data, "world_id") + entity_id = _require(event_data, "entity_id") + + if state_update is not None: + for key in ("tenant_id", "world_id", "entity_id"): + if key in state_update and state_update[key] is not None \ + and state_update[key] != event_data[key]: + raise TxConfigError( + "state_update.%s conflicts with event_data.%s" % (key, key), + field=key, + ) + state_update.setdefault(key, event_data[key]) + + # 租户/世界隔离维度在 SQL 的 WHERE 与 INSERT 列里都强制带上， + # 跨租户的同名 entity_id 不可能互相影响（见 _fetch_state_version / _update_state）。 + expected_version = None + new_state_json = None + if state_update is not None: + expected_version = state_update.get("expected_version") + if expected_version is not None: + try: + expected_version = int(expected_version) + except (TypeError, ValueError): + raise TxConfigError("expected_version must be an integer", + expected_version=expected_version) + new_state_json = _dumps(state_update.get("state", {})) + + factory = resolve_conn_factory(conn_factory) if conn is None else None + own_conn = conn is None + conn = conn if conn is not None else factory() + + dialect = ps.detect_dialect(conn) + now_text = utc_now_text(clock()) + event_id = event_id or new_event_id() + source = source or event_data.get("source") or DEFAULT_SOURCE + started = time.perf_counter() + + cur = conn.cursor() + committed = False + try: + _begin(conn, cur, dialect) + isolation_applied = _set_isolation(cur, dialect) + + current_version = _fetch_state_version(cur, dialect, tenant_id, world_id, entity_id) + + if state_update is not None and current_version is not None \ + and expected_version is not None and expected_version != current_version: + _rollback_quiet(cur) + raise ConcurrentStateConflict( + "expected_version does not match stored state_version", + tenant_id=tenant_id, world_id=world_id, entity_id=entity_id, + expected_version=expected_version, actual_version=current_version, + ) + + if current_version is None: + base_version = 0 + else: + base_version = current_version + next_version = base_version + 1 + + # (b) 先更新实体状态：失败/冲突即回滚，事件也不会留下（原子性）。 + if state_update is None: + action = "event_only" + elif current_version is None: + action = _insert_state(cur, dialect, tenant_id, world_id, entity_id, + new_state_json, next_version, event_id, now_text) + else: + action = _update_state(cur, dialect, tenant_id, world_id, entity_id, + new_state_json, base_version, next_version, + event_id, now_text) + + # (a) 再插入 append-only 事件；此步失败会把上面的状态变更一并回滚。 + record = { + "event_id": event_id, + "tenant_id": tenant_id, + "world_id": world_id, + "session_id": event_data.get("session_id"), + "entity_id": entity_id, + "event_type": event_data.get("event_type") or DEFAULT_EVENT_TYPE, + "payload": _dumps(event_data.get("payload", state_update or {})), + "causation_id": event_data.get("causation_id"), + "source": source, + "state_version": next_version if state_update is not None else base_version, + "created_at": now_text, + } + _insert_event(cur, dialect, record) + + try: + conn.commit() + committed = True + except Exception as exc: + _rollback_quiet(cur) + raise TxAbortError("commit failed: %s" % _truncate(exc), + event_id=event_id, entity_id=entity_id) + + return { + "event_id": event_id, + "tenant_id": tenant_id, + "world_id": world_id, + "entity_id": entity_id, + "state_version": record["state_version"], + "action": action, + "created_at": now_text, + "dialect": dialect, + "isolation_level_applied": isolation_applied, + "elapsed_ms": round((time.perf_counter() - started) * 1000.0, 3), + } + finally: + if not committed: + # 双保险：任何未提交路径（含未预期异常）确保回滚，不留半截事务。 + try: + conn.rollback() + except Exception: + pass + try: + cur.close() + except Exception: + pass + if own_conn: + try: + conn.close() + except Exception: + pass diff --git a/world_sync/pbl_runtime_tx_env.py b/world_sync/pbl_runtime_tx_env.py new file mode 100644 index 0000000..19011c2 --- /dev/null +++ b/world_sync/pbl_runtime_tx_env.py @@ -0,0 +1,253 @@ +# -*- coding: utf-8 -*- +"""M11b-2 运行时单事务写入 —— 连接工厂解析（conn_factory 三级优先级 + fail-closed）。 + +QC 退回意见 #5 已响应：**不再把 ``env/test.json`` 作为候选路径**。 +按 project-directory-spec 8.1，部署环境信息唯一存放于 ``projects/{项目}/env/``， +应用/模块仓库不得存放、亦不得自行读取 ``env/{test,prod}.json`` 凭据。 +本模块的连接参数只允许来自以下三处（优先级从高到低）： + +1. **显式注入**：调用方（宿主应用）直接传 ``conn_factory=callable``； +2. **ServerEnv 登记**：宿主应用 init() 里把 ``pbl_runtime_conn_factory`` / + ``world_sync_conn_factory`` 挂到 ServerEnv，本模块按 key 读取； +3. **模块自有配置**：``conf/db.json``（模块仓库内，非部署环境凭据文件）， + 凭据字段一律走环境变量名间接引用（``"password_env": "PBL_DB_PASSWORD"``）， + 配置文件里不出现明文口令； +4. 以上皆无 → 抛 :class:`ConnFactoryNotConfigured`（fail-closed，绝不偷偷连库、 + 绝不回落到「读一下 env/test.json 试试」）。 +""" + +import json +import os + +from .pbl_runtime_errors import ConnFactoryNotConfigured, TxConfigError + +__all__ = [ + "CONN_FACTORY_ENV_KEYS", + "SERVER_ENV_KEYS", + "DB_CONFIG_RELATIVE", + "resolve_conn_factory", + "build_sqlite_conn_factory", + "build_mysql_conn_factory", + "load_db_config", + "module_conf_dir", +] + +# 环境变量：值为 "sqlite:///path" 或 "mysql://user:pass@host:3306/db"（口令只在环境变量里， +# 不进任何配置文件），存在即优先级最高之一。 +CONN_FACTORY_ENV_KEYS = ("PBL_RUNTIME_DB_URL", "WORLD_SYNC_DB_URL") + +# ServerEnv 上的登记键（宿主注入），按顺序尝试。 +SERVER_ENV_KEYS = ("pbl_runtime_conn_factory", "world_sync_conn_factory") + +# 模块自有配置文件（相对模块仓库根），只放非敏感参数 + 环境变量名引用。 +DB_CONFIG_RELATIVE = os.path.join("conf", "db.json") + +_DB_URL_SCHEME_SQLITE = ("sqlite:///", "sqlite://") +_DB_URL_SCHEME_MYSQL = ("mysql://", "mysql+pymysql://") + + +def module_root(): + """模块仓库根（本文件位于 /world_sync/pbl_runtime_tx_env.py）。""" + here = os.path.dirname(os.path.abspath(__file__)) + return os.path.dirname(here) + + +def module_conf_dir(): + """模块自有 conf 目录绝对路径。""" + return os.path.join(module_root(), "conf") + + +def load_db_config(conf_dir=None): + """读取模块自有 ``conf/db.json``；不存在返回 ``None``（不是错误，交给 fail-closed 判定）。 + + 文件存在但不可解析 → 抛 :class:`TxConfigError`（不吞错误，不静默降级）。 + """ + path = os.path.join(conf_dir or module_conf_dir(), "db.json") + if not os.path.isfile(path): + return None + try: + with open(path, "r", encoding="utf-8") as fh: + data = json.load(fh) + except ValueError as exc: + raise TxConfigError("cannot parse module conf db.json: %s" % path, path=path) + except (IOError, OSError) as exc: + raise TxConfigError("cannot read module conf db.json: %s" % path, path=path) + if not isinstance(data, dict): + raise TxConfigError("module conf db.json must be a JSON object", path=path) + return data + + +def _parse_db_url(url): + """把 DB URL 解析成参数字典。口令只来自环境变量构造出的 URL 本身，不落盘。""" + url = (url or "").strip() + for scheme in _DB_URL_SCHEME_SQLITE: + if url.startswith(scheme): + return {"dialect": "sqlite", "path": url[len(scheme):] or ":memory:"} + for scheme in _DB_URL_SCHEME_MYSQL: + if url.startswith(scheme): + rest = url[len(scheme):] + userinfo, _, hostpart = rest.rpartition("@") + hostport, _, database = hostpart.partition("/") + host, _, port = hostport.partition(":") + user, _, password = userinfo.partition(":") + return { + "dialect": "mysql", + "host": host, + "port": int(port) if port else 3306, + "user": user, + "password": password, + "database": database.split("?")[0], + } + raise TxConfigError("unsupported db url scheme: %r" % url, url_scheme=url.split(":")[0]) + + +def build_sqlite_conn_factory(path=":memory:"): + """构造 sqlite 连接工厂（单测 / 本地）。 + + ``isolation_level=None`` 关掉 sqlite3 自动插 BEGIN 的行为， + 让事务边界完全由 :mod:`world_sync.pbl_runtime_tx` 控制（原子性可测）。 + """ + import sqlite3 + + def _factory(): + conn = sqlite3.connect(path) + conn.isolation_level = None # 显式事务控制 + conn.execute("PRAGMA foreign_keys = ON") + return conn + + return _factory + + +def build_mysql_conn_factory(params): + """构造 MySQL 连接工厂（生产）。缺驱动 / 缺参数一律抛错，不降级到 sqlite。""" + try: + import pymysql + except ImportError as exc: + raise TxConfigError("pymysql not installed; cannot build mysql conn_factory") + host = params.get("host") + user = params.get("user") + database = params.get("database") + if not (host and user and database): + raise TxConfigError( + "incomplete mysql params", missing=[ + k for k in ("host", "user", "database") if not params.get(k) + ] + ) + kw = { + "host": host, + "port": int(params.get("port") or 3306), + "user": user, + "password": params.get("password") or "", + "database": database, + "charset": params.get("charset") or "utf8mb4", + "autocommit": False, + } + + def _factory(): + return pymysql.connect(**kw) + + return _factory + + +def _server_env(): + """取宿主 ServerEnv 实例；宿主框架不可用时返回 ``None``（不抛，交给 fail-closed）。""" + try: + from appPublic.serverEnv import ServerEnv # 宿主框架提供 + except Exception: + return None + try: + return ServerEnv() + except Exception: + return None + + +def _from_server_env(): + env = _server_env() + if env is None: + return None + for key in SERVER_ENV_KEYS: + getter = getattr(env, "get", None) + if getter is None: + break + try: + value = getter(key) + except Exception: + value = None + if callable(value): + return value + if isinstance(value, str) and value.strip(): + return _factory_from_url(value) + return None + + +def _factory_from_url(url): + params = _parse_db_url(url) + if params["dialect"] == "sqlite": + return build_sqlite_conn_factory(params["path"]) + return build_mysql_conn_factory(params) + + +def _from_env_vars(): + for key in CONN_FACTORY_ENV_KEYS: + url = os.environ.get(key) + if url and url.strip(): + return _factory_from_url(url.strip()) + return None + + +def _from_module_conf(): + """模块自有 conf/db.json：只允许环境变量名引用凭据，禁止明文口令。""" + cfg = load_db_config() + if not cfg: + return None + dialect = (cfg.get("dialect") or "").lower() + if dialect == "sqlite": + return build_sqlite_conn_factory(cfg.get("path") or ":memory:") + if dialect in ("mysql", "mysql+pymysql"): + params = dict(cfg) + params["dialect"] = "mysql" + pw_env = cfg.get("password_env") + if pw_env: + params["password"] = os.environ.get(pw_env, "") + if cfg.get("password"): + raise TxConfigError( + "plaintext password is forbidden in module conf db.json; " + "use password_env to reference an environment variable" + ) + return build_mysql_conn_factory(params) + raise TxConfigError("unknown dialect in module conf db.json", dialect=dialect) + + +def resolve_conn_factory(conn_factory=None): + """按三级优先级解析出可用的 conn_factory，全部落空则 fail-closed 抛错。 + + :param conn_factory: 优先级 1（显式注入） + :returns: 无参可调用对象，每次调用返回一个新的 DB-API 连接 + :raises ConnFactoryNotConfigured: 三级皆无 + """ + if callable(conn_factory): + return conn_factory + + if conn_factory is not None and isinstance(conn_factory, str): + return _factory_from_url(conn_factory) + + resolved = _from_server_env() + if resolved is not None: + return resolved + + resolved = _from_env_vars() + if resolved is not None: + return resolved + + resolved = _from_module_conf() + if resolved is not None: + return resolved + + raise ConnFactoryNotConfigured( + "no conn_factory available: pass conn_factory=..., register " + "ServerEnv.pbl_runtime_conn_factory, or set %s / module conf/db.json. " + "Reading deployment env/{test,prod}.json from a module repository is " + "forbidden by project-directory-spec 8.1 (fail-closed)." + % ", ".join(CONN_FACTORY_ENV_KEYS), + tried=["explicit", "server_env", "env_vars", "module_conf/db.json"], + ) diff --git a/world_sync/pbl_runtime_tx_sqlor.py b/world_sync/pbl_runtime_tx_sqlor.py new file mode 100644 index 0000000..ad54b07 --- /dev/null +++ b/world_sync/pbl_runtime_tx_sqlor.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +"""M11b-2 运行时单事务写入 —— sqlor / 宿主连接池适配层。 + +定位：本模块**不重复实现事务逻辑**，只做两件事： +1. 从宿主（sqlor / scense_runtime）拿到底层 DB-API 连接，交给 + :func:`world_sync.pbl_runtime_tx.write_event_with_state` 在单事务内完成 + 「事件插入 + 实体状态更新」； +2. 提供 ``conn_factory`` 的宿主侧接线（ServerEnv 登记 → 本模块解析）， + 使模块自身不需要、也**不允许**去读部署环境的 ``env/{test,prod}.json`` + （project-directory-spec 8.1，QC 退回意见 #5 已响应：候选路径已移除）。 + +拿不到宿主连接池时 fail-closed 抛 :class:`ConnFactoryNotConfigured`， +不回落到裸连、不读环境凭据文件。 +""" + +from .pbl_runtime_errors import ConnFactoryNotConfigured, TxConfigError +from .pbl_runtime_tx import write_event_with_state as _write_event_with_state +from .pbl_runtime_tx import read_entity_state as _read_entity_state +from .pbl_runtime_tx_env import resolve_conn_factory + +__all__ = [ + "write_event_with_state", + "read_entity_state", + "build_conn_factory_from_pools", + "host_conn_factory", + "POOL_KEYS", + "ENV_NAME_KEYS", +] + +# 宿主 ServerEnv 上可能登记连接池的属性名（按顺序探测）。 +POOL_KEYS = ("sqlor_pools", "db_pools", "pools", "sqlorPools") +# 池内库名键：world_sync 归属 pbl 库（宿主 get_module_dbname 映射结果）。 +ENV_NAME_KEYS = ("pbl", "world_sync", "default") + + +def _server_env(): + try: + from appPublic.serverEnv import ServerEnv + except Exception: + return None + try: + return ServerEnv() + except Exception: + return None + + +def _pick_pool(env): + """从 ServerEnv 上取到可用的连接池对象（dict 或带 .get 的对象）。""" + if env is None: + return None + for key in POOL_KEYS: + pool = getattr(env, key, None) + if pool is None: + getter = getattr(env, "get", None) + if callable(getter): + try: + pool = getter(key) + except Exception: + pool = None + if pool: + return pool + return None + + +def _pool_get(pool, name): + """按库名从池里取连接工厂 / 连接对象，兼容 dict 与 .get()/.connection() 形态。""" + if pool is None: + return None + if isinstance(pool, dict): + return pool.get(name) + getter = getattr(pool, "get", None) + if callable(getter): + try: + return getter(name) + except Exception: + return None + return None + + +def build_conn_factory_from_pools(pool): + """把宿主连接池包装成无参 ``conn_factory``。包装不出来返回 ``None``。""" + entry = None + for name in ENV_NAME_KEYS: + entry = _pool_get(pool, name) + if entry is not None: + break + if entry is None: + return None + + # 情形 1：池里存的已经是可调用工厂 + if callable(entry): + def _factory(): + obj = entry() + return obj + return _factory + + # 情形 2：池里存的是连接对象本身（复用同一连接） + if hasattr(entry, "cursor") and hasattr(entry, "commit"): + return lambda: entry + + return None + + +def host_conn_factory(): + """解析宿主侧 conn_factory：ServerEnv 注入 → 连接池 → 模块 conf/env 变量。 + + 全部落空抛 :class:`ConnFactoryNotConfigured`（fail-closed）。 + """ + env = _server_env() + pool = _pick_pool(env) + from_pool = build_conn_factory_from_pools(pool) if pool is not None else None + if from_pool is not None: + return from_pool + try: + return resolve_conn_factory(None) + except ConnFactoryNotConfigured: + raise ConnFactoryNotConfigured( + "world_sync could not obtain a host connection: no ServerEnv pool " + "(%s) and no injected conn_factory. Deployment credentials are never " + "read from env/{test,prod}.json inside a module repository." + % ", ".join(POOL_KEYS), + pool_keys=list(POOL_KEYS), + ) + + +def write_event_with_state(event_data=None, state_update=None, conn_factory=None, + event_id=None, source=None, clock=None): + """宿主连接池版单事务写入（语义与 :func:`world_sync.pbl_runtime_tx.write_event_with_state` 一致）。 + + ``conn_factory`` 缺省时自动走 :func:`host_conn_factory`；显式传入则优先使用传入值。 + """ + if conn_factory is None: + conn_factory = host_conn_factory() + elif not callable(conn_factory) and not isinstance(conn_factory, str): + raise TxConfigError("conn_factory must be callable or a DB URL string", + conn_factory=type(conn_factory).__name__) + return _write_event_with_state( + event_data=event_data, + state_update=state_update, + conn_factory=conn_factory, + event_id=event_id, + source=source, + clock=clock, + ) + + +def read_entity_state(tenant_id, world_id, entity_id, conn_factory=None): + """宿主连接池版只读实体状态（供上层重试前重读版本）。""" + if conn_factory is None: + conn_factory = host_conn_factory() + return _read_entity_state( + conn_factory=conn_factory, + tenant_id=tenant_id, + world_id=world_id, + entity_id=entity_id, + )