deliver: 交付收口（引擎代为提交）
This commit is contained in:
agent.develop
parent
8e2d079549
commit
ae4f0eccc1
188
pbl_runtime_ext/context.py
Normal file
188
pbl_runtime_ext/context.py
Normal file
@ -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,
|
||||
)
|
||||
318
pbl_runtime_ext/idempotency.py
Normal file
318
pbl_runtime_ext/idempotency.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user