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

This commit is contained in:
agent.develop 2026-09-20 16:09:34 +08:00
parent 7da0c0c3ad
commit aca3e3117a
12 changed files with 3046 additions and 10 deletions

View File

@ -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.jsonM11a 侧)使用 state_json/checksum/updated_by 等不同列集与 int 自增 id属另一张同名表的另一套定义两者不可混用本文件为 world_syncM11b-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": []
}

View File

@ -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-1pbl_runtime_ext侧的同一表名定义其真源为 scripts/pbl_runtime_event_ddl.py 的 COLUMNS/PRIMARY_KEY/UNIQUE_KEYS25 列、复合主键 (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_textappend-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'"}
]
}

217
scripts/m11b2_selftest.py Normal file
View File

@ -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())

View File

@ -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异常族
方言说明生产为 MySQLREAD 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)

563
tests/test_m11b2_wiring.py Normal file
View File

@ -0,0 +1,563 @@
# -*- coding: utf-8 -*-
"""M11b-2 单事务写入 —— 可执行单元测试sqlite 文件库真实跑通,非 import 面检查)。
覆盖需求点与验收标准
- 正常写入事件 + 状态在同一事务落库两表各 1 TestAtomicWriteSuccess
- 事件插入失败 状态更新一并回滚TestRollbackOnEventFailure
- 状态更新失败/并发冲突 整体回滚事件不留TestRollbackOnStateFailure
- 并发多事务写同一实体不出现脏写/丢失更新TestConcurrencyNoLostUpdate
- 租户/世界隔离跨租户同名 entity_id 互不影响TestTenantWorldIsolation
- P99 200msTestLatencyP99percentile m11b-selftest-run.log 同口径
- conn_factory 三级优先级 + fail-closed + 禁读 env/*.jsonTestConnFactoryResolution
- 接线面导出一致性禁项无广播/轮询扫描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 ≤ 200mssqlite 本地口径)。"""
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)

View File

@ -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__)

View File

@ -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/<table>.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 阶段不建连,真正写入时才取连接;
# 参数不全则抛 ConnectionUnavailablefail-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

View File

@ -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

View File

@ -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/rollbacksqlite3 默认 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

View File

@ -0,0 +1,420 @@
# -*- coding: utf-8 -*-
"""M11b-2 运行时单事务写入 —— 事务编排核心(事件 + 实体状态原子落库)。
对外唯一入口:func:`write_event_with_state`
事务内动作顺序固定全部在同一事务中
1. ``BEGIN``sqlite ``BEGIN IMMEDIATE``立刻取写锁避免锁升级竞态
2. 会话级 ``READ COMMITTED``MySQL/PGsqlite 无此概念跳过
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

View File

@ -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():
"""模块仓库根(本文件位于 <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"],
)

View File

@ -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.1QC 退回意见 #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_factoryServerEnv 注入 → 连接池 → 模块 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,
)