848 lines
37 KiB
Python
848 lines
37 KiB
Python
"""T-13 真 MySQL 并发测试 · **T+1 受理/确认分离模型**（架构 §9 · PRD 验收 18/31）。
|
||
|
||
**为什么必须是真 MySQL**：本文件验的是 InnoDB **行锁 + 条件 UPDATE rowcount** 这套
|
||
数据库级并发原语。sqlite 内存库（`conftest.sqlite_engine`）共用单连接，多线程跑不出
|
||
真争抢——在那里"绿"是**假绿**（与本项目「方言语义一律上真库」的既有铁律一致）。
|
||
|
||
**为什么不算 CI 常规门禁**（架构 §10 原话：该用例不进 CI，归入 T-13 一并跑）：
|
||
重载用例要 50 线程 + 真库，耗时长。本文件用环境变量 **`CONVERT_STRESS=1`** 实现该语义：
|
||
未设置时重载用例逐条 skip（给出可操作的原因），常规 `pytest -q` 不受拖累。
|
||
T-13 实测：`CONVERT_STRESS=1 pytest tests/test_convert_concurrency.py -q -s`。
|
||
|
||
**T+1 语义下的并发面（v1.0 用例的映射，convert_fund 退役后整体重写）**
|
||
|
||
- **份额争抢从受理挪到确认**：受理段只做 R-3 在途占用校验（不扣份额）；
|
||
真正的 FIFO 扣减争抢在确认段 `apply_convert`（批次哨兵 `remain_qty >= :q` +
|
||
1213 死锁整事务重试）。硬不变量「不超卖」钉在**确认后**的流水/余量上。
|
||
- **受理超授竞态**：并发受理下 R-3 的占用读数会滞后（TOCTOU）→ 可能出现
|
||
Σ占用 > 池的**超授**；T+1 的兜底是确认段逐笔 FIFO 严格扣减（份额不足 →
|
||
部分成交 R-10 / `rejected`），绝不超卖 —— 这是设计语义，不是缺陷。
|
||
- **紧池 80% 病根消失**：v1.0「需求=供给并发扣减只到 80% 成交」的根因是实时
|
||
扣减的批次碎片争抢；T+1 确认**串行**（FR-C23）后同场景 100% 成交
|
||
（验收 31 设计目标，S1 用例钉住）。
|
||
- **v1.0 占位竞态用例（3b/3c）退役**：T+1 受理单是 Core 6 态状态机（无 agent
|
||
占位），同键幂等由 `uk_idem` + 客户级锁承接（N1 + 单测
|
||
`test_accept_uk_idem_race_falls_back_to_idempotent`）。
|
||
|
||
**隔离策略**（与 `test_convert_integration.py` 同款，三条同时成立，缺一即污染种子）：
|
||
1. id 前缀：`CNV-CONC-` / `TRD-CONC-`（monkeypatch `convert_service._new_id`）；
|
||
2. 数据自建：客户 `CUST-CONCTEST` / 产品 `PROD-CONCTEST?` 全自建，**绝不碰种子**——
|
||
`risk_demo_env` 的 teardown 不还原 `core_share_lot`/`core_holding`，借种子客户跑转换
|
||
会跨用例污染 `test_integration_risk.py`；
|
||
3. teardown 全清：函数级 fixture 按客户 + 前缀删两库全部自建行（幂等）。
|
||
|
||
**连接池**：并发用例用**专用 Engine**（`pool_size=60`）。默认池 `5+10` 会把 50 个线程
|
||
压成 15 路串行，削弱争抢强度、让"不超卖"变成弱证据；专用池只在压测用，常规用例仍走
|
||
`get_engine` 生产口径。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
import time
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from datetime import date, datetime, time as dtime, timedelta
|
||
from decimal import Decimal
|
||
from statistics import mean
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from sqlalchemy import create_engine, text
|
||
from sqlalchemy.exc import OperationalError
|
||
|
||
from conftest import ensure_risk_demo_ready
|
||
|
||
from app.config.settings import settings # noqa: E402
|
||
from app.gateway.convert_core_repository import ConvertCoreRepository # noqa: E402
|
||
from app.repository.convert_repository import ConvertRepository # noqa: E402
|
||
from app.repository.convert_request_repository import ConvertRequestRepository # noqa: E402
|
||
from app.repository.core_ro import CoreReadOnlyRepository # noqa: E402
|
||
from app.repository.risk_repository import RiskRepository # noqa: E402
|
||
from app.service.convert import convert_service as cs # noqa: E402
|
||
from app.service.convert.confirm_service import confirm_batch, confirm_one # noqa: E402
|
||
from app.service.convert.convert_service import ( # noqa: E402
|
||
accept_convert,
|
||
compensate_convert,
|
||
)
|
||
from app.service.convert.errors import LotConflict # noqa: E402
|
||
from app.service.convert.trading_calendar import ( # noqa: E402
|
||
parse_cutoff,
|
||
resolve_accept_date,
|
||
)
|
||
from app.service.risk.rules import RiskThresholds # noqa: E402
|
||
from app.utils.db import _resolve_credentials # noqa: E402
|
||
|
||
ensure_risk_demo_ready()
|
||
|
||
# ── 隔离种子常量（全部带 CONCTEST 前缀）──────────────────────────────
|
||
CUSTOMER = "CUST-CONCTEST"
|
||
PROD_OUT = "PROD-CONCTESTO" # 转出：债基，申购费率 0.0030
|
||
PROD_IN = "PROD-CONCTESTI" # 转入：股基，申购费率 0.0080
|
||
COMPANY = "华夏模拟基金"
|
||
TA = "TA-CN-001"
|
||
OUT_RATE = Decimal("0.0030")
|
||
IN_RATE = Decimal("0.0080")
|
||
OUT_NAV = Decimal("1.0300")
|
||
IN_NAV = Decimal("0.9500")
|
||
FEE_TIERS = [
|
||
(0, 7, "0.0150"),
|
||
(7, 30, "0.0100"),
|
||
(30, 180, "0.0050"),
|
||
(180, 365, "0.0025"),
|
||
(365, None, "0.0000"),
|
||
]
|
||
_LOT_DAYS = 100 # 全部批次同一费率档（0.0050）→ 断言只看份额守恒，不被费用档干扰
|
||
|
||
_GROUP_LIKE = "CNV-CONC-%"
|
||
|
||
#: 架构 §8.3 建议的退避间隔（瞬时失败 → 调用方重试 ≤3 次）
|
||
BACKOFF_MS = (100, 200, 400)
|
||
#: 可重试异常（架构 §8.3 外部退避兜底的对象）：批次争抢 `LotConflict` +
|
||
#: 确认事务 1213 死锁重试耗尽后上抛的 `OperationalError`。串行批处理（FR-C23）
|
||
# 下二者不应出现；绕过批处理锁并发 confirm_one 时由调用方退避兜底。
|
||
_RETRYABLE = (LotConflict, OperationalError)
|
||
_RETRYABLE_NAMES = {t.__name__ for t in _RETRYABLE}
|
||
|
||
|
||
def _stress_enabled() -> bool:
|
||
return os.environ.get("CONVERT_STRESS") == "1"
|
||
|
||
|
||
requires_stress = pytest.mark.skipif(
|
||
not _stress_enabled(),
|
||
reason="重载并发用例不进 CI 常规门禁（架构 §10）；设 CONVERT_STRESS=1 显式开启",
|
||
)
|
||
|
||
|
||
def _noop_engine(out_trade: dict, in_trade: dict) -> dict:
|
||
"""空引擎 hook：并发用例只验**份额争抢**，不掺引擎/出单噪声。"""
|
||
return {"triggered_rules": [], "alert_ids": [], "aml_hit": False}
|
||
|
||
|
||
def _new_id(prefix: str, now: datetime) -> str:
|
||
"""`_new_id` 替换：id 带 `-CONC-` 段（与 `CNV-CONC-%` 清理前缀一致）。"""
|
||
return f"{prefix}-CONC-{uuid4().hex[:10].upper()}"
|
||
|
||
|
||
# ── 专用 Engine（放大连接池）─────────────────────────────────────────
|
||
def _big_pool_engine(database: str, role: str, size: int = 60):
|
||
"""建一个连接池放大的 Engine（与 `get_engine` 同 URL/凭据，仅池参数不同）。"""
|
||
user, password = _resolve_credentials(database, role)
|
||
auth = f"{user}:{password}" if password else user
|
||
url = (
|
||
f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}"
|
||
f"/{database}?charset=utf8mb4"
|
||
)
|
||
return create_engine(url, pool_pre_ping=True, pool_size=size, max_overflow=size)
|
||
|
||
|
||
# ── 种子与清理 ──────────────────────────────────────────────────────
|
||
def _accept_date(core) -> date:
|
||
"""受理日 T（生产 `resolve_accept_date` 走真库日历实时判定，脚本不手算）。"""
|
||
return resolve_accept_date(
|
||
_NOW,
|
||
CoreReadOnlyRepository(engine=core).is_open,
|
||
parse_cutoff(settings.convert_cutoff_time),
|
||
)
|
||
|
||
|
||
def _seed(core, lots: list[str]) -> Decimal:
|
||
"""自建隔离种子；`lots` = 各批次份额。返回份额池总量。
|
||
|
||
两端净值都锚在**受理日 T**（确认段 `get_nav_on(pid, T)` 精确匹配）。
|
||
"""
|
||
today = date.today()
|
||
base = datetime.combine(today, dtime(10, 0))
|
||
accept_day = _accept_date(core)
|
||
total = sum(Decimal(q) for q in lots)
|
||
with core.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer (customer_id, display_name, open_date)"
|
||
" VALUES (:c, 'T13并发测试', :d)"
|
||
),
|
||
{"c": CUSTOMER, "d": today},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at)"
|
||
" VALUES (:c, 'C5', :t, :exp)"
|
||
),
|
||
{
|
||
"c": CUSTOMER,
|
||
"t": base - timedelta(days=30),
|
||
"exp": base + timedelta(days=300),
|
||
},
|
||
)
|
||
for pid, name, ptype, rate in [
|
||
(PROD_OUT, "T13转出基金", "bond", OUT_RATE),
|
||
(PROD_IN, "T13转入基金", "stock", IN_RATE),
|
||
]:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_product (product_id, product_name, min_risk_code,"
|
||
" product_type, can_subscribe, can_redeem, subscribe_fee_rate,"
|
||
" min_redeem_qty, min_hold_qty, min_hold_action, fund_company, ta_code)"
|
||
" VALUES (:p, :n, 'R2', :t, 1, 1, :r, 0, 0, 'force_transfer', :co, :ta)"
|
||
),
|
||
{"p": pid, "n": name, "t": ptype, "r": str(rate), "co": COMPANY, "ta": TA},
|
||
)
|
||
for mh, mh_max, rate in FEE_TIERS:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days,"
|
||
" max_hold_days, rate) VALUES (:p, 'redeem', :mh, :mm, :r)"
|
||
),
|
||
{"p": PROD_OUT, "mh": mh, "mm": mh_max, "r": rate},
|
||
)
|
||
for pid, nav in ((PROD_OUT, OUT_NAV), (PROD_IN, IN_NAV)):
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date)"
|
||
" VALUES (:p, :n, 0, :d)"
|
||
),
|
||
{"p": pid, "n": str(nav), "d": accept_day},
|
||
)
|
||
for i, qty in enumerate(lots, start=1):
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty,"
|
||
" remain_qty, nav, confirmed_at)"
|
||
" VALUES (:l, :c, :p, :q, :q, :n, :cat)"
|
||
),
|
||
{
|
||
"l": f"LOT-CONCTEST-{i}",
|
||
"c": CUSTOMER,
|
||
"p": PROD_OUT,
|
||
"q": qty,
|
||
"n": str(OUT_NAV),
|
||
"cat": base - timedelta(days=_LOT_DAYS),
|
||
},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_holding (customer_id, product_id, qty, cost_amount,"
|
||
" market_value, pnl_pct, as_of) VALUES (:c, :p, :q, :cost, :mv, 0, :d)"
|
||
),
|
||
{
|
||
"c": CUSTOMER,
|
||
"p": PROD_OUT,
|
||
"q": str(total),
|
||
"cost": str(total * OUT_NAV),
|
||
"mv": str(total * OUT_NAV),
|
||
"d": today,
|
||
},
|
||
)
|
||
return total
|
||
|
||
|
||
_CUST_LIKE = "CUST-CONCTEST%"
|
||
_PROD_LIKE = "PROD-CONCTEST%"
|
||
#: 删除顺序按真库 FK 依赖（`information_schema.KEY_COLUMN_USAGE` 实测）：
|
||
#: `core_holding`/`core_share_lot`/`core_trade` 同时引用 `core_customer` 与
|
||
#: `core_product`，故必须先删它们，再删 `core_product`，最后删 `core_customer`。
|
||
_CORE_CLEANUP_BY_CUSTOMER = [
|
||
"DELETE FROM core_convert_lot_detail WHERE convert_group_id LIKE 'CNV-CONC-%'",
|
||
"DELETE FROM core_trade WHERE customer_id LIKE :cl",
|
||
"DELETE FROM core_share_lot WHERE customer_id LIKE :cl",
|
||
"DELETE FROM core_holding WHERE customer_id LIKE :cl",
|
||
"DELETE FROM core_cash_flow WHERE customer_id LIKE :cl",
|
||
"DELETE FROM core_customer_advisor WHERE customer_id LIKE :cl",
|
||
"DELETE FROM core_customer_risk WHERE customer_id LIKE :cl",
|
||
# T-12：受理单引用 customer/product（fk_creq_*），必须在删 product 之前清
|
||
"DELETE FROM core_convert_request WHERE customer_id LIKE :cl",
|
||
]
|
||
_CORE_CLEANUP_BY_PRODUCT = [
|
||
f"DELETE FROM core_fee_rule WHERE product_id LIKE '{_PROD_LIKE}'",
|
||
f"DELETE FROM core_product_nav WHERE product_id LIKE '{_PROD_LIKE}'",
|
||
f"DELETE FROM core_convert_request WHERE from_product_id LIKE '{_PROD_LIKE}'"
|
||
f" OR to_product_id LIKE '{_PROD_LIKE}'",
|
||
f"DELETE FROM core_product WHERE product_id LIKE '{_PROD_LIKE}'",
|
||
]
|
||
|
||
|
||
def _cleanup(core, agent) -> None:
|
||
"""按客户 + 前缀清两库全部自建行（幂等；seed 失败亦可清）。"""
|
||
params = {"cl": _CUST_LIKE}
|
||
with core.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"DELETE FROM core_convert_lot_detail"
|
||
f" WHERE convert_group_id LIKE '{_GROUP_LIKE}'"
|
||
)
|
||
)
|
||
for sql in _CORE_CLEANUP_BY_CUSTOMER:
|
||
conn.execute(text(sql), params)
|
||
for sql in _CORE_CLEANUP_BY_PRODUCT:
|
||
conn.execute(text(sql))
|
||
conn.execute(text("DELETE FROM core_customer WHERE customer_id LIKE :cl"), params)
|
||
with agent.begin() as conn:
|
||
conn.execute(
|
||
text(f"DELETE FROM risk_convert_detail WHERE convert_group_id LIKE '{_GROUP_LIKE}'")
|
||
)
|
||
for table in ("risk_alert", "risk_suitability_log", "audit_log"):
|
||
conn.execute(text(f"DELETE FROM {table} WHERE customer_id LIKE :cl"), params)
|
||
|
||
|
||
@pytest.fixture()
|
||
def conc_env(risk_demo_env, monkeypatch):
|
||
"""真库并发环境：admin 引擎（teardown 需 DELETE）+ 专用大池引擎 + id 前缀注入。
|
||
|
||
种子由用例自己 `_seed`（不同用例份额池不同），fixture 只负责清场与收尾。
|
||
"""
|
||
core_admin = risk_demo_env["core"]
|
||
agent_admin = risk_demo_env["agent"]
|
||
core_ro = _big_pool_engine(settings.mysql_core_database, "ro")
|
||
core_rw = _big_pool_engine(settings.mysql_core_database, "rw")
|
||
agent_rw = _big_pool_engine(settings.mysql_database, "rw")
|
||
monkeypatch.setattr(cs, "_new_id", _new_id)
|
||
_cleanup(core_admin, agent_admin)
|
||
try:
|
||
yield {
|
||
"core": core_admin,
|
||
"agent": agent_admin,
|
||
"core_ro": core_ro,
|
||
"core_rw": core_rw,
|
||
"agent_rw": agent_rw,
|
||
}
|
||
finally:
|
||
_cleanup(core_admin, agent_admin)
|
||
for eng in (core_ro, core_rw, agent_rw):
|
||
eng.dispose()
|
||
|
||
|
||
# ── T+1 两段调用（受理 / 确认；生产账号 + 大池引擎）────────────────────
|
||
_NOW = datetime.combine(date.today(), dtime(10, 0))
|
||
|
||
|
||
def _accept(env, qty: str, cid: str | None = None, customer: str = CUSTOMER) -> dict:
|
||
"""T 日受理（不扣份额；生产口径，仅 id 工厂被替换）。"""
|
||
return accept_convert(
|
||
{
|
||
"customer_id": customer,
|
||
"from_product_id": PROD_OUT,
|
||
"to_product_id": PROD_IN,
|
||
"qty": qty,
|
||
"client_request_id": cid,
|
||
},
|
||
core_ro=CoreReadOnlyRepository(engine=env["core_ro"]),
|
||
risk_repo=RiskRepository(engine=env["agent_rw"]),
|
||
convert_repo=ConvertRepository(engine=env["agent_rw"]),
|
||
request_repo=ConvertRequestRepository(engine=env["core_rw"]),
|
||
now=_NOW,
|
||
)
|
||
|
||
|
||
def _confirm(env, gid: str, hook=_noop_engine) -> dict:
|
||
"""T+1 确认一笔（as_of=受理日，补跑语义；确认事务含 FIFO 扣减哨兵）。
|
||
|
||
`id_factory` 显式注入：`confirm_one` 的 trade_id 走 `format.new_id`
|
||
（与 `convert_service._new_id` 是两个符号），monkeypatch 罩不到 → 必须传参。
|
||
"""
|
||
return confirm_one(
|
||
gid,
|
||
core_ro=CoreReadOnlyRepository(engine=env["core_ro"]),
|
||
risk_repo=RiskRepository(engine=env["agent_rw"]),
|
||
convert_repo=ConvertRepository(engine=env["agent_rw"]),
|
||
request_repo=ConvertRequestRepository(engine=env["core_rw"]),
|
||
core_writer=ConvertCoreRepository(engine=env["core_rw"]),
|
||
thresholds=RiskThresholds.from_settings(),
|
||
now=_NOW,
|
||
as_of=_accept_date(env["core"]),
|
||
engine_hook=hook,
|
||
id_factory=_new_id,
|
||
)
|
||
|
||
|
||
class Outcome:
|
||
"""一个并发请求的终态（是否成功、重试了几次、最终异常）。"""
|
||
|
||
__slots__ = ("ok", "retries", "error", "group_id")
|
||
|
||
def __init__(self, ok, retries, error=None, group_id=None):
|
||
self.ok = ok
|
||
self.retries = retries
|
||
self.error = error
|
||
self.group_id = group_id
|
||
|
||
|
||
def _confirm_with_backoff(env, gid: str) -> Outcome:
|
||
"""按架构 §8.3 的建议间隔退避重试确认（≤3 次；1213 类瞬时失败）。"""
|
||
attempt = 0
|
||
while True:
|
||
try:
|
||
resp = _confirm(env, gid)
|
||
except _RETRYABLE as exc:
|
||
if attempt >= len(BACKOFF_MS):
|
||
return Outcome(False, attempt, type(exc).__name__)
|
||
time.sleep(BACKOFF_MS[attempt] / 1000.0)
|
||
attempt += 1
|
||
continue
|
||
except Exception as exc: # noqa: BLE001
|
||
return Outcome(False, attempt, type(exc).__name__)
|
||
return Outcome(
|
||
resp["status"] == "confirmed",
|
||
attempt,
|
||
None if resp["status"] == "confirmed" else resp["status"],
|
||
group_id=gid,
|
||
)
|
||
|
||
|
||
def _concurrently(fn, n: int, *, stagger_ms: float = 0.0) -> list:
|
||
"""n 个线程同时起跑；`stagger_ms` 为逐线程错开步长（0 = 全同时）。
|
||
|
||
起跑线用 `threading.Barrier` 对齐，避免"线程创建开销"把并发抹平。
|
||
"""
|
||
barrier = threading.Barrier(n)
|
||
|
||
def _worker(i: int):
|
||
barrier.wait()
|
||
if stagger_ms:
|
||
time.sleep(stagger_ms * i / 1000.0)
|
||
return fn(i)
|
||
|
||
with ThreadPoolExecutor(max_workers=n) as pool:
|
||
return list(pool.map(_worker, range(n)))
|
||
|
||
|
||
# ── 度量（全部直接读库，不看响应自证）────────────────────────────────
|
||
def _remain_total(core) -> Decimal:
|
||
with core.connect() as conn:
|
||
return Decimal(
|
||
str(
|
||
conn.execute(
|
||
text(
|
||
"SELECT COALESCE(SUM(remain_qty), 0) FROM core_share_lot"
|
||
" WHERE customer_id = :c AND product_id = :p"
|
||
),
|
||
{"c": CUSTOMER, "p": PROD_OUT},
|
||
).scalar_one()
|
||
)
|
||
)
|
||
|
||
|
||
def _deducted_total(core) -> Decimal:
|
||
"""实际成交的转出份额 = 全部 `redeem` 流水的 qty 之和（转出端=客户被扣的部分）。"""
|
||
with core.connect() as conn:
|
||
return Decimal(
|
||
str(
|
||
conn.execute(
|
||
text(
|
||
"SELECT COALESCE(SUM(qty), 0) FROM core_trade"
|
||
" WHERE customer_id = :c AND trade_type = 'redeem'"
|
||
),
|
||
{"c": CUSTOMER},
|
||
).scalar_one()
|
||
)
|
||
)
|
||
|
||
|
||
def _trades(core) -> list[dict]:
|
||
with core.connect() as conn:
|
||
return [
|
||
dict(r)
|
||
for r in conn.execute(
|
||
text("SELECT * FROM core_trade WHERE customer_id = :c ORDER BY trade_id"),
|
||
{"c": CUSTOMER},
|
||
).mappings()
|
||
]
|
||
|
||
|
||
def _group_counts(core) -> dict[str, int]:
|
||
"""convert_group_id → 该组流水条数（正常恒为 2）。"""
|
||
counts: dict[str, int] = {}
|
||
for row in _trades(core):
|
||
gid = row["convert_group_id"]
|
||
counts[gid] = counts.get(gid, 0) + 1
|
||
return counts
|
||
|
||
|
||
def _request_statuses(core) -> dict[str, int]:
|
||
"""受理单 6 态分布（T+1 权威状态机）。"""
|
||
with core.connect() as conn:
|
||
rows = conn.execute(
|
||
text(
|
||
"SELECT status, COUNT(*) AS n FROM core_convert_request"
|
||
" WHERE customer_id = :c GROUP BY status"
|
||
),
|
||
{"c": CUSTOMER},
|
||
).mappings()
|
||
return {r["status"]: int(r["n"]) for r in rows}
|
||
|
||
|
||
def _assert_no_oversell(core, pool: Decimal) -> None:
|
||
"""硬不变量：不超卖 / 守恒 / 每组恰好两条流水（无重复组）。"""
|
||
deducted = _deducted_total(core)
|
||
remain = _remain_total(core)
|
||
assert deducted <= pool, f"超卖：扣减 {deducted} > 池 {pool}"
|
||
assert remain == pool - deducted, f"不守恒：余 {remain} ≠ 池 {pool} − 扣 {deducted}"
|
||
counts = _group_counts(core)
|
||
assert set(counts.values()) <= {2}, f"存在流水条数 ≠2 的组：{counts}"
|
||
|
||
|
||
# ── N1. 同键并发受理：只允许落一张受理单（uk_idem + 客户级锁）──────────
|
||
def test_same_client_request_id_concurrent_produces_single_acceptance(conc_env):
|
||
"""同 `client_request_id` 并发**受理** → core_convert_request 恰 1 行。
|
||
|
||
T+1 语义（取代 v1.0「同键并发只扣一次」）：受理不扣份额，幂等收敛目标是
|
||
**受理单唯一** —— 客户级锁串行化 + `uk_idem` 兜底让路，终态允许
|
||
「幂等命中（同 gid）」或「让路 202」，绝不允许第二张受理单、也绝无 5xx。
|
||
（uk_idem 竞态兜底的单测锚点：`test_accept_uk_idem_race_falls_back_to_idempotent`）
|
||
"""
|
||
core = conc_env["core"]
|
||
_seed(core, ["50000"])
|
||
n = 8
|
||
cid = f"CONC-SAME-{uuid4().hex[:8]}"
|
||
|
||
def _one(_i: int) -> dict:
|
||
try:
|
||
return _accept(conc_env, "120", cid)
|
||
except Exception as exc: # noqa: BLE001
|
||
return {"_error": type(exc).__name__}
|
||
|
||
outcomes = _concurrently(_one, n, stagger_ms=1.5)
|
||
errs = [o["_error"] for o in outcomes if "_error" in o]
|
||
gids = {o["convert_group_id"] for o in outcomes if "_error" not in o}
|
||
print(
|
||
f"\n[同键并发受理 {n} 路] 受理单 {len(_request_statuses(core))} 张 · "
|
||
f"gid 去重 {len(gids)} · 错误 {errs}"
|
||
)
|
||
|
||
# 唯一硬约束：恰一张受理单（uk_idem），且无 5xx
|
||
statuses = _request_statuses(core)
|
||
assert sum(statuses.values()) == 1, f"同键并发落了 {sum(statuses.values())} 张受理单"
|
||
assert len(gids) <= 1, f"同键应幂等命中同一 gid：{gids}"
|
||
assert not errs, f"同键并发出现了异常：{errs}"
|
||
|
||
|
||
# ── N2. 并发补偿只落一次（T-12 的并发面 · T+1 造数）────────────────────
|
||
def test_concurrent_compensation_writes_detail_once(conc_env, monkeypatch):
|
||
"""待补偿现场并发补偿（`convert:rerun:` 锁）→ 详情只补一次、只有一组流水。
|
||
|
||
T+1 造数：真受理 → 真确认（Core 三件套 + 镜像 completed + 引擎 hook 静默）
|
||
→ 镜像 UPDATE 回 `failed` 制造「确认段第 ⑧ 步失败」现场（受理单已 confirmed，
|
||
补偿前置满足）。
|
||
"""
|
||
core = conc_env["core"]
|
||
agent = conc_env["agent"]
|
||
_seed(core, ["50000"])
|
||
|
||
resp = _accept(conc_env, "120", f"CONC-CMP-{uuid4().hex[:8]}")
|
||
gid = resp["convert_group_id"]
|
||
out = _confirm(conc_env, gid)
|
||
assert out["status"] == "confirmed"
|
||
# 制造待补偿现场：镜像回退 failed（Core 权威不动）
|
||
with agent.begin() as conn:
|
||
conn.execute(
|
||
text("UPDATE risk_convert_detail SET status = 'failed' WHERE convert_group_id = :g"),
|
||
{"g": gid},
|
||
)
|
||
|
||
def _comp(_i: int) -> str:
|
||
svc = dict(
|
||
core_ro=CoreReadOnlyRepository(engine=conc_env["core_ro"]),
|
||
risk_repo=RiskRepository(engine=conc_env["agent_rw"]),
|
||
convert_repo=ConvertRepository(engine=conc_env["agent_rw"]),
|
||
)
|
||
result = compensate_convert(gid, thresholds=RiskThresholds.from_settings(), now=_NOW, **svc)
|
||
return result["state"]
|
||
|
||
states = _concurrently(_comp, 6, stagger_ms=1.0)
|
||
print(f"\n[并发补偿 6 路] 状态分布 {states}")
|
||
assert states.count("rebuilt") == 1, f"应恰好 1 路真补写，实际 {states}"
|
||
with agent.connect() as conn:
|
||
rows = [
|
||
dict(r)
|
||
for r in conn.execute(
|
||
text("SELECT status FROM risk_convert_detail WHERE convert_group_id = :g"),
|
||
{"g": gid},
|
||
).mappings()
|
||
]
|
||
assert rows == [{"status": "completed"}], f"详情应只有一行 completed：{rows}"
|
||
assert len(_group_counts(core)) == 1 and len(_trades(core)) == 2
|
||
# 补偿不动 Core 权威：受理单保持 confirmed、流水不增
|
||
with core.connect() as conn:
|
||
status = conn.execute(
|
||
text("SELECT status FROM core_convert_request WHERE convert_group_id = :g"),
|
||
{"g": gid},
|
||
).scalar_one()
|
||
assert status == "confirmed"
|
||
|
||
|
||
# ── S1. 紧池（需求=供给）：受理 100% + 串行确认 100%（验收 31 载体）────
|
||
@requires_stress
|
||
def test_tight_pool_accept_all_then_confirm_100pct(conc_env):
|
||
"""50 并发受理各 1000、池恰好 50000（需求=供给）→ **受理 100% + 确认 100%**。
|
||
|
||
这是 PRD 验收 31 的 T+1 载体：v1.0 同场景实时扣减只到 **80%**（40/50，
|
||
批次碎片争抢，见交接文档 §B.6.6）——根因是「确认前就扣份额」。T+1 受理
|
||
不扣份额（R-3 占用校验：50×1000 恰好 ≤ 池，**全部 accepted 是确定性结果**），
|
||
确认**串行**（FR-C23）无争抢 → 逐笔 FIFO 全额成交 → 100%。
|
||
|
||
硬门禁：① 受理恰 50 张 accepted；② 确认恰 50 张 confirmed（100%）；
|
||
③ Σ扣减 = 池 = 50000（需求=供给全额消化）；④ 守恒 + 无超卖。
|
||
"""
|
||
core = conc_env["core"]
|
||
pool = _seed(core, ["10000"] * 5)
|
||
n = 50
|
||
|
||
outcomes = _concurrently(lambda i: _accept(conc_env, "1000"), n, stagger_ms=0.5)
|
||
accepted = [o for o in outcomes if o.get("accepted")]
|
||
print(
|
||
f"\n[紧池·受理] accepted {len(accepted)}/{n} · 其余 "
|
||
f"{[o.get('block_reason') or o.get('_error') for o in outcomes if not o.get('accepted')]}"
|
||
)
|
||
assert len(accepted) == n, f"受理应 100%（设计目标），实际 {len(accepted)}"
|
||
statuses = _request_statuses(core)
|
||
assert statuses.get("accepted") == n, f"受理单应全为 accepted：{statuses}"
|
||
|
||
# T+1 确认：批处理串行（as_of = 受理日，补跑语义）
|
||
accept_day = _accept_date(core)
|
||
batch = confirm_batch(
|
||
as_of=accept_day,
|
||
core_ro=CoreReadOnlyRepository(engine=conc_env["core_ro"]),
|
||
risk_repo=RiskRepository(engine=conc_env["agent_rw"]),
|
||
convert_repo=ConvertRepository(engine=conc_env["agent_rw"]),
|
||
request_repo=ConvertRequestRepository(engine=conc_env["core_rw"]),
|
||
core_writer=ConvertCoreRepository(engine=conc_env["core_rw"]),
|
||
thresholds=RiskThresholds.from_settings(),
|
||
now=_NOW,
|
||
engine_hook=_noop_engine,
|
||
id_factory=_new_id,
|
||
)
|
||
print(
|
||
f"\n[紧池·串行确认] scanned={batch['scanned']} confirmed={batch['confirmed']} "
|
||
f"rejected={batch.get('rejected', 0)} nav_pending={batch.get('nav_pending', 0)}"
|
||
)
|
||
assert batch["scanned"] == n, f"批处理应捞到全部 {n} 张"
|
||
assert batch["confirmed"] == n, f"串行确认应 100%（验收 31 目标），实际 {batch}"
|
||
statuses = _request_statuses(core)
|
||
assert statuses.get("confirmed") == n, f"受理单应全为 confirmed：{statuses}"
|
||
# 需求=供给 → 池被全额消化且守恒（无超卖硬门禁）
|
||
assert _deducted_total(core) == pool, f"扣减 {_deducted_total(core)} 应恰等于池 {pool}"
|
||
_assert_no_oversell(core, pool)
|
||
assert len({g for g in _group_counts(core)}) == n, "50 张单应有 50 个独立组"
|
||
|
||
|
||
# ── S2. 并发确认争抢同一池：不许超卖（评审 Q6 断言的 T+1 承接）──────────
|
||
@requires_stress
|
||
def test_concurrent_confirms_never_oversell(conc_env):
|
||
"""并发受理允许**超授**（R-3 占用读数滞后的 TOCTOU），并发确认兜底 → 不许超卖。
|
||
|
||
50 线程 × 各受理 2000（池 50000）：受理段占用校验读数滞后时可能出现
|
||
Σ占用 > 池的超授 —— T+1 的资金安全不依赖受理段，而在**确认段扣减哨兵**
|
||
（`remain_qty >= :q` 条件 UPDATE + 死锁重试）：份额不足时逐笔部分成交
|
||
（R-10）或 `rejected`，Σ实际扣减绝不越过池。
|
||
|
||
典型失效形态：若确认事务丢掉扣减哨兵，50 笔会**全部**全额成交 →
|
||
扣减 100000 > 池 50000 → 本用例红。
|
||
"""
|
||
core = conc_env["core"]
|
||
pool = _seed(core, ["10000"] * 5)
|
||
n = 50
|
||
|
||
# 受理段：并发窗口内 R-3 占用读数滞后 → 允许超授（Σ占用 > 池）；
|
||
# 校验命中时直接 `InsufficientShares`（受理段 4xx，异常而非 blocked）。
|
||
def _try_accept(_i: int) -> dict:
|
||
try:
|
||
return _accept(conc_env, "2000")
|
||
except Exception as exc: # noqa: BLE001 - 受理失败本身是合法观测面
|
||
return {"_error": type(exc).__name__}
|
||
|
||
accepted = _concurrently(_try_accept, n, stagger_ms=0.5)
|
||
gids = [o["convert_group_id"] for o in accepted if o.get("accepted")]
|
||
accept_errors = {o["_error"] for o in accepted if "_error" in o}
|
||
print(
|
||
f"\n[并发确认·不超卖] 受理 {len(gids)}/{n}（允许超授）· 受理失败类型 {accept_errors} · "
|
||
f"受理单分布 {_request_statuses(core)}"
|
||
)
|
||
assert gids, "至少应有部分受理成功"
|
||
|
||
# 确认段：并发 confirm_one 争抢扣减 → 兜底不允许超卖
|
||
confirmed = _concurrently(lambda i: _confirm_with_backoff(conc_env, gids[i]), len(gids))
|
||
ok = [o for o in confirmed if o.ok]
|
||
# 合法失败面（都是**零扣减**的让路终态，不碰资金安全）：
|
||
# - `rejected`：份额被抢空 → 复核拒绝（占用释放）；
|
||
# - `LotConflict`：退避耗尽仍撞批次 → 让路。并发 confirm_one 本身是**超纲场景**
|
||
# （生产由批处理锁保证串行，FR-C23），让路单受理单保持 accepted，
|
||
# 占用保持、下轮批处理再确认 —— 与 v1.0 用例 1 允许 LotConflict 同口径。
|
||
legit_fail = {"rejected", "LotConflict"}
|
||
unexpected = {o.error for o in confirmed if not o.ok and o.error not in legit_fail}
|
||
print(
|
||
f"\n[并发确认·不超卖] confirmed {len(ok)}/{len(gids)} · 让路/拒绝 "
|
||
f"{sorted({o.error for o in confirmed if not o.ok})} · "
|
||
f"未映射异常 {unexpected} · 扣减 {_deducted_total(core)} / 池 {pool}"
|
||
)
|
||
# ① 硬不变量：不超卖 / 守恒 / 每组恰好两条流水
|
||
_assert_no_oversell(core, pool)
|
||
# ② 失败面：终态只能是 confirmed/rejected/LotConflict（部分成交含在 confirmed 内），
|
||
# 不得有未映射异常直穿
|
||
assert not unexpected, f"确认出现未映射异常：{unexpected}"
|
||
# ③ 让路不吞单：没确认成的单，受理单必须保持 accepted（占用保持，下轮可确认）
|
||
statuses = _request_statuses(core)
|
||
assert statuses.get("accepted", 0) == len(gids) - len(ok), (
|
||
f"让路单受理单状态异常：{statuses}（成功 {len(ok)}/{len(gids)}）"
|
||
)
|
||
# ④ 反"假绿"地板：confirmed 塌穿 20 说明扣减路径退化（哨兵过严/锁失效）
|
||
assert len(ok) >= 20, f"确认仅 {len(ok)} 笔，扣减路径疑似退化"
|
||
|
||
|
||
# ── S3. 多客户并发确认：互不阻塞，但会出 1213 死锁（双层重试收敛）───────
|
||
@requires_stress
|
||
def test_concurrent_different_customers_need_deadlock_retry(conc_env):
|
||
"""8 个客户**并发确认**同一转出产品 → 全部成功，但**必须靠重试死锁**。
|
||
|
||
**实测发现（2026-09-10，v1.0 阶段一）**：跨客户并发在 `core_holding`/
|
||
`core_share_lot` 的唯一索引上会触发 InnoDB **1213 死锁**（RR 隔离级别下
|
||
INSERT/UPDATE 的插入意向锁与间隙锁互斥）。T+1 下同一争抢面在**确认事务**
|
||
（`apply_convert`，1213 自动重试整事务，settings 可配）；本用例保留 §8.3
|
||
的**外部退避兜底**，双层重试叠加下 8 客户并发确认仍须全部成功、份额守恒、
|
||
group_id 两两不同。受理段先串行完成（不扣份额、无争抢面）。
|
||
"""
|
||
core = conc_env["core"]
|
||
_seed(core, ["5000"] * 2)
|
||
|
||
others = [f"CUST-CONCTEST-{i}" for i in range(1, 8)]
|
||
with core.begin() as conn:
|
||
for c in others:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer (customer_id, display_name, open_date)"
|
||
" VALUES (:c, 'T13并发客户', CURDATE())"
|
||
),
|
||
{"c": c},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at,"
|
||
" expires_at) VALUES (:c, 'C5', NOW(), DATE_ADD(NOW(), INTERVAL 300 DAY))"
|
||
),
|
||
{"c": c},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty,"
|
||
" remain_qty, nav, confirmed_at)"
|
||
" VALUES (:l, :c, :p, 5000, 5000, :n, DATE_SUB(NOW(), INTERVAL 100 DAY))"
|
||
),
|
||
{"l": f"LOT-CONCTEST-OTH-{c}", "c": c, "p": PROD_OUT, "n": str(OUT_NAV)},
|
||
)
|
||
|
||
customers = [CUSTOMER, *others]
|
||
# 受理先串行完成（不扣份额；其他客户的受理走同款服务路径）
|
||
gids = {}
|
||
for c in customers:
|
||
resp = _accept(conc_env, "1000", customer=c)
|
||
assert resp.get("accepted"), f"{c} 受理失败：{resp}"
|
||
gids[c] = resp["convert_group_id"]
|
||
|
||
def _one(i: int) -> Outcome:
|
||
return _confirm_with_backoff(conc_env, gids[customers[i]])
|
||
|
||
outcomes = _concurrently(_one, len(customers))
|
||
ok = [o for o in outcomes if o.ok]
|
||
deadlock_retries = sum(o.retries for o in outcomes)
|
||
print(
|
||
f"\n[8 客户并发确认] 成功 {len(ok)}/8 · 死锁重试累计 {deadlock_retries} 次 · "
|
||
f"失败 {[o.error for o in outcomes if not o.ok]}"
|
||
)
|
||
assert len(ok) == 8, f"退避重试后应全部成功：{[o.error for o in outcomes if not o.ok]}"
|
||
with core.connect() as conn:
|
||
total = Decimal(
|
||
str(
|
||
conn.execute(
|
||
text(
|
||
"SELECT COALESCE(SUM(remain_qty), 0) FROM core_share_lot"
|
||
" WHERE product_id = :p AND lot_id LIKE 'LOT-CONCTEST%'"
|
||
),
|
||
{"p": PROD_OUT},
|
||
).scalar_one()
|
||
)
|
||
)
|
||
# 8 个客户各转出 1000；主客户有 2 批 × 5000 = 10000，其余 7 个各 5000
|
||
# → 初始合计 45000，扣 8000 → 37000
|
||
expected = Decimal("45000") - Decimal("1000") * len(customers)
|
||
assert total == expected, f"剩余合计应为 {expected}，实际 {total}"
|
||
assert len({o.group_id for o in ok}) == 8, "8 个客户的 group_id 必须两两不同"
|
||
|
||
|
||
# ── S4. 性能实测（PRD §9 第 18 条 · T+1 两段）──────────────────────────
|
||
class _TimingWriter:
|
||
"""`ConvertCoreRepository` 代理：只测**确认事务**耗时，不改任何行为。"""
|
||
|
||
def __init__(self, inner):
|
||
self._inner = inner
|
||
self.samples: list[float] = []
|
||
|
||
def apply_convert(self, req):
|
||
t0 = time.perf_counter()
|
||
try:
|
||
return self._inner.apply_convert(req)
|
||
finally:
|
||
self.samples.append((time.perf_counter() - t0) * 1000.0)
|
||
|
||
|
||
def _pct(values: list[float], p: float) -> float:
|
||
ordered = sorted(values)
|
||
idx = min(len(ordered) - 1, max(0, int(round((p / 100.0) * len(ordered) + 0.5)) - 1))
|
||
return ordered[idx]
|
||
|
||
|
||
@requires_stress
|
||
def test_performance_probe_accept_confirm_end_to_end(conc_env):
|
||
"""实测：**受理事务 + 确认事务 + 端到端两段**耗时（PRD §9 第 18 条补录来源）。
|
||
|
||
测量条件：本机 MySQL 8.0.46（127.0.0.1）、模拟库、**单线程顺序**、
|
||
份额池充足的稳态（每笔 1 份，避免 FIFO 跨批波动）；端到端含空引擎 hook
|
||
（真实引擎的额外开销在 `test_convert_integration` 另行体现）。
|
||
端到端 = 受理（校验+落单+镜像+审计）+ 确认（折算+扣减+流水+镜像+审计）两段之和，
|
||
PRD 验收 18 的硬指标是 **< 2s**。
|
||
"""
|
||
core = conc_env["core"]
|
||
n = 60
|
||
_seed(core, [str(n)] * 1)
|
||
|
||
writer = _TimingWriter(ConvertCoreRepository(engine=conc_env["core_rw"]))
|
||
accept_ms: list[float] = []
|
||
confirm_ms: list[float] = []
|
||
for i in range(n):
|
||
t0 = time.perf_counter()
|
||
resp = _accept(conc_env, "1", f"CONC-PERF-{i}")
|
||
accept_ms.append((time.perf_counter() - t0) * 1000.0)
|
||
gid = resp["convert_group_id"]
|
||
|
||
t1 = time.perf_counter()
|
||
out = confirm_one(
|
||
gid,
|
||
core_ro=CoreReadOnlyRepository(engine=conc_env["core_ro"]),
|
||
risk_repo=RiskRepository(engine=conc_env["agent_rw"]),
|
||
convert_repo=ConvertRepository(engine=conc_env["agent_rw"]),
|
||
request_repo=ConvertRequestRepository(engine=conc_env["core_rw"]),
|
||
core_writer=writer,
|
||
thresholds=RiskThresholds.from_settings(),
|
||
now=_NOW,
|
||
as_of=_accept_date(core),
|
||
engine_hook=_noop_engine,
|
||
id_factory=_new_id,
|
||
)
|
||
confirm_ms.append((time.perf_counter() - t1) * 1000.0)
|
||
assert out["status"] == "confirmed", out
|
||
|
||
# 首个样本含连接池冷启动，剔除（否则把"建连"算进"业务耗时"）
|
||
stage_confirm = writer.samples[1:]
|
||
acc = accept_ms[1:]
|
||
cfm = confirm_ms[1:]
|
||
e2e = [a + c for a, c in zip(acc, cfm)]
|
||
print(
|
||
f"\n[性能实测 n={len(e2e)}] 受理 ms: P50={_pct(acc, 50):.1f} P95={_pct(acc, 95):.1f} "
|
||
f"max={max(acc):.1f} mean={mean(acc):.1f}"
|
||
f"\n 确认事务 ms: P50={_pct(stage_confirm, 50):.1f} "
|
||
f"P95={_pct(stage_confirm, 95):.1f} max={max(stage_confirm):.1f} mean={mean(stage_confirm):.1f}"
|
||
f"\n 确认全程 ms: P50={_pct(cfm, 50):.1f} P95={_pct(cfm, 95):.1f} max={max(cfm):.1f}"
|
||
f"\n 端到端(受理+确认) ms: P50={_pct(e2e, 50):.1f} P95={_pct(e2e, 95):.1f} "
|
||
f"max={max(e2e):.1f} mean={mean(e2e):.1f}"
|
||
)
|
||
# PRD §9 第 18 条的验收硬指标是端到端 < 2s
|
||
assert max(e2e) < 2000, f"端到端最大 {max(e2e):.1f}ms 超 PRD §9 第 18 条的 2s"
|