deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
fc359f8dc5
commit
ff5f881d51
@ -5,7 +5,7 @@
|
||||
"title": "运行时事件流(append-only)",
|
||||
"primary": ["id", "created_at"],
|
||||
"catelog": "entity",
|
||||
"comment": "【本表定位 · world_sync 侧内联副本】运行时事件流（append-only：禁 UPDATE/DELETE，由权威模块 pbl_runtime_ext 的存储过程 pbl_assert_append_only + BEFORE UPDATE/BEFORE DELETE 触发器强制；按 created_at 做按月 RANGE COLUMNS(created_at) 分区，分区保留 14 个月）。物理主键为复合主键 (id, created_at) —— MySQL/MariaDB 要求分区表的所有唯一键（含主键）必须包含分区列，否则建表报错 1503。【结构权威源 · PM 裁决】PM 裁决（任务 u0EO4x3ajntR41acaHj5H 问答）：选 B —— 以 pbl_runtime_ext 为唯一真源；结构权威定义位于 modules/pbl_runtime_ext/models/pbl_runtime_event.json（M11b-1 已批准产物，本任务禁止反向修改），DDL 派生源为该模块 scripts/pbl_runtime_event_ddl.py 与 scripts/migrations/*.sql，建表/分区/触发器归 pbl_runtime_ext 所有，本文件不复制那套逻辑。【为什么是内联而不是引用】database-table-definition-spec 只承认 summary/fields/indexes/codes 四段式；scripts/validate_models_json.py（M11b-2a 子任务 A 修复版）对 key=schema 引用式写法直接判 FAIL（QC #5），故本文件按真源逐字段内联权威列集，作为 world_sync 侧的消费方契约视图；两份定义的一致性由 scripts/m11b2b_column_diff.py 与 scripts/m11b2b_selftest.py 机械核对，真源列集变更时必须同步本副本。【id 类型偏离处理 · 本子任务的选择与理由】真源 id 为 type=long + auto_increment（BIGINT AUTO_INCREMENT，真源侧已在其 summary[0].comment 自行登记该偏离）；子任务 A 的最终登记表 COMPLIANCE_DEVIATIONS 为空 {}（A 按 QC #5 选 (a)：不放行任何历史失效项），且本任务范围明确禁止修改校验器脚本，因此 world_sync 侧不再走「登记放行 id=long」路线，而是内联为 spec 合规写法 type=str / length=32、不带 auto_increment 键，由应用层生成 32 位 uuid 去横线十六进制串作为主键值（见 world_sync/pbl_runtime_tx.py::new_event_uid，正是 uuid4().hex 32 位）。与真源的 id 类型差异由 world_sync 侧消费方代码适配：写路径 INSERT 不显式提交 id 时交由真源 DDL 的 AUTO_INCREMENT 生成（BIGINT），本地/测试库（scripts 渲染的 sqlite DDL）则使用应用层生成的 str(32)；对外幂等与跨系统追溯一律使用 event_uid（str 64，uk_event_uid 唯一键），不依赖 id 的物理类型。此差异为有意选择、非疏漏，评审请勿据此判为结构缺陷。【消费方清单】modules/world_sync/world_sync/pbl_runtime_sql.py（EVENT_TABLE/EVENT_COLUMNS）、modules/world_sync/world_sync/pbl_runtime_tx.py::write_event_with_state（单事务事件+状态写入）、modules/world_sync/world_sync/pbl_runtime_authority.py（列契约解析层）；上述代码引用的列名必须全部出现在本文件 fields[] 中。【打包镜像说明】apps/scense/pkgs/world_sync/models/pbl_runtime_event.json 是本文件的打包镜像，只由 modules/world_sync/scripts/sync_models_mirror.py 生成，禁止手工双写；一致性由 modules/world_sync/scripts/m11b2b_selftest.py 的 sha256 断言守护。【私有键剥离说明】真源使用的 spec 未定义顶层扩展键 module/engine/charset/collate/tenant_scoped/append_only/write_paths/guards/partition/ddl_source，以及 fields[].ddl_type/comment/auto_increment、indexes[].comment、codes[].comment，在本副本中一律不保留（本文件根键仅 summary/fields/indexes/codes，summary[0] 仅 name/title/primary/catelog/comment，fields[] 仅 name/title/type/length/dec/nullable/default，indexes[] 仅 name/idxtype/idxfields，codes[] 仅 field/table/valuefield/textfield/cond）；这些物理实现细节以真源与其 DDL 派生源为准。"
|
||||
"comment": "【本表定位】world_sync 侧消费方契约副本：运行时事件流（append-only，禁 UPDATE/DELETE；生产库由权威模块 pbl_runtime_ext 的存储过程 pbl_assert_append_only + BEFORE UPDATE/BEFORE DELETE 触发器强制，本副本只声明列契约、不复制那套物理实现）。【结构权威源 · PM 裁决】PM 裁决（任务 u0EO4x3ajntR41acaHj5H）选 B：以 modules/pbl_runtime_ext/models/pbl_runtime_event.json 为唯一真源（M11b-1 已批准产物，本任务禁止反向修改）；建表 DDL 派生源为该模块 scripts/pbl_runtime_event_ddl.py 与 scripts/migrations/*.sql。本文件是按真源逐字段内联的 world_sync 侧副本（25 列全量抄录），不是引用式 schema —— database-table-definition-spec 只承认 summary/fields/indexes/codes 四段式，scripts/validate_models_json.py 对 key=schema 引用式写法直接判 FAIL（QC #5），故 QC 要求内联。两份定义的一致性由 scripts/m11b2b_column_diff.py（代码列名 vs 权威列集双向 diff）与 scripts/m11b2b_selftest.py（四段式/抽象类型/主键/索引/编码 6 项断言 + 真源镜像 sha256）机械核对，真源列集变更时必须同步本副本。【id 生成方 · 单一口径，QC #4】本副本声明 id 由应用层生成：world_sync/pbl_runtime_tx.py::new_event_uid()（uuid4().hex，32 位小写十六进制），写路径 INSERT 列集（= 本文件 fields[] 全列，见 pbl_runtime_sql.py::EVENT_COLUMNS 由 fields[] 机械导出）显式提交 id，不依赖任何数据库 AUTO_INCREMENT。真源侧 id 为 type=long + auto_increment（BIGINT AUTO_INCREMENT），属其私有物理实现，本副本按 spec 归一为 str(32) 且不保留 auto_increment 键；两者差异已在交付件正文登记为 DEVIATION(WS-2) 并冒泡 agent.pm 裁决生产库列型收敛方案（改 VARCHAR(32) 或统一走 event_uid 定位），本副本不再同时声称两种生成方。对外幂等与跨系统追溯一律使用 event_uid（str 64，uk_event_uid），不依赖 id 的物理类型。【复合主键偏离 · 已登记，QC #4】summary[0].primary=[\"id\",\"created_at\"] 是对 spec「单键表恒为 [\\\"id\\\"]」的有意偏离：本表按 created_at 做按月 RANGE COLUMNS 分区，MySQL/MariaDB 要求分区表所有唯一键（含主键）必须包含分区列，否则建表报错 1503。真源依据行号：modules/pbl_runtime_ext/models/pbl_runtime_event.json:12（primary=[\"id\",\"created_at\"]）、同文件:6（comment 中 1503 约束说明与分区保留 14 个月）、同文件:40（fields[0].id type=long + auto_increment）。本副本不写 partition/guards/engine 等私有键，但主键必须与真源一致，否则 uk_event_uid/uk_tenant_idem（含 created_at）无法在本副本声明的 PK 下成立。该偏离已在 M11b-2b-B 交付件正文登记（DEVIATION(WS-1)），未写进校验器 COMPLIANCE_DEVIATIONS（子任务 A 最终登记表为空 {}，且本任务禁止修改校验器）。【私有键剥离】真源使用的 spec 未定义顶层扩展键 module/engine/charset/collate/tenant_scoped/append_only/write_paths/guards/partition/ddl_source，以及 fields[].ddl_type/comment/auto_increment、indexes[].comment、codes[].comment，本副本一律不保留：根键仅 summary/fields/indexes/codes，summary[0] 仅 name/title/primary/catelog/comment，fields[] 仅 name/title/type/length/dec/nullable/default，indexes[] 仅 name/idxtype/idxfields（主键不写进 indexes，由 summary[0].primary 表达），codes[] 仅 field/table/valuefield/textfield/cond。物理细节（分区、触发器、字符集、存储过程）以真源与其 DDL 派生源为准。【消费方清单】modules/world_sync/world_sync/pbl_runtime_sql.py（EVENT_TABLE/EVENT_COLUMNS，由本文件 fields[] 机械导出）、modules/world_sync/world_sync/pbl_runtime_tx.py::write_event_with_state（单事务事件+状态写入）、modules/world_sync/world_sync/pbl_runtime_authority.py（列契约解析层，已适配本内联四段式副本）、modules/world_sync/world_sync/init.py（事件写入门面）。上述代码引用的列名必须全部出现在本文件 fields[] 中，核对证据见 M11b-2b-B 交付件「字段交叉核对清单」。【打包镜像】apps/scense/pkgs/world_sync/models/pbl_runtime_event.json 是本文件的打包镜像，只由 modules/world_sync/scripts/sync_models_mirror.py 生成，禁止手工双写；一致性由 scripts/m11b2b_selftest.py 的 sha256 断言守护。"
|
||||
}
|
||||
],
|
||||
"fields": [
|
||||
|
||||
@ -1,25 +1,29 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""pbl_runtime_authority —— M11b-2b：``key=schema`` 引用的解析层（列名唯一权威源）。
|
||||
"""pbl_runtime_authority —— M11b-2b：表定义 JSON 的列契约解析层（列名唯一入口）。
|
||||
|
||||
**为什么存在**
|
||||
PM 裁决（任务 ``u0EO4x3ajntR41acaHj5H``，方案 B）：``pbl_runtime_event`` /
|
||||
``pbl_entity_state`` 两张表的**唯一真源**是 ``modules/pbl_runtime_ext/models/*.json``
|
||||
（M11b-1 已批准产物）。world_sync 侧 ``models/*.json`` 只保留 ``key=schema`` 引用，
|
||||
**不再自带** summary/fields/indexes/codes 结构体。
|
||||
**结构现状（M11b-2b-B 起，QC #1/#2/#3 整改）**
|
||||
PM 裁决 B：``pbl_runtime_event`` / ``pbl_entity_state`` 的**结构权威源**是
|
||||
``modules/pbl_runtime_ext/models/*.json``（M11b-1 已批准产物，禁止反向修改）。
|
||||
但 QC 明确要求 world_sync 侧 ``models/*.json`` 必须是 **database-table-definition-spec
|
||||
四段式（summary/fields/indexes/codes）内联副本**，而不是 ``key=schema`` 引用式
|
||||
——校验器 ``scripts/validate_models_json.py`` 对引用式写法直接判 FAIL。
|
||||
因此本模块现在解析的是「按真源逐字段内联、并按 spec 归一（抽象类型、去私有键）的
|
||||
world_sync 侧副本」，它同时就是 world_sync 消费方的列契约（列名与真源一致）。
|
||||
|
||||
于是「消费方列契约」必须从权威定义**机械导出**，而不是人工抄一遍——人工双写正是
|
||||
QC #1/#3 判定的「多份互相矛盾定义」的根因。本模块就是那个导出层：
|
||||
本模块是列契约的**机械导出层**，不是人工抄写层：
|
||||
|
||||
* :func:`load_reference` —— 读本模块 ``models/<table>.json``（key=schema 引用）；
|
||||
* :func:`authority_model` —— 顺着引用解析到 pbl_runtime_ext 的真源 JSON；
|
||||
* :func:`load_model` —— 读本模块 ``models/<table>.json``（四段式）并校验结构；
|
||||
* :func:`columns` / :func:`insert_columns` / :func:`primary` / :func:`indexes`
|
||||
—— 暴露权威列集（``insert_columns`` 自动剔除 auto_increment 列）；
|
||||
* :func:`ddl` —— 由权威列集渲染本地建表语句（仅供测试/本地起库；生产 DDL、
|
||||
分区与 append-only 触发器归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py``
|
||||
与 ``scripts/migrations/*.sql`` 所有，本模块不复制那套逻辑）。
|
||||
—— 暴露列契约（``insert_columns`` 自动剔除自增列，兼容真源私有 ``auto_increment`` 键）；
|
||||
* :func:`authority_path` —— 反查结构权威源（pbl_runtime_ext）在磁盘上的真实路径，
|
||||
供审计与 QC 取证（写入结果的 ``authority`` 字段）；
|
||||
* :func:`ddl` —— 由列契约渲染本地建表语句（仅供测试/本地起库；生产 DDL、分区与
|
||||
append-only 触发器归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py``
|
||||
与 ``scripts/migrations/*.sql`` 所有，本模块不复制那套逻辑）；
|
||||
* :func:`column_diff` —— 消费方用到的列 vs 列契约的双向 diff（fail-closed 的判据）。
|
||||
|
||||
**fail-closed**：引用缺失、真源找不到、真源不是四段式、列集为空 —— 一律抛
|
||||
:class:`AuthorityResolutionError`，绝不退化成「猜列名」。
|
||||
**fail-closed**：文件缺失、JSON 不可解析、不是四段式、fields 为空 —— 一律抛
|
||||
:class:`AuthorityResolutionError`，绝不退化成「猜列名」或回落到旧的手写常量。
|
||||
"""
|
||||
|
||||
import json
|
||||
@ -27,10 +31,13 @@ import os
|
||||
|
||||
__all__ = [
|
||||
"AuthorityResolutionError",
|
||||
"REFERENCE_ROOT_KEYS",
|
||||
"REFERENCE_SCHEMA_KEYS",
|
||||
"SPEC_ROOT_KEYS",
|
||||
"load_reference",
|
||||
"SPEC_SUMMARY_KEYS",
|
||||
"SPEC_FIELD_KEYS",
|
||||
"SPEC_INDEX_KEYS",
|
||||
"SPEC_CODE_KEYS",
|
||||
"AUTHORITY_MODULE",
|
||||
"load_model",
|
||||
"authority_path",
|
||||
"authority_model",
|
||||
"columns",
|
||||
@ -44,15 +51,16 @@ __all__ = [
|
||||
"column_diff",
|
||||
]
|
||||
|
||||
# models/<table>.json 是 key=schema 引用：根键只允许这两个
|
||||
REFERENCE_ROOT_KEYS = ("key", "schema")
|
||||
# 引用体内允许的元信息键（白名单，自测脚本按此做机械断言）
|
||||
REFERENCE_SCHEMA_KEYS = (
|
||||
"table", "authority_module", "authority_path", "owner", "milestone",
|
||||
"decision_ref", "decision_basis", "structure_note", "consumers", "mirror_note",
|
||||
)
|
||||
# 真源（表定义）必须是四段式
|
||||
# world_sync 侧 models/<table>.json 必须是 spec 四段式（database-table-definition-spec）
|
||||
SPEC_ROOT_KEYS = ("summary", "fields", "indexes", "codes")
|
||||
# 各段允许的键（白名单，自测脚本按此做机械断言；comment 为 spec 允许的可选说明键）
|
||||
SPEC_SUMMARY_KEYS = ("name", "title", "primary", "catelog", "comment")
|
||||
SPEC_FIELD_KEYS = ("name", "title", "type", "length", "dec", "nullable", "default")
|
||||
SPEC_INDEX_KEYS = ("name", "idxtype", "idxfields")
|
||||
SPEC_CODE_KEYS = ("field", "table", "valuefield", "textfield", "cond")
|
||||
|
||||
# 结构权威源所在模块（PM 裁决 B；本模块只反查路径，不消费其私有键）
|
||||
AUTHORITY_MODULE = "pbl_runtime_ext"
|
||||
|
||||
MODULE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
MODELS_DIR = os.path.join(MODULE_ROOT, "models")
|
||||
@ -77,7 +85,7 @@ _MYSQL_TYPES = {
|
||||
|
||||
|
||||
class AuthorityResolutionError(Exception):
|
||||
"""引用/真源解析失败（fail-closed，调用方不得吞掉）。"""
|
||||
"""模型缺失/不可解析/非四段式（fail-closed，调用方不得吞掉）。"""
|
||||
|
||||
def __init__(self, message, **details):
|
||||
Exception.__init__(self, message)
|
||||
@ -91,61 +99,83 @@ class AuthorityResolutionError(Exception):
|
||||
return "%s (%s)" % (self.message, parts)
|
||||
|
||||
|
||||
def _reference_dir():
|
||||
def _models_dir():
|
||||
return MODELS_DIR
|
||||
|
||||
|
||||
def load_reference(table):
|
||||
"""读取本模块 ``models/<table>.json``，校验它是合法的 key=schema 引用。"""
|
||||
if not table or not isinstance(table, str):
|
||||
raise AuthorityResolutionError("table name is required", table=table)
|
||||
path = os.path.join(_reference_dir(), table + ".json")
|
||||
def _read_json(path):
|
||||
if not os.path.isfile(path):
|
||||
raise AuthorityResolutionError("schema reference missing", path=path)
|
||||
raise AuthorityResolutionError("model file missing", path=path)
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
ref = json.load(fh)
|
||||
data = json.load(fh)
|
||||
except ValueError as exc:
|
||||
raise AuthorityResolutionError("schema reference not valid JSON: %s" % exc,
|
||||
path=path)
|
||||
if not isinstance(ref, dict) or ref.get("key") != "schema":
|
||||
raise AuthorityResolutionError("model not valid JSON: %s" % exc, path=path)
|
||||
if not isinstance(data, dict):
|
||||
raise AuthorityResolutionError("model must be a JSON object", path=path)
|
||||
return data
|
||||
|
||||
|
||||
_MODEL_CACHE = {}
|
||||
|
||||
|
||||
def load_model(table):
|
||||
"""读取并校验本模块 ``models/<table>.json``（spec 四段式内联副本）。
|
||||
|
||||
校验项（fail-closed）：根键含 summary/fields/indexes/codes；summary 恰好 1 条；
|
||||
fields 非空且每列有 name。列集与真源的一致性由 scripts/m11b2b_column_diff.py
|
||||
与 scripts/m11b2b_selftest.py 机械核对，不在运行期重复解析真源私有键。
|
||||
"""
|
||||
if not table or not isinstance(table, str):
|
||||
raise AuthorityResolutionError("table name is required", table=table)
|
||||
if table in _MODEL_CACHE:
|
||||
return _MODEL_CACHE[table]
|
||||
path = os.path.join(_models_dir(), table + ".json")
|
||||
model = _read_json(path)
|
||||
missing = [k for k in SPEC_ROOT_KEYS if k not in model]
|
||||
if missing:
|
||||
raise AuthorityResolutionError(
|
||||
"models/%s.json must be a key=schema reference (no local structure)" % table,
|
||||
path=path, root_keys=sorted(ref) if isinstance(ref, dict) else type(ref).__name__)
|
||||
extra = set(ref) - set(REFERENCE_ROOT_KEYS)
|
||||
if extra:
|
||||
raise AuthorityResolutionError("reference carries non-whitelist root keys",
|
||||
path=path, extra=sorted(extra))
|
||||
schema = ref.get("schema")
|
||||
if not isinstance(schema, dict):
|
||||
raise AuthorityResolutionError("reference.schema must be an object", path=path)
|
||||
if schema.get("table") != table:
|
||||
raise AuthorityResolutionError("reference.schema.table mismatches filename",
|
||||
path=path, declared=schema.get("table"))
|
||||
bad = set(schema) - set(REFERENCE_SCHEMA_KEYS)
|
||||
if bad:
|
||||
raise AuthorityResolutionError("reference.schema carries non-whitelist keys",
|
||||
path=path, extra=sorted(bad))
|
||||
return ref
|
||||
"model is not database-table-definition-spec four-section",
|
||||
path=path, missing=missing,
|
||||
hint="world_sync 侧必须是内联四段式副本（PM 裁决 B + QC #1/#5），不得用 key=schema 引用式")
|
||||
summary = model.get("summary")
|
||||
if not isinstance(summary, list) or len(summary) != 1 or not isinstance(summary[0], dict):
|
||||
raise AuthorityResolutionError("summary must be an array with exactly 1 record",
|
||||
path=path, got=type(summary).__name__)
|
||||
if not str(summary[0].get("name") or "").strip():
|
||||
raise AuthorityResolutionError("summary[0].name is required", path=path)
|
||||
fields = model.get("fields")
|
||||
if not isinstance(fields, list) or not fields:
|
||||
raise AuthorityResolutionError("model has no fields", path=path)
|
||||
for f in fields:
|
||||
if not isinstance(f, dict) or not str(f.get("name") or "").strip():
|
||||
raise AuthorityResolutionError("field without name", path=path, field=f)
|
||||
_MODEL_CACHE[table] = model
|
||||
return model
|
||||
|
||||
|
||||
# 兼容旧调用名：本模块过去暴露 authority_model（解析引用指向的真源），现在列契约
|
||||
# 就是内联副本本身，语义等价。
|
||||
def authority_model(table):
|
||||
"""列契约模型（= world_sync 侧内联四段式副本，逐字段抄自 pbl_runtime_ext 真源）。"""
|
||||
return load_model(table)
|
||||
|
||||
|
||||
def authority_path(table):
|
||||
"""把引用解析成磁盘上真实存在的真源路径（多候选，全部落空即报错）。"""
|
||||
ref = load_reference(table)
|
||||
schema = ref["schema"]
|
||||
mod = schema.get("authority_module")
|
||||
candidates = []
|
||||
declared = schema.get("authority_path")
|
||||
if declared:
|
||||
candidates.append(os.path.normpath(os.path.join(WORKSPACE_ROOT, declared)))
|
||||
candidates.append(os.path.normpath(os.path.join(MODULE_ROOT, declared)))
|
||||
if mod:
|
||||
candidates.append(os.path.join(WORKSPACE_ROOT, "modules", mod, "models",
|
||||
table + ".json"))
|
||||
candidates.append(os.path.join(WORKSPACE_ROOT, "apps", "scense", "pkgs", mod,
|
||||
"models", table + ".json"))
|
||||
candidates.append(os.path.normpath(os.path.join(MODULE_ROOT, os.pardir, mod,
|
||||
"models", table + ".json")))
|
||||
"""反查结构权威源（pbl_runtime_ext）在磁盘上的真实路径，供审计取证。
|
||||
|
||||
只用于写入结果里的 ``authority`` 字段与 QC 核对，**不参与列集解析**（列集来自
|
||||
本模块内联副本）。找不到真源不抛错，返回声明式相对路径（真源缺失属部署期问题，
|
||||
不应让运行时写入失败）。
|
||||
"""
|
||||
declared = os.path.join("modules", AUTHORITY_MODULE, "models", table + ".json")
|
||||
candidates = [
|
||||
os.path.normpath(os.path.join(WORKSPACE_ROOT, declared)),
|
||||
os.path.normpath(os.path.join(MODULE_ROOT, os.pardir, AUTHORITY_MODULE,
|
||||
"models", table + ".json")),
|
||||
os.path.join(WORKSPACE_ROOT, "apps", "scense", "pkgs", AUTHORITY_MODULE,
|
||||
"models", table + ".json"),
|
||||
]
|
||||
tried = []
|
||||
for cand in candidates:
|
||||
if cand in tried:
|
||||
@ -153,55 +183,28 @@ def authority_path(table):
|
||||
tried.append(cand)
|
||||
if os.path.isfile(cand):
|
||||
return cand
|
||||
raise AuthorityResolutionError("authority model not found for table",
|
||||
table=table, tried=tried)
|
||||
|
||||
|
||||
_MODEL_CACHE = {}
|
||||
|
||||
|
||||
def authority_model(table):
|
||||
"""读取并校验真源 JSON（四段式 + fields 非空），带缓存。"""
|
||||
if table in _MODEL_CACHE:
|
||||
return _MODEL_CACHE[table]
|
||||
path = authority_path(table)
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as fh:
|
||||
model = json.load(fh)
|
||||
except ValueError as exc:
|
||||
raise AuthorityResolutionError("authority model not valid JSON: %s" % exc,
|
||||
path=path)
|
||||
if not isinstance(model, dict):
|
||||
raise AuthorityResolutionError("authority model must be an object", path=path)
|
||||
missing = set(SPEC_ROOT_KEYS) - set(model)
|
||||
if missing:
|
||||
raise AuthorityResolutionError("authority model is not four-section",
|
||||
path=path, missing=sorted(missing))
|
||||
fields = model.get("fields")
|
||||
if not isinstance(fields, list) or not fields:
|
||||
raise AuthorityResolutionError("authority model has no fields", path=path)
|
||||
_MODEL_CACHE[table] = model
|
||||
return model
|
||||
return declared
|
||||
|
||||
|
||||
def field_map(table):
|
||||
"""``{列名: 列定义 dict}``（保持真源顺序）。"""
|
||||
"""``{列名: 列定义 dict}``（保持模型顺序）。"""
|
||||
out = {}
|
||||
for f in authority_model(table)["fields"]:
|
||||
name = f.get("name")
|
||||
if not name:
|
||||
raise AuthorityResolutionError("authority field without name", table=table)
|
||||
out[name] = f
|
||||
for f in load_model(table)["fields"]:
|
||||
out[f["name"]] = f
|
||||
return out
|
||||
|
||||
|
||||
def columns(table):
|
||||
"""权威全列集（有序元组）。"""
|
||||
"""权威全列集（有序元组）—— 消费方 SQL 里允许出现的列名以此为唯一白名单。"""
|
||||
return tuple(field_map(table).keys())
|
||||
|
||||
|
||||
def auto_increment_columns(table):
|
||||
"""由数据库自增生成的列（INSERT 时不得出现）。"""
|
||||
"""由数据库自增生成的列（INSERT 时不得出现）。
|
||||
|
||||
内联副本按 spec 剥离了 ``auto_increment`` 私有键，故通常为空集；保留该逻辑是为了
|
||||
当列契约指向真源（其 id 带 auto_increment）时仍能正确剔除该列。
|
||||
"""
|
||||
return tuple(n for n, f in field_map(table).items()
|
||||
if f.get("auto_increment") or str(f.get("default", "")).upper()
|
||||
in ("AUTO_INCREMENT", "IDENTITY"))
|
||||
@ -214,29 +217,38 @@ def insert_columns(table):
|
||||
|
||||
|
||||
def primary(table):
|
||||
summary = authority_model(table).get("summary") or []
|
||||
if not summary or not isinstance(summary[0], dict):
|
||||
raise AuthorityResolutionError("authority model has no summary[0]", table=table)
|
||||
"""summary[0].primary（非空数组）。主键不写进 indexes（spec 约定）。"""
|
||||
summary = load_model(table).get("summary") or [{}]
|
||||
pk = summary[0].get("primary") or summary[0].get("key")
|
||||
if isinstance(pk, str):
|
||||
pk = [p.strip() for p in pk.split(",") if p.strip()]
|
||||
return tuple(pk or ())
|
||||
if not isinstance(pk, list) or not pk:
|
||||
raise AuthorityResolutionError("model summary[0].primary must be a non-empty array",
|
||||
table=table)
|
||||
cols = set(columns(table))
|
||||
unknown = [p for p in pk if p not in cols]
|
||||
if unknown:
|
||||
raise AuthorityResolutionError("primary references unknown column",
|
||||
table=table, unknown=unknown)
|
||||
return tuple(pk)
|
||||
|
||||
|
||||
def indexes(table):
|
||||
"""规范化索引：``[{'name','idxtype','idxfields'}]``（兼容 unique/fields 写法）。"""
|
||||
"""二级索引/唯一键列表：``[{name, idxtype, idxfields}]``（兼容真源 unique/fields 写法）。"""
|
||||
out = []
|
||||
for idx in authority_model(table).get("indexes") or []:
|
||||
for idx in (load_model(table).get("indexes") or []):
|
||||
if not isinstance(idx, dict):
|
||||
raise AuthorityResolutionError("index must be an object", table=table, index=idx)
|
||||
fields = idx.get("idxfields") or idx.get("fields") or []
|
||||
if isinstance(fields, str):
|
||||
fields = [f.strip() for f in fields.split(",") if f.strip()]
|
||||
kind = idx.get("idxtype") or ("unique" if idx.get("unique") else "index")
|
||||
out.append({"name": idx.get("name"), "idxtype": kind, "idxfields": tuple(fields)})
|
||||
idxtype = idx.get("idxtype") or ("unique" if idx.get("unique") else "index")
|
||||
out.append({"name": idx.get("name"), "idxtype": idxtype, "idxfields": list(fields)})
|
||||
return out
|
||||
|
||||
|
||||
def table_title(table):
|
||||
summary = authority_model(table).get("summary") or [{}]
|
||||
summary = load_model(table).get("summary") or [{}]
|
||||
return summary[0].get("title") or summary[0].get("name") or table
|
||||
|
||||
|
||||
@ -264,21 +276,19 @@ def _nullable(f):
|
||||
|
||||
|
||||
def ddl(table, dialect="sqlite"):
|
||||
"""由权威列集渲染 CREATE TABLE（测试/本地起库用）。
|
||||
"""由列契约渲染 CREATE TABLE（测试/本地起库用）。
|
||||
|
||||
生产库的 DDL（分区、append-only 触发器、存储过程）归 pbl_runtime_ext 所有，
|
||||
本函数**只保证列集与主键/唯一键一致**，不复制分区与触发器逻辑。
|
||||
"""
|
||||
if dialect not in ("sqlite", "mysql"):
|
||||
raise AuthorityResolutionError("unsupported ddl dialect", dialect=dialect)
|
||||
model = authority_model(table)
|
||||
model = load_model(table)
|
||||
gen = set(auto_increment_columns(table))
|
||||
lines = []
|
||||
for f in model["fields"]:
|
||||
col = "`%s`" % f["name"] if dialect == "mysql" else f["name"]
|
||||
col = ("`%s`" % f["name"]) if dialect == "mysql" else f["name"]
|
||||
piece = [" %s %s" % (col, _render_type(f, dialect))]
|
||||
gen = bool(f.get("auto_increment"))
|
||||
if dialect == "mysql" and gen:
|
||||
piece.append(" AUTO_INCREMENT")
|
||||
if not _nullable(f) and not gen:
|
||||
piece.append(" NOT NULL")
|
||||
default = f.get("default")
|
||||
|
||||
@ -2,8 +2,15 @@
|
||||
"""M11b-2 运行时单事务写入 —— SQL 方言层（纯 SQL 文本 + 占位符，不绑定任何 ORM/驱动）。
|
||||
|
||||
职责边界：本模块只负责「把 SQL 写对」——方言识别、占位符风格、READ COMMITTED 设置语句、
|
||||
悲观锁子句、以及两张表的建表 DDL（与 M11b-1 的 pbl_runtime_event schema 对齐）。
|
||||
真正的事务编排在 :mod:`world_sync.pbl_runtime_tx`。
|
||||
悲观锁子句、以及两张表的建表 DDL。真正的事务编排在 :mod:`world_sync.pbl_runtime_tx`。
|
||||
|
||||
**列集单一真源（M11b-2b-B 起，QC #3 整改）**
|
||||
``EVENT_COLUMNS`` / ``STATE_COLUMNS`` 不再手写常量，而是从
|
||||
``modules/world_sync/models/pbl_runtime_event.json`` / ``pbl_entity_state.json``
|
||||
（spec 四段式内联副本，结构权威源 = pbl_runtime_ext，PM 裁决 B）的 ``fields[]``
|
||||
**机械导出**。旧版在本文件里手抄的 ``event_id`` / ``world_id``(state 表) / ``state``
|
||||
/ ``updated_by_event`` 等列名在真源中不存在，构成「代码一套、JSON 另一套」的双写分叉，
|
||||
是 QC #3 判定的消费方契约断裂根因；现在这类漂移在结构上不可能再发生。
|
||||
|
||||
设计要点：
|
||||
1. 用 DB-API 2.0 规范接口（``paramstyle`` / ``execute`` / ``rowcount`` / ``commit`` /
|
||||
@ -12,11 +19,15 @@
|
||||
``pyformat`` → ``%(name)s``），避免在 MySQL 上写出 sqlite 的 ``?``；
|
||||
3. ``SELECT ... FOR UPDATE`` 只有 MySQL/PG 支持，sqlite 侧返回空串（sqlite 写锁由事务本身保证），
|
||||
并发正确性在两条路径上都由「乐观锁 state_version」兜底；
|
||||
4. 不引入任何广播 / 轮询 / 时延优化逻辑（明确排除在 M11b-2 范围之外）。
|
||||
4. 不引入任何广播 / 轮询 / 时延优化逻辑（明确排除在 M11b-2 范围之外）；
|
||||
5. 建表 DDL 由 :func:`world_sync.pbl_runtime_authority.ddl` 按同一份模型渲染，
|
||||
本模块不再自带第二套列定义（生产库的分区/触发器 DDL 归 pbl_runtime_ext 所有）。
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
|
||||
from . import pbl_runtime_authority as auth
|
||||
|
||||
__all__ = [
|
||||
"EVENT_TABLE",
|
||||
"STATE_TABLE",
|
||||
@ -34,6 +45,7 @@ __all__ = [
|
||||
"ddl_for",
|
||||
"EVENT_COLUMNS",
|
||||
"STATE_COLUMNS",
|
||||
"columns_of",
|
||||
"percentile",
|
||||
]
|
||||
|
||||
@ -44,30 +56,24 @@ 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 columns_of(table):
|
||||
"""取某表在模型 JSON（spec 四段式内联副本）里声明的全列集（有序）。
|
||||
|
||||
fail-closed：模型缺失 / 不是四段式 / fields 为空，一律由
|
||||
:mod:`world_sync.pbl_runtime_authority` 抛
|
||||
:class:`~world_sync.pbl_runtime_authority.AuthorityResolutionError`，不退回旧常量。
|
||||
"""
|
||||
return auth.columns(table)
|
||||
|
||||
|
||||
# append-only 事件流列集（= models/pbl_runtime_event.json fields[] 顺序，25 列，含应用层
|
||||
# 生成的 id；id 生成方见 world_sync/pbl_runtime_tx.py::new_event_uid 与该文件 summary[0].comment）
|
||||
EVENT_COLUMNS = columns_of(EVENT_TABLE)
|
||||
|
||||
# 实体状态列集（= models/pbl_entity_state.json fields[] 顺序，10 列；定位键是
|
||||
# tenant_id + session_id + entity_id，对应唯一键 uk_es，真源无 world_id / state 列）
|
||||
STATE_COLUMNS = columns_of(STATE_TABLE)
|
||||
|
||||
|
||||
def detect_dialect(conn_or_module):
|
||||
@ -162,80 +168,18 @@ def supports_transaction_control(dialect):
|
||||
|
||||
|
||||
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,
|
||||
]
|
||||
"""sqlite 建表语句列表（单测 / 本地环境用），由模型 JSON 渲染，无第二套列定义。"""
|
||||
return [auth.ddl(EVENT_TABLE, DIALECT_SQLITE), auth.ddl(STATE_TABLE, DIALECT_SQLITE)]
|
||||
|
||||
|
||||
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,
|
||||
]
|
||||
"""MySQL 建表语句列表（本地/预发用，InnoDB + utf8mb4），同样由模型 JSON 渲染。
|
||||
|
||||
生产库真正的 ``pbl_runtime_event`` DDL（按月 RANGE COLUMNS 分区、append-only 存储过程
|
||||
与触发器）归 pbl_runtime_ext 的 ``scripts/pbl_runtime_event_ddl.py`` 与
|
||||
``scripts/migrations/*.sql`` 所有，本函数不复制那套物理细节。
|
||||
"""
|
||||
return [auth.ddl(EVENT_TABLE, DIALECT_MYSQL), auth.ddl(STATE_TABLE, DIALECT_MYSQL)]
|
||||
|
||||
|
||||
#: 各方言建表语句集合，供宿主初始化 / 单测建库使用。
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user