fix(business): 唯一键冲突重试改为审计上下文内循环,单条审计无递归

三处 create(业务/发货/回款)的 return _retry_create 是吞异常后正常
退出:外层审计把已回滚的首次尝试记成 success(假成功),重入整个
create_* 违反 audit_service「不支持嵌套」约定,且 _NO_RETRY 死常量
导致持续冲突时无界递归至 RecursionError。改为 audit 内 for 循环
(上限 _NO_RETRY),rollback 后重新锁行与取号,成功/最终失败各恰好
一条审计。新增集成用例锁定两条路径的日志增量。

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
liuwenxin 2026-10-14 15:09:05 +08:00
parent cb977979da
commit 5a3778a424
4 changed files with 290 additions and 139 deletions

View File

@ -301,58 +301,54 @@ def create_business(
if payload.assistant_personnel_id is not None: if payload.assistant_personnel_id is not None:
_ensure_personnel_active(session, payload.assistant_personnel_id, "协助") _ensure_personnel_active(session, payload.assistant_personnel_id, "协助")
try: # 编号唯一键冲突的重试整体收在审计上下文内:rollback 后同一事务重来,
business = Business( # 成功/最终失败各只记一条审计(吞异常 return 会被记成假成功,重入
business_no=_generate_no(session), # 整个 create_* 则违反「不支持嵌套」约定);上限 _NO_RETRY 防无限重试
customer_id=payload.customer_id, # 掩盖死锁类缺陷。审计摘要预取的是首次编号——重试成功时入库编号见
business_date=payload.business_date, # 应用日志(AuditRecord frozen,不回填)。
status=BUSINESS_STATUS_ACTIVE, for attempt in range(_NO_RETRY + 1):
owner_personnel_id=payload.owner_personnel_id, try:
assistant_personnel_id=payload.assistant_personnel_id, business = Business(
is_special=payload.is_special, business_no=_generate_no(session),
special_note=payload.special_note, customer_id=payload.customer_id,
remark=payload.remark, business_date=payload.business_date,
) status=BUSINESS_STATUS_ACTIVE,
business_repository.create(session, business) owner_personnel_id=payload.owner_personnel_id,
business_repository.replace_items( assistant_personnel_id=payload.assistant_personnel_id,
session, business.id, _build_item_rows(business.id, payload.items) is_special=payload.is_special,
) special_note=payload.special_note,
_recalc_totals(session, business) remark=payload.remark,
business_repository.add_status_log( )
session, business_repository.create(session, business)
BusinessStatusLog( business_repository.replace_items(
business_id=business.id, session, business.id, _build_item_rows(business.id, payload.items)
old_status=BUSINESS_STATUS_ACTIVE, )
new_status=BUSINESS_STATUS_ACTIVE, _recalc_totals(session, business)
operator_id=operator.id, business_repository.add_status_log(
operator_name=operator.username, session,
), BusinessStatusLog(
) business_id=business.id,
customer = customer_repository.get_by_id(session, business.customer_id) old_status=BUSINESS_STATUS_ACTIVE,
out = _to_out((business, customer.company_name if customer else None, None, None)) new_status=BUSINESS_STATUS_ACTIVE,
session.commit() operator_id=operator.id,
except IntegrityError as exc: operator_name=operator.username,
session.rollback() ),
if _is_duplicate_entry(exc): )
# 编号生成的残余竞态:唯一键兜底重试一次(防无限重试掩盖死锁) customer = customer_repository.get_by_id(session, business.customer_id)
return _retry_create(session, payload, operator=operator, meta=meta) out = _to_out((business, customer.company_name if customer else None, None, None))
raise new_id, new_no = business.id, business.business_no # 固化:commit 后不碰 ORM
session.commit()
break
except IntegrityError as exc:
session.rollback()
if _is_duplicate_entry(exc) and attempt < _NO_RETRY:
continue
raise
logger.info("创建业务 | business_id=%s | no=%s", business.id, business.business_no) logger.info("创建业务 | business_id=%s | no=%s", new_id, new_no)
return out return out
def _retry_create(
session: Session,
payload: BusinessCreateIn,
*,
operator: User,
meta: RequestMeta | None,
) -> BusinessOut:
"""编号撞唯一键后的单次重试(防无限重试掩盖死锁类缺陷)。"""
return create_business(session, payload, operator=operator, meta=meta)
def update_business( def update_business(
session: Session, session: Session,
business_id: int, business_id: int,

View File

@ -55,6 +55,9 @@ from app.services.no_generator import next_no
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 编号唯一键冲突后的重试次数(模式同 business_service._NO_RETRY)
_NO_RETRY = 1
def _to_out(row: tuple) -> ReceiptOut: def _to_out(row: tuple) -> ReceiptOut:
"""(receipt, business_no, shipment_no, ship_date) → 响应模型。 """(receipt, business_no, shipment_no, ship_date) → 响应模型。
@ -84,9 +87,7 @@ def _to_out(row: tuple) -> ReceiptOut:
def _generate_no(session: Session) -> str: def _generate_no(session: Session) -> str:
"""生成回款批次号(HK 前缀;MAX+FOR UPDATE 与 1062 兜底见 no_generator)。""" """生成回款批次号(HK 前缀;MAX+FOR UPDATE 与 1062 兜底见 no_generator)。"""
return next_no( return next_no(session, model=Receipt, column_name="receipt_batch_no", prefix=NO_PREFIX_RECEIPT)
session, model=Receipt, column_name="receipt_batch_no", prefix=NO_PREFIX_RECEIPT
)
def _ensure_shipment_of_business( def _ensure_shipment_of_business(
@ -198,59 +199,55 @@ def create_receipt(
), ),
meta=meta, meta=meta,
): ):
if business is None: # 批次号唯一键冲突的重试整体收在审计上下文内(模式同
raise NotFoundError("业务不存在") # business_service.create_business):rollback 释放旧锁,重试前重新
if business.status != BUSINESS_STATUS_ACTIVE: # 锁业务行与生成批次号——成功/最终失败各只记一条审计,无递归
raise BadRequestError("业务已完成或已终止,禁止新增回款") for attempt in range(_NO_RETRY + 1):
shipment_no, ship_date = _ensure_shipment_of_business( try:
session, payload.shipment_id, payload.business_id if business is None:
) raise NotFoundError("业务不存在")
if business.status != BUSINESS_STATUS_ACTIVE:
try: raise BadRequestError("业务已完成或已终止,禁止新增回款")
receipt = receipt_repository.create( shipment_no, ship_date = _ensure_shipment_of_business(
session, session, payload.shipment_id, payload.business_id
Receipt( )
receipt_batch_no=batch_no, receipt = receipt_repository.create(
business_id=payload.business_id, session,
shipment_id=payload.shipment_id, Receipt(
receipt_date=payload.receipt_date, receipt_batch_no=batch_no,
amount=Decimal(payload.amount), business_id=payload.business_id,
is_barter=payload.is_barter, shipment_id=payload.shipment_id,
settlement_status=SETTLEMENT_PENDING, receipt_date=payload.receipt_date,
remark=payload.remark, amount=Decimal(payload.amount),
), is_barter=payload.is_barter,
) settlement_status=SETTLEMENT_PENDING,
_recalc_totals(session, business) remark=payload.remark,
out = _to_out((receipt, business.business_no, shipment_no, ship_date)) ),
new_id = receipt.id )
session.commit() _recalc_totals(session, business)
except IntegrityError as exc: out = _to_out((receipt, business.business_no, shipment_no, ship_date))
session.rollback() new_id, new_no = receipt.id, batch_no # 固化:commit 后不碰 ORM
if _is_duplicate_entry(exc): session.commit()
# 批次号撞唯一键的残余竞态:兜底重试一次(防无限重试掩盖死锁) break
return _retry_create(session, payload, operator=operator, meta=meta) except IntegrityError as exc:
raise session.rollback()
if _is_duplicate_entry(exc) and attempt < _NO_RETRY:
business = business_repository.get_by_id(
session, payload.business_id, for_update=True
)
batch_no = _generate_no(session)
continue
raise
logger.info( logger.info(
"新增回款 | receipt_id=%s | no=%s | business_id=%s", "新增回款 | receipt_id=%s | no=%s | business_id=%s",
new_id, new_id,
batch_no, new_no,
payload.business_id, payload.business_id,
) )
return out return out
def _retry_create(
session: Session,
payload: ReceiptCreateIn,
*,
operator,
meta: RequestMeta | None,
) -> ReceiptOut:
"""批次号撞唯一键后的单次重试(重新锁行与生成编号,防死锁类缺陷掩盖)。"""
return create_receipt(session, payload, operator=operator, meta=meta)
def update_receipt( def update_receipt(
session: Session, session: Session,
receipt_id: int, receipt_id: int,

View File

@ -56,6 +56,9 @@ from app.services.no_generator import next_no
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
# 编号唯一键冲突后的重试次数(模式同 business_service._NO_RETRY)
_NO_RETRY = 1
def _to_out(row: tuple[Shipment, str]) -> ShipmentOut: def _to_out(row: tuple[Shipment, str]) -> ShipmentOut:
"""(shipment, business_no) → 响应模型;金额出参 str。""" """(shipment, business_no) → 响应模型;金额出参 str。"""
@ -117,9 +120,7 @@ def _build_item_rows(items: list[ShipmentItemIn]) -> list[ShipmentItem]:
def _generate_no(session: Session) -> str: def _generate_no(session: Session) -> str:
"""生成发货单号(SH 前缀;MAX+FOR UPDATE 与 1062 兜底见 no_generator)。""" """生成发货单号(SH 前缀;MAX+FOR UPDATE 与 1062 兜底见 no_generator)。"""
return next_no( return next_no(session, model=Shipment, column_name="shipment_no", prefix=NO_PREFIX_SHIPMENT)
session, model=Shipment, column_name="shipment_no", prefix=NO_PREFIX_SHIPMENT
)
def _ensure_business_writable(session: Session, business_id: int, action: str) -> Business: def _ensure_business_writable(session: Session, business_id: int, action: str) -> Business:
@ -220,57 +221,54 @@ def create_shipment(
), ),
meta=meta, meta=meta,
): ):
if business is None: # 单号唯一键冲突的重试整体收在审计上下文内(模式同
raise NotFoundError("业务不存在") # business_service.create_business):rollback 释放旧锁,重试前重新
if business.status != BUSINESS_STATUS_ACTIVE: # 锁业务行与生成编号——成功/最终失败各只记一条审计,无递归
raise BadRequestError("业务已完成或已终止,禁止新发货") for attempt in range(_NO_RETRY + 1):
try:
if business is None:
raise NotFoundError("业务不存在")
if business.status != BUSINESS_STATUS_ACTIVE:
raise BadRequestError("业务已完成或已终止,禁止新发货")
try: shipment = Shipment(
shipment = Shipment( shipment_no=shipment_no,
shipment_no=shipment_no, business_id=payload.business_id,
business_id=payload.business_id, ship_date=payload.ship_date,
ship_date=payload.ship_date, freight_amount=_amount_or_zero(payload.freight_amount),
freight_amount=_amount_or_zero(payload.freight_amount), total_amount=Decimal("0"),
total_amount=Decimal("0"), remark=payload.remark,
remark=payload.remark, )
) shipment_repository.create(session, shipment)
shipment_repository.create(session, shipment) shipment_repository.replace_items(
shipment_repository.replace_items( session, shipment.id, _build_item_rows(payload.items)
session, shipment.id, _build_item_rows(payload.items) )
) total = shipment_repository.sum_item_amount(session, shipment.id)
total = shipment_repository.sum_item_amount(session, shipment.id) shipment.total_amount = total if total is not None else Decimal("0")
shipment.total_amount = total if total is not None else Decimal("0") _recalc_totals(session, business)
_recalc_totals(session, business) out = _to_out((shipment, business.business_no))
out = _to_out((shipment, business.business_no)) new_id, new_no = shipment.id, shipment_no # 固化:commit 后不碰 ORM
new_id = shipment.id session.commit()
session.commit() break
except IntegrityError as exc: except IntegrityError as exc:
session.rollback() session.rollback()
if _is_duplicate_entry(exc): if _is_duplicate_entry(exc) and attempt < _NO_RETRY:
# 单号撞唯一键的残余竞态:兜底重试一次(防无限重试掩盖死锁) business = business_repository.get_by_id(
return _retry_create(session, payload, operator=operator, meta=meta) session, payload.business_id, for_update=True
raise )
shipment_no = _generate_no(session)
continue
raise
logger.info( logger.info(
"新增发货单 | shipment_id=%s | no=%s | business_id=%s", "新增发货单 | shipment_id=%s | no=%s | business_id=%s",
new_id, new_id,
shipment_no, new_no,
payload.business_id, payload.business_id,
) )
return out return out
def _retry_create(
session: Session,
payload: ShipmentCreateIn,
*,
operator: User,
meta: RequestMeta | None,
) -> ShipmentOut:
"""单号撞唯一键后的单次重试(重新锁行与生成编号,防死锁类缺陷掩盖)。"""
return create_shipment(session, payload, operator=operator, meta=meta)
def update_shipment( def update_shipment(
session: Session, session: Session,
shipment_id: int, shipment_id: int,

View File

@ -0,0 +1,160 @@
"""单据编号唯一键冲突重试的行为锁定(audit 上下文内循环重试)。
守:1062 兜底重试若以「吞异常后 return 重入整个 create_*」实现,会产生
三重缺陷 —— 外层审计把已回滚的首次尝试记成 success=True(假成功,与
audit_service.py「上下文内不得吞掉异常后正常退出」的约定相反)、重试再
开一层审计(违反「不支持嵌套」)、持续冲突时递归至 RecursionError。
修复口径:审计上下文内 for 循环重试(上限 _NO_RETRY),成功/最终失败
各恰好一条操作日志(以 POST 前的日志条数为基线断言增量)。
制造冲突的手段:monkeypatch ``business_service._generate_no`` 首次返回
库内已存在编号 → flush 撞唯一键 → 走重试分支。审计按 username 冗余溯
源(账号删除后 user_id 被 SET NULL),断言以基线增量免受既有数据干扰。
"""
import uuid
import pytest
from fastapi.testclient import TestClient
from sqlalchemy import func, select
from app.core.constants import ACTION_CREATE, MODULE_BUSINESS
from app.core.db import get_session_factory
from app.main import app
from app.models import OperationLog
from app.services import business_service
pytestmark = pytest.mark.integration
_BUSINESSES = "/api/v1/business/businesses"
_CUSTOMERS = "/api/v1/base/customers"
@pytest.fixture
def quiet_client():
"""持续冲突用例依赖全局异常处理器产出 500 信封。
TestClient 默认对 ServerErrorMiddleware 的响应直接 re-raise
(raise_server_exceptions=True),会掩盖「原子处理器已正确收敛 500」
的行为本身 —— 审计断言反而无从谈起。
"""
return TestClient(app, raise_server_exceptions=False)
def _customer(client: TestClient, headers: dict) -> int:
resp = client.post(
_CUSTOMERS,
json={
"company_name": f"t_retry公司_{uuid.uuid4().hex[:8]}",
"owner_name": "张老板",
"customer_type": "personal",
},
headers=headers,
)
assert resp.status_code == 200, resp.text
return int(resp.json()["data"]["id"])
def _business_payload(customer_id: int) -> dict:
return {
"customer_id": customer_id,
"business_date": "2026-01-01",
"items": [
{
"product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
"product_category": "municipal_water",
"quantity": "1.00",
"sale_price": "10.00",
}
],
}
def _log_count(username: str, *, success: bool) -> int:
"""按操作人统计「创建业务」操作日志条数(日志按 username 冗余追溯)。"""
factory = get_session_factory()
with factory() as session:
stmt = (
select(func.count())
.select_from(OperationLog)
.where(
OperationLog.username == username,
OperationLog.module == MODULE_BUSINESS,
OperationLog.action == ACTION_CREATE,
OperationLog.summary.like("创建业务%"),
OperationLog.success == int(success),
)
)
return int(session.execute(stmt).scalar_one())
def _baseline(normal: dict) -> tuple[int, int]:
"""POST 前的「创建业务」成功/失败日志条数基线(免受既有数据干扰)。"""
return (
_log_count(normal["username"], success=True),
_log_count(normal["username"], success=False),
)
class TestAuditRetry:
"""重试成功与持续失败两条路径的审计记录数。"""
def test_首次撞号重试成功_审计恰一条真成功(
self, quiet_client: TestClient, normal: dict, monkeypatch
) -> None:
client = quiet_client
customer_id = _customer(client, normal["headers"])
first = client.post(
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
)
assert first.status_code == 200, first.text
existing_no = first.json()["data"]["business_no"]
real_generate = business_service._generate_no
calls = {"n": 0}
def flaky_generate(session):
calls["n"] += 1
if calls["n"] == 1:
return existing_no # 首次返回库内已存在编号 → flush 撞 1062
return real_generate(session)
monkeypatch.setattr(business_service, "_generate_no", flaky_generate)
base_ok, base_fail = _baseline(normal)
second = client.post(
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
)
assert second.status_code == 200, second.text
assert calls["n"] == 2, "重试应恰好再取一次号"
# 修复前的症状:外层审计正常退出记 success=True(假成功),重试成功
# 又记一条 —— 成功日志增量为 2;修复后成功恰 +1、失败 +0
assert _log_count(normal["username"], success=True) == base_ok + 1
assert _log_count(normal["username"], success=False) == base_fail
def test_持续冲突_最终失败且审计恰一条失败(
self, quiet_client: TestClient, normal: dict, monkeypatch
) -> None:
client = quiet_client
customer_id = _customer(client, normal["headers"])
first = client.post(
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
)
assert first.status_code == 200, first.text
existing_no = first.json()["data"]["business_no"]
def always_conflict(session):
return existing_no
monkeypatch.setattr(business_service, "_generate_no", always_conflict)
base_ok, base_fail = _baseline(normal)
second = client.post(
_BUSINESSES, json=_business_payload(customer_id), headers=normal["headers"]
)
# 未捕获 IntegrityError 由全局异常兜底收敛 500
assert second.status_code == 500, second.text
# 修复前的症状:return _retry_create 无界递归直至 RecursionError,
# 每层各写一条记录;修复后重试有上限,最终失败仅一条失败日志
assert _log_count(normal["username"], success=False) == base_fail + 1
assert _log_count(normal["username"], success=True) == base_ok