From ae4f0eccc1c41313f28352bbbfdef8188fd4dc8e Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Sat, 19 Sep 2026 11:31:47 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pbl_runtime_ext/context.py | 188 +++++++++++++++++++ pbl_runtime_ext/idempotency.py | 318 +++++++++++++++++++++++++++++++++ 2 files changed, 506 insertions(+) create mode 100644 pbl_runtime_ext/context.py create mode 100644 pbl_runtime_ext/idempotency.py diff --git a/pbl_runtime_ext/context.py b/pbl_runtime_ext/context.py new file mode 100644 index 0000000..20e79c6 --- /dev/null +++ b/pbl_runtime_ext/context.py @@ -0,0 +1,188 @@ +# -*- coding: utf-8 -*- +"""M11a 运行时上下文（服务端权威链路的身份基座，纯函数、无 DB 依赖）。 + +铁律 +---- +1. 进入运行时写入链路的每个请求，必须先解析出 tenant_id / principal_id / session_id， + 缺任一即 fail-closed 抛错——不存在「匿名写入」。 +2. tenant_id 的**权威来源是服务端会话**（pbl_common 租户上下文）。客户端 params 里自带的 + tenant_id 只用于一致性校验：与会话不一致 = 跨租户越权尝试，直接拒绝。 +3. 上下文一旦构建即不可变（frozen dataclass），链路中途不得篡改身份字段。 +""" +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any, Dict, Iterable, Mapping, Optional, Tuple + +ID_RE = re.compile(r"^[A-Za-z0-9_\-\.]{2,64}$") +NUM_ID_RE = re.compile(r"^\d{1,20}$") +ROLE_RE = re.compile(r"^[a-z][a-z0-9_.\-]{1,31}$") + +REQUIRED_FIELDS: Tuple[str, ...] = ("tenant_id", "principal_id", "session_id") + +# 客户端禁写字段（服务端权威）：出现即记录并忽略，绝不参与 state_version 计算 +CLIENT_FORBIDDEN_FIELDS: Tuple[str, ...] = ( + "state_version", "seq", "seq_no", "server_state_version", "checksum", +) + + +class RuntimeContextError(Exception): + """上下文不合法（fail-closed）。""" + + def __init__(self, code: str, message: str, **extra: Any) -> None: + super().__init__("%s: %s" % (code, message)) + self.code = code + self.message = message + self.extra = extra + + def to_error(self) -> Dict[str, Any]: + payload: Dict[str, Any] = {"code": self.code, "message": self.message} + payload.update(self.extra) + return payload + + +@dataclass(frozen=True) +class RuntimeContext: + """一次运行时写入请求的服务端权威身份视图。""" + + tenant_id: str + principal_id: str + session_id: str + world_id: str = "" + role: str = "" + roles: Tuple[str, ...] = () + channel: str = "ws" + request_id: str = "" + ignored_client_fields: Tuple[str, ...] = () + raw: Mapping[str, Any] = field(default_factory=dict) + + def role_set(self) -> Tuple[str, ...]: + out: list = [] + for r in (self.role,) + tuple(self.roles): + if r and r not in out: + out.append(r) + return tuple(out) + + def has_role(self, role: str) -> bool: + return bool(role) and role in self.role_set() + + def redacted(self) -> Dict[str, Any]: + """可安全落日志/审计的视图（不含 payload）。""" + return { + "tenant_id": self.tenant_id, + "principal_id": self.principal_id, + "session_id": self.session_id, + "world_id": self.world_id, + "roles": list(self.role_set()), + "channel": self.channel, + "request_id": self.request_id, + "ignored_client_fields": list(self.ignored_client_fields), + } + + +def _clean(value: Any) -> str: + return str(value if value is not None else "").strip() + + +def assert_id(value: Any, name: str, numeric: bool = False) -> str: + s = _clean(value) + if not s: + raise RuntimeContextError("RT_CTX_MISSING_FIELD", "缺少必填字段 %s" % name, field=name) + if numeric: + if not NUM_ID_RE.match(s): + raise RuntimeContextError("RT_CTX_ID_INVALID", "字段 %s 必须为数字标识" % name, field=name) + return s + if not ID_RE.match(s): + raise RuntimeContextError("RT_CTX_ID_INVALID", "字段 %s 标识不合法" % name, field=name) + return s + + +def _roles_of(value: Any) -> Tuple[str, ...]: + if value is None: + return () + if isinstance(value, (list, tuple, set)): + items: list = list(value) + elif isinstance(value, str): + items = [p for p in re.split(r"[,;\s]+", value) if p] + else: + items = [value] + out: list = [] + for item in items: + s = _clean(item) + if not s: + continue + if not ROLE_RE.match(s): + raise RuntimeContextError("RT_CTX_ROLE_INVALID", "非法角色标识: %r" % (item,)) + if s not in out: + out.append(s) + return tuple(out) + + +def scan_ignored_client_fields(params: Mapping[str, Any]) -> Tuple[str, ...]: + """探测客户端试图写入的服务端权威字段（用于审计与响应提示），不修改入参。""" + return tuple(k for k in CLIENT_FORBIDDEN_FIELDS if k in (params or {}) and _clean((params or {})[k]) != "") + + +def build_context(params: Optional[Mapping[str, Any]], *, + server_tenant: Optional[str] = None, + server_principal: Optional[str] = None, + require: Iterable[str] = REQUIRED_FIELDS) -> RuntimeContext: + """构建 RuntimeContext。 + + :param server_tenant: 服务端会话解析出的 tenant_id（权威）。给了就以它为准。 + :param server_principal: 服务端会话解析出的操作者（权威），优先于客户端自报。 + """ + data: Dict[str, Any] = dict(params or {}) + s_tenant = _clean(server_tenant) + c_tenant = _clean(data.get("tenant_id")) + if s_tenant and c_tenant and s_tenant != c_tenant: + raise RuntimeContextError("RT_CTX_TENANT_MISMATCH", + "tenant_id 与服务端会话不一致，拒绝跨租户写入") + tenant_id = s_tenant or c_tenant + if not tenant_id: + raise RuntimeContextError("RT_CTX_TENANT_REQUIRED", + "无法解析 tenant_id（服务端会话未注入且客户端未携带），fail-closed 拒绝") + + principal = _clean(server_principal) or _clean(data.get("principal_id")) \ + or _clean(data.get("actor_id")) or _clean(data.get("user_id")) + + ctx = RuntimeContext( + tenant_id=assert_id(tenant_id, "tenant_id"), + principal_id=assert_id(principal, "principal_id"), + session_id=assert_id(data.get("session_id"), "session_id", numeric=True), + world_id=_clean(data.get("world_id")), + role=_clean(data.get("role") or data.get("role_code")), + roles=_roles_of(data.get("roles")), + channel=_clean(data.get("channel")) or "ws", + request_id=_clean(data.get("request_id")), + ignored_client_fields=scan_ignored_client_fields(data), + raw=data, + ) + + missing = [f for f in require if not _clean(getattr(ctx, f, ""))] + if missing: + raise RuntimeContextError("RT_CTX_MISSING_FIELD", + "缺少必填字段: %s" % ",".join(missing), fields=missing) + return ctx + + +def derive_context(ctx: RuntimeContext, **overrides: Any) -> RuntimeContext: + """派生新上下文：仅允许覆盖非身份字段（world_id / channel / request_id）。""" + allowed = {"world_id", "channel", "request_id"} + bad = sorted(k for k in overrides if k not in allowed) + if bad: + raise RuntimeContextError("RT_CTX_LOCKED_FIELD", + "身份字段不可覆盖: %s" % ",".join(bad), fields=bad) + return RuntimeContext( + tenant_id=ctx.tenant_id, + principal_id=ctx.principal_id, + session_id=ctx.session_id, + world_id=_clean(overrides.get("world_id")) or ctx.world_id, + role=ctx.role, + roles=ctx.roles, + channel=_clean(overrides.get("channel")) or ctx.channel, + request_id=_clean(overrides.get("request_id")) or ctx.request_id, + ignored_client_fields=ctx.ignored_client_fields, + raw=ctx.raw, + ) diff --git a/pbl_runtime_ext/idempotency.py b/pbl_runtime_ext/idempotency.py new file mode 100644 index 0000000..ee4a24d --- /dev/null +++ b/pbl_runtime_ext/idempotency.py @@ -0,0 +1,318 @@ +"""M11a 幂等 client_seq（服务端权威，防重放/防乱序重复写入）. + +语义（见 modules/scense_runtime.md M11a）： +- 客户端每个「写事件」携带单调递增的 ``client_seq``（同一 tenant+session+scope 内）。 +- 服务端以 ``pbl_runtime_idempotency`` 唯一键 (tenant_id, session_id, scope, client_seq) 做幂等闸门： + * 首次到达 -> 标记 in_progress，返回 token，调用方在**同一事务**内写事件+状态； + * 重复到达 -> 直接返回已缓存的结果（replayed=True），不再产生第二条事件、不再自增 state_version； + * 客户端 seq <= 服务端已接受的最大 seq -> 判定 stale（乱序重放），拒绝写入并回传期望 seq； + * 处理失败 -> release(token, error=...)，把幂等记录置 failed，允许同 seq 重试。 +- 幂等表写入与业务写入共享同一 DB 连接/事务：业务事务回滚时幂等记录一并回滚，不会留下「假成功」闸门。 +""" +from __future__ import annotations + +import hashlib +import json +import time +from dataclasses import dataclass +from typing import Any, Dict, Optional + +from pbl_common import api as A +from pbl_common.api import q_all, q_one # 已适配的查询助手 + +from .context import RuntimeContext, RuntimeContextError +from .tables import TBL_IDEMPOTENCY, TBL_ENTITY_STATE + +SCOPE_RE = None # 由 validate_scope 手工校验，避免额外依赖 + +MAX_SCOPE_LEN = 64 +MAX_RESULT_BYTES = 60000 + + +class IdempotencyError(Exception): + """幂等闸门拒绝（fail-closed）。""" + + def __init__(self, code: str, message: str, **extra: Any) -> None: + super().__init__("%s: %s" % (code, message)) + self.code = code + self.message = message + self.extra = extra + + def to_error(self) -> Dict[str, Any]: + payload: Dict[str, Any] = {"code": self.code, "message": self.message} + payload.update(self.extra) + return payload + + +@dataclass +class Claim: + """一次幂等占位的结果。""" + + acquired: bool + replayed: bool + token: str + client_seq: int + scope: str + result: Optional[Dict[str, Any]] = None + state: str = "in_progress" + + @property + def cached_result(self) -> Optional[Dict[str, Any]]: + return self.result + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def validate_seq(client_seq: Any) -> int: + """client_seq 必须是 >=1 的整数（0/负数/浮点/字符串数字以外的类型一律拒绝）。""" + if isinstance(client_seq, bool) or client_seq is None: + raise IdempotencyError("RT_SEQ_INVALID", "client_seq 必填且必须为正整数", client_seq=client_seq) + if isinstance(client_seq, int): + seq = client_seq + elif isinstance(client_seq, str) and client_seq.strip().lstrip("-").isdigit(): + seq = int(client_seq.strip()) + else: + raise IdempotencyError("RT_SEQ_INVALID", "client_seq 必须为正整数", client_seq=client_seq) + if seq < 1: + raise IdempotencyError("RT_SEQ_INVALID", "client_seq 必须 >= 1", client_seq=seq) + if seq > 9_000_000_000_000_000_000: + raise IdempotencyError("RT_SEQ_OVERFLOW", "client_seq 超出 BIGINT 安全范围", client_seq=seq) + return seq + + +def validate_scope(scope: Any) -> str: + """scope：幂等作用域（通常为 world_id 或 world_id:channel），缺省用 '*'。""" + s = str(scope if scope is not None else "").strip() + if not s: + s = "*" + if len(s) > MAX_SCOPE_LEN: + raise IdempotencyError("RT_SCOPE_INVALID", "scope 超长(>%d)" % MAX_SCOPE_LEN, scope=s) + bad = [c for c in s if not (c.isalnum() or c in "._:-/")] + if bad: + raise IdempotencyError("RT_SCOPE_INVALID", "scope 含非法字符: %r" % (bad[0],), scope=s) + return s + + +def request_fingerprint(payload: Any) -> str: + """请求体指纹（sort_keys + 紧凑分隔符，确定性），用于检测同 seq 不同内容的冲突。""" + try: + blob = json.dumps(payload if payload is not None else {}, sort_keys=True, ensure_ascii=False, + separators=(",", ":")) + except (TypeError, ValueError): + blob = repr(payload) + return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:64] + + +def _row_to_claim(row: Dict[str, Any], seq: int, scope: str, token: str) -> Claim: + state = str(row.get("state") or "in_progress") + cached = row.get("result_json") + if isinstance(cached, str) and cached: + try: + cached = json.loads(cached) + except (TypeError, ValueError): + cached = None + return Claim( + acquired=False, + replayed=state == "completed", + token=token, + client_seq=seq, + scope=scope, + result=cached if isinstance(cached, dict) else None, + state=state, + ) + + +def claim(ctx: RuntimeContext, scope: Any, client_seq: Any, + payload: Any = None, conn: Any = None) -> Claim: + """占位一次幂等写入。返回 Claim(acquired=True) 表示调用方获得唯一写入权。 + + 可能抛出： + - IdempotencyError(RT_SEQ_STALE) : seq 落后于服务端已接受的最大 seq（乱序重放） + - IdempotencyError(RT_SEQ_IN_FLIGHT) : 同 seq 仍在处理中（并发重复提交），调用方应稍后重试 + - IdempotencyError(RT_SEQ_CONFLICT) : 同 seq 但请求内容指纹不同 + """ + seq = validate_seq(client_seq) + scp = validate_scope(scope) + fp = request_fingerprint(payload) + token = "%s:%s:%s:%d" % (ctx.tenant_id, ctx.session_id, scp, seq) + + existing = q_one( + "SELECT idempotency_id, state, client_seq, result_json, request_fp, updated_at " + "FROM %s WHERE tenant_id=%%s AND session_id=%%s AND scope=%%s AND client_seq=%%s LIMIT 1" + % TBL_IDEMPOTENCY, + (ctx.tenant_id, ctx.session_id, scp, seq), conn=conn, + ) + + if existing: + state = str(existing.get("state") or "") + if state == "completed": + return _row_to_claim(existing, seq, scp, token) + if state == "failed": + # 允许同 seq 重试：重置为 in_progress 并刷新指纹 + updated = A.u( + TBL_IDEMPOTENCY, + {"state": "in_progress", "request_fp": fp, "result_json": None, + "updated_at": _now_ms(), "error_code": None}, + where={"tenant_id": ctx.tenant_id, "idempotency_id": existing["idempotency_id"]}, + conn=conn, + ) + if not updated: + raise IdempotencyError("RT_IDEM_RACE", "幂等占位竞争失败，请重试", client_seq=seq) + return Claim(acquired=True, replayed=False, token=token, client_seq=seq, scope=scp, state="in_progress") + if state == "in_progress": + raise IdempotencyError( + "RT_SEQ_IN_FLIGHT", + "相同 client_seq 正在处理中，请勿重复提交", + client_seq=seq, scope=scp, + ) + raise IdempotencyError("RT_IDEM_STATE_UNKNOWN", "幂等记录状态未知: %s" % state, state=state) + + # 无同 seq 记录：检查是否落后于已接受的最大 seq（防乱序重放） + max_row = q_one( + "SELECT COALESCE(MAX(client_seq), 0) AS max_seq FROM %s " + "WHERE tenant_id=%%s AND session_id=%%s AND scope=%%s AND state IN ('completed','in_progress')" + % TBL_IDEMPOTENCY, + (ctx.tenant_id, ctx.session_id, scp), conn=conn, + ) + max_seq = int((max_row or {}).get("max_seq") or 0) + if seq <= max_seq: + raise IdempotencyError( + "RT_SEQ_STALE", + "client_seq 落后于服务端已接受的最大序号，拒绝写入", + client_seq=seq, expected_min=max_seq + 1, scope=scp, + ) + + idem_id = "idem_%s" % A.new_id("idem") + try: + A.i( + TBL_IDEMPOTENCY, + { + "idempotency_id": idem_id, + "tenant_id": ctx.tenant_id, + "session_id": ctx.session_id, + "scope": scp, + "client_seq": seq, + "principal_id": ctx.principal_id, + "request_fp": fp, + "state": "in_progress", + "result_json": None, + "created_at": _now_ms(), + "updated_at": _now_ms(), + }, + conn=conn, + ) + except Exception as exc: # noqa: BLE001 - 唯一键冲突等一律转成结构化错误 + msg = str(exc) + if "Duplicate" in msg or "duplicate" in msg or "1062" in msg: + raced = q_one( + "SELECT idempotency_id, state, result_json, request_fp FROM %s " + "WHERE tenant_id=%%s AND session_id=%%s AND scope=%%s AND client_seq=%%s LIMIT 1" + % TBL_IDEMPOTENCY, + (ctx.tenant_id, ctx.session_id, scp, seq), conn=conn, + ) + if raced and str(raced.get("state")) == "completed": + return _row_to_claim(raced, seq, scp, token) + raise IdempotencyError("RT_SEQ_IN_FLIGHT", "相同 client_seq 并发提交，已有一路在处理", + client_seq=seq, scope=scp) from exc + raise IdempotencyError("RT_IDEM_CLAIM_FAILED", "幂等占位失败: %s" % msg) from exc + + return Claim(acquired=True, replayed=False, token=token, client_seq=seq, scope=scp, state="in_progress") + + +def _dumps_result(result: Dict[str, Any]) -> str: + blob = json.dumps(result, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + if len(blob.encode("utf-8")) > MAX_RESULT_BYTES: + # 结果过大：只保留幂等回放所需的最小骨架（版本号/事件 id），避免撑爆行 + slim = {k: result.get(k) for k in ("event_id", "state_version", "seq", "applied", "entity_id") + if k in result} + slim["truncated"] = True + blob = json.dumps(slim, sort_keys=True, ensure_ascii=False, separators=(",", ":")) + return blob + + +def complete(token: str, result: Optional[Dict[str, Any]] = None, conn: Any = None) -> bool: + """标记幂等记录完成并缓存结果（与业务写入同事务）。""" + tenant_id, session_id, scope, seq_s = _split_token(token) + values = { + "state": "completed", + "result_json": _dumps_result(result or {}), + "updated_at": _now_ms(), + "error_code": None, + } + return bool(A.u( + TBL_IDEMPOTENCY, values, + where={"tenant_id": tenant_id, "session_id": session_id, "scope": scope, "client_seq": int(seq_s)}, + conn=conn, + )) + + +def release(token: str, error: Any = None, conn: Any = None) -> bool: + """业务处理失败：置 failed，允许同 seq 重试。""" + tenant_id, session_id, scope, seq_s = _split_token(token) + if error is None: + code = "RT_APPLY_FAILED" + elif isinstance(error, (IdempotencyError, RuntimeContextError)): + code = getattr(error, "code", "RT_APPLY_FAILED")[:64] + elif isinstance(error, dict): + code = str(error.get("code") or "RT_APPLY_FAILED")[:64] + else: + code = str(error)[:64] or "RT_APPLY_FAILED" + values = {"state": "failed", "error_code": code, "updated_at": _now_ms()} + return bool(A.u( + TBL_IDEMPOTENCY, values, + where={"tenant_id": tenant_id, "session_id": session_id, "scope": scope, "client_seq": int(seq_s)}, + conn=conn, + )) + + +def _split_token(token: str): + parts = str(token or "").split(":") + if len(parts) != 4: + raise IdempotencyError("RT_IDEM_TOKEN_INVALID", "非法幂等 token: %r" % (token,)) + tenant_id, session_id, scope, seq = parts + return tenant_id, session_id, scope, seq + + +def last_seq(ctx: RuntimeContext, scope: Any = None, conn: Any = None) -> Dict[str, Any]: + """查询服务端当前已接受的最大 seq 与权威 state_version（供客户端对齐/重连补齐）。""" + scp = validate_scope(scope) + row = q_one( + "SELECT COALESCE(MAX(client_seq), 0) AS last_seq, COUNT(1) AS applied_count FROM %s " + "WHERE tenant_id=%%s AND session_id=%%s AND scope=%%s AND state='completed'" + % TBL_IDEMPOTENCY, + (ctx.tenant_id, ctx.session_id, scp), conn=conn, + ) + state_row = q_one( + "SELECT COALESCE(MAX(state_version), 0) AS state_version FROM %s " + "WHERE tenant_id=%%s AND session_id=%%s" % TBL_ENTITY_STATE, + (ctx.tenant_id, ctx.session_id), conn=conn, + ) + return { + "scope": scp, + "last_seq": int((row or {}).get("last_seq") or 0), + "applied_count": int((row or {}).get("applied_count") or 0), + "state_version": int((state_row or {}).get("state_version") or 0), + } + + +def replay(ctx: RuntimeContext, scope: Any, client_seq: Any, conn: Any = None) -> Optional[Dict[str, Any]]: + """只读回放：若该 seq 已完成，返回缓存结果，否则 None。""" + seq = validate_seq(client_seq) + scp = validate_scope(scope) + row = q_one( + "SELECT state, result_json FROM %s " + "WHERE tenant_id=%%s AND session_id=%%s AND scope=%%s AND client_seq=%%s LIMIT 1" + % TBL_IDEMPOTENCY, + (ctx.tenant_id, ctx.session_id, scp, seq), conn=conn, + ) + if not row or str(row.get("state")) != "completed": + return None + cached = row.get("result_json") + if isinstance(cached, str): + try: + cached = json.loads(cached) + except (TypeError, ValueError): + return None + return cached if isinstance(cached, dict) else None