From a337cca31ae3088b75fc43b0231888f42f59e41e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=8D=BF=E4=BA=91=E7=A7=8B=E6=9C=88?= <15273589815@163.com> Date: Mon, 14 Sep 2026 20:19:55 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20T002=20=E4=B8=8B=E5=8D=95?= =?UTF-8?q?=E7=9A=84=E4=B8=A4=E4=B8=AA=20P0=EF=BC=9A=E6=97=A0=E5=B9=82?= =?UTF-8?q?=E7=AD=89=E4=BF=9D=E6=8A=A4=E3=80=81=E8=AF=BB=E8=B4=A6=E6=88=B7?= =?UTF-8?q?/=E6=8C=81=E4=BB=93=E6=97=A0=E8=A1=8C=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## P0-1 下单没有幂等保护（实测复现过重复扣款） `app/api/controllers/trading.py` 的 T002 此前**没有任何幂等保护** —— 全仓 7 个 controller 共 29 处声明了 `Idempotency-Key`，**唯独 trading.py 一处都没有**， 而 `docs/05` §5.2 要求写端点必须带。 **修复前实测**（同一 key 连发两次）： 两个不同 order_no：SO...FD48A488 / SO...2FB182C0 委托 +2、可用现金 -910.50（2×455.25）、510300 持仓 3500 -> 3700 ⇒ 重复扣款 + 重复建仓 **改法**：接 `ApiTransactionService.execute_in` —— 业务写入与幂等回执**同一个事务** （`docs/05` §5.2），同键同 body 的第二次请求直接回放上次响应。 - `trade_service.submit_order` 新增 `commit: bool = True`：包装层调用时传 `commit=False`，由 `execute_in` 统一提交。**不做这一步就会"内层提交外层事务"**， 幂等回执与业务写入分处两个事务，回执写失败时业务已落库，重放失去意义。 - 响应经 `model_dump(mode="json")` 统一 JSON 化后再 `model_validate` 还原 —— 保证"首次"与"重放"两条路径返回**同一形状**，且对外结构不变 （金额仍是字符串化 Decimal）。 **修复后实测**（同一 key 连发两次）： 两次 order_no 完全相同：SO202609141216512C7166D4 两次响应逐字段相同（真正的回放） 可用现金 -455.25（单次）、510300 持仓 +100（单次） T003 核对：43 条委托里只多出 1 条 ## P0-3 下单读账户/持仓无行锁（并发可扣穿余额） `trade_service.py` 全文 **零 `with_for_update`**，而项目其它 15 个 service 共 38 处 用了它 —— 规范早已建立，这里是遗漏。 **改法**：`_load_account` / `_load_holding` 增加 `for_update` 开关（默认关， 只读路径不加锁、不牺牲并发），`submit_order` 以 `for_update=True` 调用。 **加锁顺序固定「账户 → 持仓」**：并发事务按同一顺序取锁才不会成环， 这一点写在两处 docstring 里，改顺序前必须先想清楚。 **实测**（真并发 2 笔，各 45520 元，合计 91040 > 余额 49103）： 第 1 笔：201 成交 SO...59568811 第 2 笔：422 INSUFFICIENT_FUNDS「可用余额 3578.91 不足」 成交 1/2，最终余额 3578.91 >= 0 **关键证据**：被拒那笔读到的是 **3578.91（已扣减后）**而不是初始的 49103.46 —— 证明两个事务被行锁串行化了。无锁时两笔都会读到 49103.46 而双双通过， 余额会变成 -41936。 ## 实测汇总 - `pytest tests/unit tests/contract` -> **1427 passed, 2 skipped, 2 failed** （那 2 个是既有的：投顾页面被替换、docs/05 §19 分组行，均与本提交无关） - `ruff check` -> All checks passed - 两次实测的完整证据见上；两份验证脚本在 %TEMP%（未入库） ## 顺带修正一个我自己脚本的 bug 验证脚本里用 `GET /users/me/orders?limit=200` 读委托数，而 T003 的 `limit` 上限是 **100**（`Query(le=100)`）→ 422 → 读到 0 条，一度让我误判"委托没增加"。 改用 `limit=100` 后确认委托确实只 +1。 --- app/api/controllers/trading.py | 41 ++++++++++++++++--- app/service/trade_service.py | 72 +++++++++++++++++++++++++--------- 2 files changed, 89 insertions(+), 24 deletions(-) diff --git a/app/api/controllers/trading.py b/app/api/controllers/trading.py index afac8e0..3451e60 100644 --- a/app/api/controllers/trading.py +++ b/app/api/controllers/trading.py @@ -22,20 +22,24 @@ from __future__ import annotations -from fastapi import APIRouter, Depends, Query, status +from fastapi import APIRouter, Depends, Header, Query, status from sqlalchemy.ext.asyncio import AsyncSession from app.api.dependencies.auth import build_request_context from app.api.dependencies.database import get_session -from app.api.schemas.trading import OrderCreateRequest +from app.api.schemas.trading import OrderCreateRequest, OrderCreateResponse from app.api.views.envelope import envelope, list_envelope from app.core.contracts import RequestContext +from app.service.api_transaction_service import ApiTransactionService from app.service.authorization_service import AuthorizationService from app.service.suitability_service import SuitabilityService from app.service.trade_service import TradeService router = APIRouter(prefix="/api/v1/users/me", tags=["trading"]) +#: T002 的幂等作用域。与权限码同名，便于审计时一眼对上是哪个写操作。 +ORDER_SCOPE = "trade:order:create" + def _service(session: AsyncSession, context: RequestContext) -> TradeService: return TradeService(session, suitability_evaluator=SuitabilityService()) @@ -58,15 +62,42 @@ async def get_account_dashboard( # T002 提交委托（市价立即成交） +# +# ⚠️ 幂等：本端点此前**没有任何幂等保护** —— 全仓 7 个 controller 共 29 处声明了 +# `Idempotency-Key`，唯独 trading.py 一处都没有（`docs/05` §5.2 要求写端点必须带）。 +# 后果是**实测复现**过的：同一笔意愿在网关超时后被客户端按标准重试重发， +# 两次请求各自生成新的 `order_no`、各扣一次款、各建一次仓 +# （实测：可用现金 -910.50 = 2×455.25，持仓 3500 -> 3700）。 +# +# 现在接 `ApiTransactionService.execute_in`：业务写入与幂等回执**同一个事务**， +# 同键同 body 的第二次请求直接回放上次响应，不再重复成交。 @router.post("/orders", status_code=status.HTTP_201_CREATED) async def submit_order( payload: OrderCreateRequest, context: RequestContext = Depends(build_request_context), # noqa: B008 session: AsyncSession = Depends(get_session), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), ) -> dict[str, object]: - await _authorize(context, "trade:order:create") - data = await _service(session, context).submit_order(payload, context) - return envelope(data, context) + await _authorize(context, ORDER_SCOPE) + + async def action(inner: AsyncSession) -> dict[str, object]: + # `commit=False`：提交由 `execute_in` 统一做，业务写入与幂等回执同事务。 + response = await _service(inner, context).submit_order(payload, context, commit=False) + # 统一 JSON 化。`execute_in` 回放的是**已落库的 dict**，首次返回值必须是 + # 同一形状，否则"重放"与"首次"两种路径返回的字段类型会不一致。 + return response.model_dump(mode="json") + + data = await ApiTransactionService().execute_in( + session, + context, + ORDER_SCOPE, + key, + payload.model_dump(mode="json"), + action, + ) + # 还原成响应模型再套信封：保持与改动前**完全一致**的响应结构 + # （金额仍是字符串化 Decimal，见 docs/05 §3.3）。 + return envelope(OrderCreateResponse.model_validate(data), context) # T003 委托列表 diff --git a/app/service/trade_service.py b/app/service/trade_service.py index c600c73..8cf1f3b 100644 --- a/app/service/trade_service.py +++ b/app/service/trade_service.py @@ -220,13 +220,26 @@ class TradeService: # ---------- 账户 / 持仓 ---------- - async def _load_account(self, customer_id: int | str) -> FundSimAccount: + async def _load_account( + self, customer_id: int | str, *, for_update: bool = False + ) -> FundSimAccount: + """加载虚拟账户。 + + ⚠️ `for_update=True` 时加 `SELECT … FOR UPDATE` 行锁，**下单路径必须用**： + 否则同一客户的并发下单会各自读到旧余额、各自判"够不够"、各自扣减， + 结果是可用现金被扣成负数。 + + **加锁顺序固定为 账户 → 持仓**（见 `submit_order`）：并发事务只要按同一顺序 + 取锁就不会互相等待成环，这是避免死锁的关键，改顺序前请先想清楚。 + 只读路径（看板、持仓列表、流水）不加锁 —— 它们不修改余额，加锁只会降低并发。 + """ customer_id_int = int(customer_id) if isinstance(customer_id, str) else customer_id - account = ( - await self._session.execute( - select(FundSimAccount).where(FundSimAccount.customer_id == customer_id_int) - ) - ).scalar_one_or_none() + statement = select(FundSimAccount).where( + FundSimAccount.customer_id == customer_id_int + ) + if for_update: + statement = statement.with_for_update() + account = (await self._session.execute(statement)).scalar_one_or_none() if account is None: raise AccountNotFoundError(f"客户 {customer_id_int} 未开户") if account.status != "正常": @@ -234,16 +247,21 @@ class TradeService: return account async def _load_holding( - self, customer_id: int, product_id: int + self, customer_id: int, product_id: int, *, for_update: bool = False ) -> FundHolding | None: - return ( - await self._session.execute( - select(FundHolding).where( - FundHolding.customer_id == customer_id, - FundHolding.product_id == product_id, - ) - ) - ).scalar_one_or_none() + """加载持仓。 + + `for_update=True` 的用途与风险同 `_load_account`：不锁的话， + "持仓 1000 份，两笔各卖 1000 份"两笔都能读到 `available_quantity=1000` + 并通过校验，最终把可用份额扣成负数（超卖）。 + """ + statement = select(FundHolding).where( + FundHolding.customer_id == customer_id, + FundHolding.product_id == product_id, + ) + if for_update: + statement = statement.with_for_update() + return (await self._session.execute(statement)).scalar_one_or_none() # ---------- 费率规则 ---------- @@ -286,16 +304,31 @@ class TradeService: # ---------- 委托主流程 ---------- async def submit_order( - self, payload: OrderCreateRequest, context: RequestContext + self, + payload: OrderCreateRequest, + context: RequestContext, + *, + commit: bool = True, ) -> OrderCreateResponse: + """提交委托。 + + `commit=False` 供**幂等包装层**使用：T002 现在由 + `ApiTransactionService.execute_in` 包住，业务写入与幂等回执必须在 + **同一个事务**里（`docs/05` §5.2），所以内层不能再自己提交 —— + 否则就成了"内层提交外层事务"，幂等记录与业务写入会分处两个事务， + 回执写失败时业务已经落库，重放就失去了意义。 + """ customer_id = int(context.user_id) # RequestContext.user_id 是 str（如 '9001'） now = datetime.now(UTC).replace(tzinfo=None) product = await self._load_tradable_product(payload.product_code) await self._check_suitability(customer_id, product, context) quote = await self._fetch_quote(product) - account = await self._load_account(customer_id) - holding = await self._load_holding(customer_id, product.id) + # ⚠️ 「校验 + 扣减」必须在**同一个行锁**内完成，加锁顺序固定：账户 → 持仓。 + # 不加锁时，并发下单各自读到旧余额/旧份额，各自的校验都通过，最终把 + # 可用现金扣成负数、或让卖出超出可用份额（超卖），也可能突破持仓比例上限。 + account = await self._load_account(customer_id, for_update=True) + holding = await self._load_holding(customer_id, product.id, for_update=True) quantity = payload.quantity.quantize(FOUR_PLACES, rounding=ROUND_HALF_UP) if quantity <= 0: @@ -442,7 +475,8 @@ class TradeService: ) self._session.add(ledger) - await self._session.commit() + if commit: + await self._session.commit() return OrderCreateResponse( order_no=order_no, status="已成交",