aed40d3491
- QC #17: SLO_TOLERANCE_MS=500.0 由 scripts/m11b4_selftest.py 本地定义上提为 world_sync/pbl_runtime_polling.py 模块常量（与 FALLBACK_SLO_MS 同段）， selftest 改为 import 引用；按三处同步补 __init__.py 导入+__all__ 与 init.py env.SLO_TOLERANCE_MS / env.SEEN_IDS_IN_SQL 登记 - QC #15/#16: selftest AC5.2 断言改为核验 poll_since 真实返回键 rows/count （recovered_count 属 poll_once/run 口径），并显式断言不含 recovered_count - 重跑全量 selftest（真实 time.sleep(3.0)，无打桩）：RESULT: PASS FAILURES=0，31/31； AC5.1 缺失名 -> []；AC1.4 -> 3000.5ms <= 3500.0ms - 重新落盘 scripts/m11b4_selftest_output.txt 与 scripts/m11b4_latency_report.json （run_at=2026-09-22 13:56:47 +0800，samples=200，含 segment_p99_ms/broadcast_stats/verdict） - skill/SKILL.md 补两条陷阱：SLO_TOLERANCE_MS 属模块源码常量禁止测试侧重定义； poll_since 返回键为 count 不改 M11b-1/2/3 核心逻辑（write_event_with_state / emit_post_commit_broadcast / 表结构索引）
641 lines
25 KiB
Python
641 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""world_sync 宿主挂载层（module-development-spec：init.py 三处同步注册）。
|
||
|
||
职责：
|
||
1. world_sync 业务表 CRUD（list/get/create/update/delete/execute）；
|
||
2. M11b-1 表定义注册（pbl_runtime_event / pbl_entity_state schema 元数据，零写入）；
|
||
3. M11b-2 单事务「事件 + 实体状态」原子写入契约登记到 ServerEnv
|
||
（``env.write_event_with_state`` / ``env.read_entity_state`` / ``env.pbl_runtime_conn_factory``）。
|
||
|
||
铁律：
|
||
- 不硬编码库名：取库名走 ``ServerEnv().get_module_dbname('world_sync')``；
|
||
- 导入期不建立任何数据库连接（连接工厂缺失时由写入调用点 fail-closed 抛异常）；
|
||
- 不吞错误：本文件所有 import 均为**硬依赖**，任一失败即抛（QC 退回意见 #1/#2）。
|
||
|
||
M11b-2a 迁移修正（响应 QC #1）：本文件此前导入的
|
||
``RuntimeWriteError/InvalidWriteRequest/ConnectionUnavailable/TransactionAborted`` 与
|
||
``RuntimeStateEventWriter/build_plan/configure_writer/get_writer/write_event_with_state_async``
|
||
以及 ``pbl_runtime_sql.MYSQL/SQLITE`` 在真源模块中并不存在（照抄 apps 侧旧 API 草案），
|
||
现改为只导入迁移后的真源实际存在的名字；旧语义名以「同一类对象的别名」形式保留
|
||
（见 pbl_runtime_errors.py 末尾别名段），不存在双真源。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
|
||
# 宿主基础包（appPublic / sqlor / ahserver）：M11b-2a 起改为「容错导入 + 显式记录」——
|
||
# 目的是让「三处接线」自检（import 本包 + 调 load_world_sync()）不依赖宿主部署环境，
|
||
# 而不是吞错误：基础包缺失时记入 HOST_DEPS_ERROR，任何真正访问数据库的函数在调用点
|
||
# 通过 _require_host_deps() fail-fast 抛出带原因的 RuntimeError（不静默返回 None）。
|
||
HOST_DEPS_ERROR = None
|
||
try:
|
||
from appPublic.uniqueID import getID
|
||
from appPublic.timeUtils import curDateString, timestampstr
|
||
from sqlor.dbpools import DBPools
|
||
from ahserver.serverenv import ServerEnv
|
||
HOST_DEPS_AVAILABLE = True
|
||
except ImportError as _exc: # pragma: no cover - 仅在无宿主基础包的开发/自检环境触发
|
||
HOST_DEPS_AVAILABLE = False
|
||
HOST_DEPS_ERROR = _exc
|
||
getID = curDateString = timestampstr = None
|
||
DBPools = ServerEnv = None
|
||
|
||
|
||
class HostDepsNotAvailable(RuntimeError):
|
||
"""宿主基础包（appPublic/sqlor/ahserver）不可用时，访问数据库的入口 fail-fast。"""
|
||
|
||
def __init__(self, op):
|
||
super().__init__(
|
||
'%s requires host foundation packages (appPublic/sqlor/ahserver); '
|
||
'import error was: %r' % (op, HOST_DEPS_ERROR))
|
||
self.op = op
|
||
|
||
|
||
def _require_host_deps(op):
|
||
if not HOST_DEPS_AVAILABLE:
|
||
raise HostDepsNotAvailable(op)
|
||
return True
|
||
|
||
|
||
class _StandaloneEnv(object):
|
||
"""无宿主 ServerEnv 时的登记载体（仅供接线自检，不建连、不接库）。"""
|
||
|
||
def __init__(self):
|
||
self._kv = {}
|
||
|
||
def __setattr__(self, k, v):
|
||
object.__setattr__(self, k, v)
|
||
|
||
def set(self, k, v):
|
||
self._kv[k] = v
|
||
|
||
def get(self, k, default=None):
|
||
return getattr(self, k, self._kv.get(k, default))
|
||
|
||
def get_module_dbname(self, m):
|
||
raise HostDepsNotAvailable('get_module_dbname(%s)' % m)
|
||
|
||
|
||
def _resolve_env(server_env=None):
|
||
"""解析登记目标 ServerEnv：显式注入优先，其次宿主全局单例，最后 standalone 载体。"""
|
||
if server_env is not None:
|
||
return server_env
|
||
if ServerEnv is not None:
|
||
try:
|
||
return ServerEnv()
|
||
except Exception as exc: # 宿主单例不可用：记录后降级，不中断接线登记
|
||
standalone_env_error(str(exc))
|
||
return _StandaloneEnv()
|
||
|
||
|
||
def standalone_env_error(message):
|
||
"""记录 ServerEnv 降级原因（幂等，不抛错）。"""
|
||
global STANDALONE_ENV_NOTICES
|
||
STANDALONE_ENV_NOTICES = list(STANDALONE_ENV_NOTICES) + [message]
|
||
return message
|
||
|
||
|
||
STANDALONE_ENV_NOTICES = []
|
||
|
||
# --- M11b-2: 单事务写入（事件 + 实体状态）注册接线（真源实际存在的名字） ---
|
||
from .pbl_runtime_errors import (
|
||
ConcurrentStateConflict,
|
||
ConnFactoryNotConfigured,
|
||
ConnectionUnavailable,
|
||
EventWriteError,
|
||
InvalidWriteRequest,
|
||
PblRuntimeError,
|
||
RuntimeWriteError,
|
||
StateWriteError,
|
||
TransactionAborted,
|
||
TxAbortError,
|
||
TxConfigError,
|
||
)
|
||
from .pbl_runtime_sql import (
|
||
DIALECT_MYSQL,
|
||
DIALECT_SQLITE,
|
||
EVENT_TABLE,
|
||
STATE_TABLE,
|
||
ddl_for,
|
||
detect_dialect,
|
||
percentile,
|
||
)
|
||
from .pbl_runtime_tx import (
|
||
DEFAULT_EVENT_TYPE,
|
||
DEFAULT_SOURCE,
|
||
EVENT_REQUIRED,
|
||
STATE_KEY_COLUMNS,
|
||
new_event_id,
|
||
new_event_uid,
|
||
read_entity_state as read_entity_state_via_tx,
|
||
utc_now_text,
|
||
write_event_with_state as write_event_with_state_via_tx,
|
||
)
|
||
from .pbl_runtime_tx_env import (
|
||
SERVER_ENV_KEYS,
|
||
build_mysql_conn_factory,
|
||
build_sqlite_conn_factory,
|
||
load_db_config,
|
||
module_conf_dir,
|
||
resolve_conn_factory,
|
||
)
|
||
from .pbl_runtime_tx_sqlor import (
|
||
ENV_NAME_KEYS,
|
||
POOL_KEYS,
|
||
build_conn_factory_from_pools,
|
||
host_conn_factory,
|
||
read_entity_state,
|
||
write_event_with_state,
|
||
)
|
||
|
||
# --- M11b-3: 事务提交后广播（post-commit broadcast）注册接线 ---
|
||
from .pbl_runtime_broadcast import (
|
||
BROADCAST_CHANNEL,
|
||
BROADCAST_LATENCY_SLO_MS,
|
||
DEFAULT_EVENT_CHANNEL,
|
||
PAYLOAD_SUMMARY_MAX_CHARS,
|
||
BaseChannel,
|
||
Broadcaster,
|
||
ChannelUnavailable,
|
||
InProcessChannel,
|
||
NoopChannel,
|
||
RedisPubSubChannel,
|
||
WebhookChannel,
|
||
broadcast_latency_percentile_ms,
|
||
broadcast_last_message,
|
||
broadcast_stats,
|
||
build_broadcast_message,
|
||
configure_broadcast,
|
||
drain_broadcast,
|
||
emit_post_commit_broadcast,
|
||
get_broadcaster,
|
||
install_broadcast_hook,
|
||
is_broadcast_hook_installed,
|
||
reset_broadcaster,
|
||
shutdown_broadcast,
|
||
subscribe_broadcast,
|
||
unsubscribe_broadcast,
|
||
wrap_with_post_commit_broadcast,
|
||
)
|
||
|
||
# --- M11b-4: 3s 轮询兜底 + 端到端时延验证 注册接线 ---
|
||
from .pbl_runtime_polling import (
|
||
FALLBACK_SLO_MS,
|
||
POLL_INTERVAL_S,
|
||
POLL_LIMIT,
|
||
POLL_TIMEOUT_S,
|
||
SEEN_IDS_IN_SQL,
|
||
SLO_TOLERANCE_MS,
|
||
PollingFallback,
|
||
build_poll_sql,
|
||
configure_fallback,
|
||
fallback_stats,
|
||
get_fallback,
|
||
normalize_watermark,
|
||
plan_uses_scene_index,
|
||
explain_poll_query,
|
||
poll_since,
|
||
reset_fallback,
|
||
run_fallback,
|
||
wait_event,
|
||
)
|
||
from .pbl_runtime_latency import (
|
||
E2E_LATENCY_SLO_MS,
|
||
latency_report,
|
||
measure_once,
|
||
run_chain,
|
||
summarize_latency,
|
||
)
|
||
|
||
|
||
MODULE_NAME = 'world_sync'
|
||
MODULE_VERSION = '0.3.0-m11b3'
|
||
#: 宿主 ServerEnv 上登记 conn_factory 的键（与 pbl_runtime_tx_env.SERVER_ENV_KEYS 一致）
|
||
SERVER_ENV_CONN_FACTORY_KEYS = SERVER_ENV_KEYS
|
||
|
||
|
||
def _get_dbname():
|
||
_require_host_deps('_get_dbname')
|
||
return ServerEnv().get_module_dbname('world_sync')
|
||
|
||
|
||
def _clean(ns):
|
||
for k in list(ns.keys()):
|
||
if k.endswith('_text'):
|
||
ns.pop(k, None)
|
||
return ns
|
||
|
||
|
||
async def list_world_syncs(params=None):
|
||
_require_host_deps('list_world_syncs')
|
||
db = DBPools()
|
||
async with db.sqlorContext(_get_dbname()) as sor:
|
||
return await sor.sqlExe("select id, world_id, sync_type, source, status, result_json, sync_date, created_at from world_sync order by created_at desc", {})
|
||
|
||
|
||
async def get_world_sync(ns):
|
||
_require_host_deps('get_world_sync')
|
||
rec_id = (ns or {}).get('id')
|
||
if not rec_id:
|
||
return None
|
||
db = DBPools()
|
||
async with db.sqlorContext(_get_dbname()) as sor:
|
||
rows = await sor.sqlExe("select * from world_sync where id=${id}$", {'id': rec_id})
|
||
return rows[0] if rows else None
|
||
|
||
|
||
async def create_world_sync(ns):
|
||
_require_host_deps('create_world_sync')
|
||
ns = _clean(dict(ns))
|
||
ns['id'] = ns.get('id') or getID()
|
||
if not ns.get('sync_date'):
|
||
ns['sync_date'] = curDateString()
|
||
ns['created_at'] = curDateString()
|
||
if not ns.get('status'):
|
||
ns['status'] = '0'
|
||
db = DBPools()
|
||
async with db.sqlorContext(_get_dbname()) as sor:
|
||
await sor.C('world_sync', ns)
|
||
return {'success': True, 'id': ns['id']}
|
||
|
||
|
||
async def update_world_sync(ns):
|
||
_require_host_deps('update_world_sync')
|
||
ns = _clean(dict(ns))
|
||
rec_id = ns.get('id')
|
||
if not rec_id:
|
||
return {'success': False, 'message': '缺少 id'}
|
||
db = DBPools()
|
||
async with db.sqlorContext(_get_dbname()) as sor:
|
||
await sor.U('world_sync', ns)
|
||
return {'success': True}
|
||
|
||
|
||
async def delete_world_sync(ns):
|
||
_require_host_deps('delete_world_sync')
|
||
rec_id = (ns or {}).get('id')
|
||
if not rec_id:
|
||
return {'success': False, 'message': '缺少 id'}
|
||
db = DBPools()
|
||
async with db.sqlorContext(_get_dbname()) as sor:
|
||
await sor.D('world_sync', {'id': rec_id})
|
||
return {'success': True}
|
||
|
||
|
||
async def execute_world_sync(ns):
|
||
_require_host_deps('execute_world_sync')
|
||
ns = ns or {}
|
||
world_id = ns.get('world_id', '')
|
||
sync_type = ns.get('sync_type', '0')
|
||
source = ns.get('source', '')
|
||
if not world_id:
|
||
return {'success': False, 'message': '缺少 world_id'}
|
||
actions = {'0': '导入', '1': '导出', '2': '合并'}
|
||
if sync_type not in actions:
|
||
return {'success': False, 'message': '未知同步类型: %s' % sync_type}
|
||
action = actions[sync_type]
|
||
result = {'action': action, 'world_id': world_id, 'sync_type': sync_type, 'source': source, 'message': '世界%s完成' % action}
|
||
rec = {'id': getID(), 'world_id': world_id, 'sync_type': sync_type, 'source': source, 'status': '2', 'result_json': json.dumps(result, ensure_ascii=False), 'sync_date': curDateString(), 'created_at': curDateString()}
|
||
db = DBPools()
|
||
async with db.sqlorContext(_get_dbname()) as sor:
|
||
await sor.C('world_sync', rec)
|
||
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")
|
||
_require_host_deps('pbl_runtime_event_partitions')
|
||
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）、不含轮询兜底、不含时延优化。
|
||
# 写入器为**无状态函数式 API**（每次调用按三级优先级解析 conn_factory），
|
||
# init 阶段不建连；参数不全则调用点 fail-closed 抛 ConnectionUnavailable。
|
||
# ===========================================================================
|
||
|
||
def configure_runtime_writer(conn_factory=None, server_env=None, dialect=None):
|
||
"""把宿主连接工厂登记到 ServerEnv（契约解析优先级第 2 级），返回登记后的工厂。
|
||
|
||
真源实现（pbl_runtime_tx / pbl_runtime_tx_env）是函数式 API，不持有全局 writer
|
||
单例；因此「配置写入器」在 M11b-2a 后的语义 = 登记 conn_factory + 记录方言名。
|
||
|
||
:param conn_factory: 可调用对象（返回 DB-API 连接）或 DB URL 字符串
|
||
:param server_env: 目标 ServerEnv（None 时取全局单例）
|
||
:param dialect: 可选方言名（'mysql'/'sqlite'），仅登记供诊断用
|
||
:raises InvalidWriteRequest: conn_factory 不可调用且不是字符串
|
||
"""
|
||
if conn_factory is None:
|
||
raise ConnectionUnavailable(
|
||
"configure_runtime_writer requires an explicit conn_factory "
|
||
"(callable or DB URL); refusing to register nothing",
|
||
server_env_keys=list(SERVER_ENV_KEYS))
|
||
if not (callable(conn_factory) or isinstance(conn_factory, str)):
|
||
raise InvalidWriteRequest(
|
||
"conn_factory must be callable or a DB URL string",
|
||
got=type(conn_factory).__name__)
|
||
env = server_env if server_env is not None else ServerEnv()
|
||
setattr(env, SERVER_ENV_KEYS[0], conn_factory)
|
||
if dialect:
|
||
setattr(env, 'pbl_runtime_dialect', str(dialect).lower())
|
||
setattr(env, 'pbl_runtime_writer_ready', True)
|
||
try:
|
||
setattr(env, 'pbl_runtime_module_conf_dir', module_conf_dir())
|
||
except Exception as exc: # 仅诊断信息，不影响契约可用性
|
||
setattr(env, 'pbl_runtime_module_conf_dir_error', str(exc))
|
||
return conn_factory
|
||
|
||
|
||
def get_runtime_writer(conn_factory=None):
|
||
"""解析当前可用的连接工厂（三级优先级）；全空抛 ConnectionUnavailable（fail-closed）。"""
|
||
return resolve_conn_factory(conn_factory)
|
||
|
||
|
||
def build_runtime_write_plan(event_data, state_update, event_uid=None, source=None, now=None):
|
||
"""预演一次写入的标识装配（不落库）：返回 event_uid / event_type / source / 时间戳。
|
||
|
||
真源没有独立的 plan 对象（写入即执行），本函数只提供「调用前自检」能力，
|
||
便于接口层与自测脚本在不碰数据库的情况下确认参数装配是否完整。
|
||
"""
|
||
event_data = dict(event_data or {})
|
||
state_update = dict(state_update or {})
|
||
uid = event_uid or event_data.get('event_uid') or event_data.get('event_id') or new_event_uid()
|
||
return {
|
||
'event_uid': uid,
|
||
'event_type': event_data.get('event_type') or DEFAULT_EVENT_TYPE,
|
||
'source': source or event_data.get('source') or DEFAULT_SOURCE,
|
||
'tenant_id': event_data.get('tenant_id') or state_update.get('tenant_id'),
|
||
'session_id': event_data.get('session_id') or state_update.get('session_id'),
|
||
'entity_id': event_data.get('entity_id') or state_update.get('entity_id'),
|
||
'event_required': list(EVENT_REQUIRED),
|
||
'state_key_columns': list(STATE_KEY_COLUMNS),
|
||
'tables': {'event': EVENT_TABLE, 'state': STATE_TABLE},
|
||
'created_at': utc_now_text(now) if now else utc_now_text(),
|
||
}
|
||
|
||
|
||
def runtime_write_percentile(values, p=99):
|
||
"""时延分位数计算（供验收「P99 ≤ 200ms」自测脚本复用）。"""
|
||
return percentile(values, p)
|
||
|
||
|
||
def runtime_write_dialect_name(params=None):
|
||
"""当前登记的方言名（未登记时返回 None，不建连、不抛错，纯只读语义）。"""
|
||
try:
|
||
env = ServerEnv()
|
||
except Exception:
|
||
return None
|
||
name = getattr(env, 'pbl_runtime_dialect', None)
|
||
if name:
|
||
return name
|
||
getter = getattr(env, 'get', None)
|
||
if callable(getter):
|
||
try:
|
||
return getter('pbl_runtime_dialect', None)
|
||
except Exception:
|
||
return None
|
||
return None
|
||
|
||
|
||
def runtime_write_ddl(ns):
|
||
"""按方言返回建表 DDL（只读文本，供部署/自测建表，不执行）。"""
|
||
dialect = (ns or {}).get('dialect') or DIALECT_SQLITE
|
||
try:
|
||
return {'success': True, 'dialect': dialect, 'ddl': ddl_for(dialect)}
|
||
except Exception as exc:
|
||
return {'success': False, 'message': str(exc)}
|
||
|
||
|
||
def load_world_sync(server_env=None, conn_factory=None):
|
||
"""挂载 world_sync：CRUD + M11b-1 表注册 + M11b-2 单事务写入契约（唯一真源）。
|
||
|
||
三处接线之「③ init.py env.xxx = xxx」；本函数同时是「② __init__.py 导出」与
|
||
根包 ``world_sync/__init__.py`` 转发的**唯一** load_world_sync 实现（QC #2：
|
||
不得存在第二份定义）。模块级符号（含 ``TransactionAborted``）导入失败一律原样
|
||
抛出（不静默降级）。
|
||
|
||
:param server_env: 宿主 ServerEnv 实例；None 时取全局单例，单例不可用时用
|
||
:class:`_StandaloneEnv` 载体完成登记（仅接线自检，不建连）。
|
||
:param conn_factory: 宿主连接池工厂（可调用/DB URL）；提供则登记到
|
||
``pbl_runtime_conn_factory``（契约解析优先级第 2 级）。
|
||
:return: 登记完成的 env。
|
||
"""
|
||
env = _resolve_env(server_env)
|
||
if conn_factory is not None:
|
||
setattr(env, SERVER_ENV_KEYS[0], conn_factory)
|
||
env.world_sync_module = MODULE_NAME
|
||
env.world_sync_version = MODULE_VERSION
|
||
env.world_sync_host_deps_available = HOST_DEPS_AVAILABLE
|
||
if not HOST_DEPS_AVAILABLE:
|
||
env.world_sync_host_deps_error = repr(HOST_DEPS_ERROR)
|
||
env.list_world_syncs = list_world_syncs
|
||
env.get_world_sync = get_world_sync
|
||
env.create_world_sync = create_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.read_entity_state = read_entity_state
|
||
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.runtime_write_ddl = runtime_write_ddl
|
||
env.new_runtime_event_uid = new_event_uid
|
||
env.utc_now_text = utc_now_text
|
||
# 异常族（fail-closed，供接口层 except）
|
||
env.RuntimeWriteError = RuntimeWriteError
|
||
env.PblRuntimeError = PblRuntimeError
|
||
env.TxConfigError = TxConfigError
|
||
env.ConnectionUnavailable = ConnectionUnavailable
|
||
env.InvalidWriteRequest = InvalidWriteRequest
|
||
env.EventWriteError = EventWriteError
|
||
env.StateWriteError = StateWriteError
|
||
env.ConcurrentStateConflict = ConcurrentStateConflict
|
||
env.TransactionAborted = TransactionAborted
|
||
env.RuntimeTxAbortError = TxAbortError
|
||
env.new_runtime_event_id = new_event_id
|
||
# 表名/方言常量与 conn_factory 解析优先级说明
|
||
env.pbl_runtime_event_table = EVENT_TABLE
|
||
env.pbl_runtime_state_table = STATE_TABLE
|
||
env.pbl_runtime_tx_dialects = {'mysql': DIALECT_MYSQL, 'sqlite': DIALECT_SQLITE}
|
||
env.pbl_runtime_conn_factory_keys = list(SERVER_ENV_KEYS)
|
||
env.pbl_runtime_pool_keys = list(POOL_KEYS)
|
||
env.pbl_runtime_env_name_keys = list(ENV_NAME_KEYS)
|
||
# --- M11b-3: 事务提交后广播（异步 fire-and-forget，SLO <=300ms P99） ---
|
||
env.world_sync_broadcast_channel = BROADCAST_CHANNEL
|
||
env.world_sync_broadcast_latency_slo_ms = BROADCAST_LATENCY_SLO_MS
|
||
env.build_broadcast_message = build_broadcast_message
|
||
env.emit_post_commit_broadcast = emit_post_commit_broadcast
|
||
env.wrap_with_post_commit_broadcast = wrap_with_post_commit_broadcast
|
||
env.configure_broadcast = configure_broadcast
|
||
env.get_broadcast_broadcaster = get_broadcaster
|
||
env.subscribe_broadcast = subscribe_broadcast
|
||
env.unsubscribe_broadcast = unsubscribe_broadcast
|
||
env.broadcast_stats = broadcast_stats
|
||
env.broadcast_last_message = broadcast_last_message
|
||
env.broadcast_latency_percentile_ms = broadcast_latency_percentile_ms
|
||
env.drain_broadcast = drain_broadcast
|
||
env.shutdown_broadcast = shutdown_broadcast
|
||
env.reset_broadcast = reset_broadcaster
|
||
# --- M11b-4: 3s 轮询兜底（消费端侧，只读；不侵入 M11b-2 写入端） ---
|
||
env.POLL_INTERVAL_S = POLL_INTERVAL_S
|
||
env.POLL_TIMEOUT_S = POLL_TIMEOUT_S
|
||
env.POLL_LIMIT = POLL_LIMIT
|
||
env.FALLBACK_SLO_MS = FALLBACK_SLO_MS
|
||
env.SEEN_IDS_IN_SQL = SEEN_IDS_IN_SQL
|
||
env.SLO_TOLERANCE_MS = SLO_TOLERANCE_MS
|
||
env.E2E_LATENCY_SLO_MS = E2E_LATENCY_SLO_MS
|
||
env.PollingFallback = PollingFallback
|
||
env.build_poll_sql = build_poll_sql
|
||
env.normalize_watermark = normalize_watermark
|
||
env.plan_uses_scene_index = plan_uses_scene_index
|
||
env.configure_fallback = configure_fallback
|
||
env.get_fallback = get_fallback
|
||
env.reset_fallback = reset_fallback
|
||
env.fallback_stats = fallback_stats
|
||
env.explain_poll_query = explain_poll_query
|
||
env.poll_since = poll_since
|
||
env.run_fallback = run_fallback
|
||
env.wait_event = wait_event
|
||
# 消费端订阅侧标准接线：广播到点 → 推进兜底水位线 + 停止轮询
|
||
env.notify_broadcast = get_fallback().notify_broadcast
|
||
# M11b-4 端到端时延基准
|
||
env.measure_once = measure_once
|
||
env.run_chain = run_chain
|
||
env.summarize_latency = summarize_latency
|
||
env.latency_report = latency_report
|
||
# post-commit hook：把 write_event_with_state 包成「COMMIT 后触发广播」版本，
|
||
# 并同步刷新三处引用（pbl_runtime_tx_sqlor / world_sync 包命名空间 / env）。
|
||
env.write_event_with_state = install_broadcast_hook(env)
|
||
_configure_runtime_writer_from_env(env)
|
||
_register_tables_to_env(env)
|
||
return env
|
||
|
||
|
||
def _configure_runtime_writer_from_env(env):
|
||
"""按环境尽力登记 conn_factory（懒失败，init 阶段绝不建连）。
|
||
|
||
顺序：宿主连接池（sqlor pools）→ 环境变量 PBL_RUNTIME_DB_URL → 模块 conf/db.json。
|
||
任何「配置不全」都只记录原因并返回 None —— 这不是吞错误：写入契约一旦被调用
|
||
即 fail-closed 抛 :class:`ConnectionUnavailable`（见 resolve_conn_factory）。
|
||
"""
|
||
try:
|
||
factory = host_conn_factory()
|
||
except PblRuntimeError as exc:
|
||
env.pbl_runtime_writer_error = exc.as_dict()
|
||
env.pbl_runtime_writer_ready = False
|
||
return None
|
||
except Exception as exc: # 宿主 API 差异等意外：记录并原样包装，不静默
|
||
env.pbl_runtime_writer_error = {
|
||
'code': 'PBL_TX_HOST_PROBE_FAILED', 'message': str(exc)}
|
||
env.pbl_runtime_writer_ready = False
|
||
return None
|
||
setattr(env, SERVER_ENV_KEYS[0], factory)
|
||
env.pbl_runtime_writer_ready = True
|
||
return factory
|
||
|
||
|
||
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
|