基金转换 T-13：全量回归 + 50 并发压测 + 性能实测补录
交付：新增 tests/test_convert_concurrency.py（9 条 · CONVERT_STRESS 门禁）+ test_convert_service 同键重试 3 条；app/repository/convert_repository.insert_placeholder 改三步法（R-a）、app/service/convert/convert_service.py 幂等判定 pending→202。
修复 2 处并发缺陷（压测暴露，sqlite 不可见）：① 阶段一失败后同键重试永久 503（占位朴素 INSERT 撞 uk_group/uk_idem）；② 同键竞态子窗口 A 真·双扣 / B 双扣+503（撞 uk_idem 直穿）。
验证：pytest 736 passed / 10 skipped（+5，跑两遍稳定）；CONVERT_STRESS=1 并发 9/9；按 SOP 重灌双库复跑零回归；7 个 convert 真库脚本复跑零回归；突变 4 组全部精准命中。50 并发不超卖（成交 20~25）、紧池退避 40/50=80%（不够用）、松池 100%、跨客户 1213 死锁登记待评估；性能端到端 P50 44.2/max 59.8ms、阶段一 P50 9.8/max 16.7ms（未触阈值、未改实测值）。
文档：PRD v0.9.3（§9 第 18 条实测补录）+ 开发计划 §10（10.1~10.5）+ docs/memory/{TODO,MEMORY,FRAMEWORK,ITERATION,2026-09-10}。
This commit is contained in:
@@ -12,16 +12,20 @@ S2（评审）：清理**标记不硬删**——`mark_expired` 置 `status='expi
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import date, datetime, timedelta
|
||||
from decimal import Decimal
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.engine import Engine
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from app.config.settings import settings
|
||||
from app.utils.db import get_engine
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _to_bind(value: Any) -> Any:
|
||||
"""Decimal 转 float 再绑定（sqlite 不支持直接绑定 Decimal；MySQL DECIMAL 列自动收口）。
|
||||
@@ -45,25 +49,67 @@ class ConvertRepository:
|
||||
|
||||
# ---------- 阶段零：占位（uk_idem 兜底） ----------
|
||||
|
||||
def insert_placeholder(self, group_id: str, client_request_id: str | None) -> None:
|
||||
"""阶段零占位：插一行 `status='pending'`。
|
||||
def insert_placeholder(self, group_id: str, client_request_id: str | None) -> bool:
|
||||
"""阶段零占位：插一行 `status='pending'`；**已存在则置回 pending**。
|
||||
|
||||
返回 **True = 本笔持有该占位，可以继续**；**False = 该 `client_request_id`
|
||||
已被另一个 `group_id` 占住**（同键并发，本笔必须让路，由调用方回 202）。
|
||||
|
||||
`client_request_id` 为 None 时绑 NULL——MySQL / sqlite 的 UNIQUE 约束均允许多个 NULL，
|
||||
故「无幂等键的请求」可重复占位、互不冲突（uk_idem 仅对非空键兜底）。
|
||||
`estimated=0`：convert 占位是真实请求，非风控预估单（与 risk_alert 语义区分）。
|
||||
|
||||
**三步法（R-a：弃用方言 UPSERT，改「先查再 INSERT 或 UPDATE」）**。
|
||||
为什么必须容错「行已存在」：阶段一失败（典型是 `LotConflict` 409）时占位已被
|
||||
`mark_failed` 置为 `failed`，而架构 §8.3 要求调用方带**同一** `client_request_id`
|
||||
退避重试（≤3 次、100/200/400ms）；重试会走「复用原 group_id 重跑」分支再次进入
|
||||
阶段零——此时 `uk_group` 与 `uk_idem` 都已被那一行占用，朴素的 INSERT 必撞唯一键，
|
||||
把**可重试的 409 升级成 `IdempotencyUnavailable`(503)**，且是**确定性的**（重试永不成功），
|
||||
与 errors.py 里「瞬时状态、恢复后重试即可成功」的注释相反。（T-13 压测前置修复）
|
||||
|
||||
置回 `pending` 是正确语义：同一 `group_id` 的这一次尝试正在进行中。
|
||||
到达此处时既有行的状态只可能是 `pending`（上一轮中途崩溃）/ `failed`（阶段一或阶段二失败）；
|
||||
`completed` 已在 `convert_fund` 幂等前置分支返回，不会走到这里。
|
||||
|
||||
**为什么还要 catch IntegrityError**：`convert:idem:{cid}` 锁只包住幂等判定
|
||||
（出块即释放），两笔同键请求可能**都判定为"无占位"**、各自生成了不同的 `group_id`
|
||||
（T-13 真库实测：8 路并发下偶发）；后插入的那笔撞 `uk_idem` → 此前会直穿 503。
|
||||
这里把它收敛为 False（让路），交由调用方回 202（架构 §9「同键并发 → 202」）。
|
||||
"""
|
||||
sql = text(
|
||||
"""
|
||||
INSERT INTO risk_convert_detail
|
||||
(convert_group_id, client_request_id, status, estimated)
|
||||
VALUES (:gid, :cid_req, :status, 0)
|
||||
"""
|
||||
)
|
||||
with self._engine.begin() as conn:
|
||||
conn.execute(
|
||||
sql,
|
||||
{"gid": group_id, "cid_req": client_request_id, "status": _STATUS_PENDING},
|
||||
try:
|
||||
with self._engine.begin() as conn:
|
||||
existing = conn.execute(
|
||||
text("SELECT 1 FROM risk_convert_detail WHERE convert_group_id = :gid"),
|
||||
{"gid": group_id},
|
||||
).first()
|
||||
if existing is not None:
|
||||
conn.execute(
|
||||
text(
|
||||
"UPDATE risk_convert_detail"
|
||||
" SET status = :s WHERE convert_group_id = :gid"
|
||||
),
|
||||
{"s": _STATUS_PENDING, "gid": group_id},
|
||||
)
|
||||
return True
|
||||
conn.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO risk_convert_detail
|
||||
(convert_group_id, client_request_id, status, estimated)
|
||||
VALUES (:gid, :cid_req, :status, 0)
|
||||
"""
|
||||
),
|
||||
{"gid": group_id, "cid_req": client_request_id, "status": _STATUS_PENDING},
|
||||
)
|
||||
except IntegrityError:
|
||||
# uk_idem 竞态：同键的另一笔刚刚占位成功 → 本笔不持有，让路
|
||||
logger.info(
|
||||
"幂等键已被并发请求占用（本次让路）：cid=%s gid=%s",
|
||||
client_request_id,
|
||||
group_id,
|
||||
)
|
||||
return False
|
||||
return True
|
||||
|
||||
# ---------- 阶段二：回写 completed + 详情 ----------
|
||||
|
||||
|
||||
@@ -427,8 +427,22 @@ def convert_fund(
|
||||
)
|
||||
if rebuilt is not None:
|
||||
return rebuilt
|
||||
# 阶段一未成 → 复用同一 group_id 重跑，杜绝第二组流水
|
||||
group_id = hit_gid
|
||||
# 阶段一未成：必须区分「确定没跑成」与「在飞/未知」，否则同键并发会双跑
|
||||
# —— `convert:idem:` 锁只包住本段幂等判定（出块即释放），窗口内重入会
|
||||
# 与在飞的那笔**同时进入阶段一**（T-13 真库确定性交错实测：
|
||||
# 未映射的 IntegrityError 直穿 → 500；若日后 `in_lot_id` 的派生规则被
|
||||
# 改成随机值，同一窗口会升级为**真·双扣**）。
|
||||
# · `failed`：阶段一定性失败（无流水）→ 复用同一 group_id 重跑，
|
||||
# 这正是架构 §8.3「LOT_CONFLICT → 调用方退避重试」的服务端契约；
|
||||
# · `expired`：占位已被 SLA 巡检判死（`cleanup_pending_convert.py`，S2）
|
||||
# → 同样可安全复用；
|
||||
# · `pending`：在飞与崩溃**不可区分** → 按并发处理，返回 202 交客户端
|
||||
# 稍后重试（架构 §9「同键并发 → 202」）；崩溃残留由 SLA 巡检置 expired
|
||||
# 后自动放行（急用可 `cleanup_pending_convert.py --hours 0` 立即判死）。
|
||||
if str(existing["status"]) in ("failed", "expired"):
|
||||
group_id = hit_gid
|
||||
else:
|
||||
return {"status": PROCESSING, "convert_group_id": hit_gid}
|
||||
else:
|
||||
group_id = new_id("CNV", now)
|
||||
placeholder = True
|
||||
@@ -496,11 +510,17 @@ def convert_fund(
|
||||
# ⑤ 阶段零：占位（仅带幂等键时；在全部 4xx 之后，故 4xx 不留占位）
|
||||
if placeholder:
|
||||
try:
|
||||
crepo.insert_placeholder(group_id, cid_req)
|
||||
owns = crepo.insert_placeholder(group_id, cid_req)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
# 占位失败 = 无法保证幂等 → 不放行（PRD §7.3）
|
||||
logger.exception("convert 占位失败：%s", group_id)
|
||||
raise IdempotencyUnavailable(f"幂等占位失败：{exc}") from exc
|
||||
if not owns:
|
||||
# 同一 client_request_id 的并发请求已占住 uk_idem（各自生成过不同 group_id）
|
||||
# → 本笔让路，回 202（架构 §9「同键并发 → 202」），由客户端稍后重试并发起幂等命中。
|
||||
# 不这样收敛的话，此处会直穿未映射的 IntegrityError(503)。（T-13 真库实测）
|
||||
logger.info("同键并发占位让路：cid=%s gid=%s", cid_req, group_id)
|
||||
return {"status": PROCESSING, "convert_group_id": None}
|
||||
|
||||
out_trade_id = new_id("TRD", now)
|
||||
in_trade_id = new_id("TRD", now)
|
||||
|
||||
Reference in New Issue
Block a user