diff --git a/backend/app/services/business_service.py b/backend/app/services/business_service.py index 4a49f2d..062e864 100644 --- a/backend/app/services/business_service.py +++ b/backend/app/services/business_service.py @@ -301,58 +301,54 @@ def create_business( if payload.assistant_personnel_id is not None: _ensure_personnel_active(session, payload.assistant_personnel_id, "协助") - try: - business = Business( - business_no=_generate_no(session), - customer_id=payload.customer_id, - business_date=payload.business_date, - status=BUSINESS_STATUS_ACTIVE, - owner_personnel_id=payload.owner_personnel_id, - assistant_personnel_id=payload.assistant_personnel_id, - is_special=payload.is_special, - special_note=payload.special_note, - remark=payload.remark, - ) - business_repository.create(session, business) - business_repository.replace_items( - session, business.id, _build_item_rows(business.id, payload.items) - ) - _recalc_totals(session, business) - business_repository.add_status_log( - session, - BusinessStatusLog( - business_id=business.id, - old_status=BUSINESS_STATUS_ACTIVE, - new_status=BUSINESS_STATUS_ACTIVE, - operator_id=operator.id, - operator_name=operator.username, - ), - ) - customer = customer_repository.get_by_id(session, business.customer_id) - out = _to_out((business, customer.company_name if customer else None, None, None)) - session.commit() - except IntegrityError as exc: - session.rollback() - if _is_duplicate_entry(exc): - # 编号生成的残余竞态：唯一键兜底重试一次（防无限重试掩盖死锁） - return _retry_create(session, payload, operator=operator, meta=meta) - raise + # 编号唯一键冲突的重试整体收在审计上下文内：rollback 后同一事务重来， + # 成功/最终失败各只记一条审计（吞异常 return 会被记成假成功，重入 + # 整个 create_* 则违反「不支持嵌套」约定）；上限 _NO_RETRY 防无限重试 + # 掩盖死锁类缺陷。审计摘要预取的是首次编号——重试成功时入库编号见 + # 应用日志（AuditRecord frozen，不回填）。 + for attempt in range(_NO_RETRY + 1): + try: + business = Business( + business_no=_generate_no(session), + customer_id=payload.customer_id, + business_date=payload.business_date, + status=BUSINESS_STATUS_ACTIVE, + owner_personnel_id=payload.owner_personnel_id, + assistant_personnel_id=payload.assistant_personnel_id, + is_special=payload.is_special, + special_note=payload.special_note, + remark=payload.remark, + ) + business_repository.create(session, business) + business_repository.replace_items( + session, business.id, _build_item_rows(business.id, payload.items) + ) + _recalc_totals(session, business) + business_repository.add_status_log( + session, + BusinessStatusLog( + business_id=business.id, + old_status=BUSINESS_STATUS_ACTIVE, + new_status=BUSINESS_STATUS_ACTIVE, + operator_id=operator.id, + operator_name=operator.username, + ), + ) + customer = customer_repository.get_by_id(session, business.customer_id) + out = _to_out((business, customer.company_name if customer else None, None, None)) + 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 -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( session: Session, business_id: int, diff --git a/backend/app/services/receipt_service.py b/backend/app/services/receipt_service.py index 162acac..cb1e3af 100644 --- a/backend/app/services/receipt_service.py +++ b/backend/app/services/receipt_service.py @@ -55,6 +55,9 @@ from app.services.no_generator import next_no logger = logging.getLogger(__name__) +# 编号唯一键冲突后的重试次数（模式同 business_service._NO_RETRY） +_NO_RETRY = 1 + def _to_out(row: tuple) -> ReceiptOut: """(receipt, business_no, shipment_no, ship_date) → 响应模型。 @@ -84,9 +87,7 @@ def _to_out(row: tuple) -> ReceiptOut: def _generate_no(session: Session) -> str: """生成回款批次号（HK 前缀；MAX+FOR UPDATE 与 1062 兜底见 no_generator）。""" - return next_no( - session, model=Receipt, column_name="receipt_batch_no", prefix=NO_PREFIX_RECEIPT - ) + return next_no(session, model=Receipt, column_name="receipt_batch_no", prefix=NO_PREFIX_RECEIPT) def _ensure_shipment_of_business( @@ -198,59 +199,55 @@ def create_receipt( ), meta=meta, ): - if business is None: - raise NotFoundError("业务不存在") - if business.status != BUSINESS_STATUS_ACTIVE: - raise BadRequestError("业务已完成或已终止，禁止新增回款") - shipment_no, ship_date = _ensure_shipment_of_business( - session, payload.shipment_id, payload.business_id - ) - - try: - receipt = receipt_repository.create( - session, - Receipt( - receipt_batch_no=batch_no, - business_id=payload.business_id, - shipment_id=payload.shipment_id, - receipt_date=payload.receipt_date, - amount=Decimal(payload.amount), - is_barter=payload.is_barter, - settlement_status=SETTLEMENT_PENDING, - remark=payload.remark, - ), - ) - _recalc_totals(session, business) - out = _to_out((receipt, business.business_no, shipment_no, ship_date)) - new_id = receipt.id - session.commit() - except IntegrityError as exc: - session.rollback() - if _is_duplicate_entry(exc): - # 批次号撞唯一键的残余竞态：兜底重试一次（防无限重试掩盖死锁） - return _retry_create(session, payload, operator=operator, meta=meta) - raise + # 批次号唯一键冲突的重试整体收在审计上下文内（模式同 + # business_service.create_business）：rollback 释放旧锁，重试前重新 + # 锁业务行与生成批次号——成功/最终失败各只记一条审计，无递归 + for attempt in range(_NO_RETRY + 1): + try: + if business is None: + raise NotFoundError("业务不存在") + if business.status != BUSINESS_STATUS_ACTIVE: + raise BadRequestError("业务已完成或已终止，禁止新增回款") + shipment_no, ship_date = _ensure_shipment_of_business( + session, payload.shipment_id, payload.business_id + ) + receipt = receipt_repository.create( + session, + Receipt( + receipt_batch_no=batch_no, + business_id=payload.business_id, + shipment_id=payload.shipment_id, + receipt_date=payload.receipt_date, + amount=Decimal(payload.amount), + is_barter=payload.is_barter, + settlement_status=SETTLEMENT_PENDING, + remark=payload.remark, + ), + ) + _recalc_totals(session, business) + out = _to_out((receipt, business.business_no, shipment_no, ship_date)) + new_id, new_no = receipt.id, batch_no # 固化：commit 后不碰 ORM + session.commit() + break + except IntegrityError as exc: + 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( "新增回款 | receipt_id=%s | no=%s | business_id=%s", new_id, - batch_no, + new_no, payload.business_id, ) 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( session: Session, receipt_id: int, diff --git a/backend/app/services/shipment_service.py b/backend/app/services/shipment_service.py index 7ed085c..059c83f 100644 --- a/backend/app/services/shipment_service.py +++ b/backend/app/services/shipment_service.py @@ -56,6 +56,9 @@ from app.services.no_generator import next_no logger = logging.getLogger(__name__) +# 编号唯一键冲突后的重试次数（模式同 business_service._NO_RETRY） +_NO_RETRY = 1 + def _to_out(row: tuple[Shipment, str]) -> ShipmentOut: """(shipment, business_no) → 响应模型；金额出参 str。""" @@ -117,9 +120,7 @@ def _build_item_rows(items: list[ShipmentItemIn]) -> list[ShipmentItem]: def _generate_no(session: Session) -> str: """生成发货单号（SH 前缀；MAX+FOR UPDATE 与 1062 兜底见 no_generator）。""" - return next_no( - session, model=Shipment, column_name="shipment_no", prefix=NO_PREFIX_SHIPMENT - ) + return next_no(session, model=Shipment, column_name="shipment_no", prefix=NO_PREFIX_SHIPMENT) def _ensure_business_writable(session: Session, business_id: int, action: str) -> Business: @@ -220,57 +221,54 @@ def create_shipment( ), meta=meta, ): - if business is None: - raise NotFoundError("业务不存在") - if business.status != BUSINESS_STATUS_ACTIVE: - raise BadRequestError("业务已完成或已终止，禁止新发货") + # 单号唯一键冲突的重试整体收在审计上下文内（模式同 + # business_service.create_business）：rollback 释放旧锁，重试前重新 + # 锁业务行与生成编号——成功/最终失败各只记一条审计，无递归 + 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_no=shipment_no, - business_id=payload.business_id, - ship_date=payload.ship_date, - freight_amount=_amount_or_zero(payload.freight_amount), - total_amount=Decimal("0"), - remark=payload.remark, - ) - shipment_repository.create(session, shipment) - shipment_repository.replace_items( - session, shipment.id, _build_item_rows(payload.items) - ) - total = shipment_repository.sum_item_amount(session, shipment.id) - shipment.total_amount = total if total is not None else Decimal("0") - _recalc_totals(session, business) - out = _to_out((shipment, business.business_no)) - new_id = shipment.id - session.commit() - except IntegrityError as exc: - session.rollback() - if _is_duplicate_entry(exc): - # 单号撞唯一键的残余竞态：兜底重试一次（防无限重试掩盖死锁） - return _retry_create(session, payload, operator=operator, meta=meta) - raise + shipment = Shipment( + shipment_no=shipment_no, + business_id=payload.business_id, + ship_date=payload.ship_date, + freight_amount=_amount_or_zero(payload.freight_amount), + total_amount=Decimal("0"), + remark=payload.remark, + ) + shipment_repository.create(session, shipment) + shipment_repository.replace_items( + session, shipment.id, _build_item_rows(payload.items) + ) + total = shipment_repository.sum_item_amount(session, shipment.id) + shipment.total_amount = total if total is not None else Decimal("0") + _recalc_totals(session, business) + out = _to_out((shipment, business.business_no)) + new_id, new_no = shipment.id, shipment_no # 固化：commit 后不碰 ORM + session.commit() + break + except IntegrityError as exc: + session.rollback() + if _is_duplicate_entry(exc) and attempt < _NO_RETRY: + business = business_repository.get_by_id( + session, payload.business_id, for_update=True + ) + shipment_no = _generate_no(session) + continue + raise logger.info( "新增发货单 | shipment_id=%s | no=%s | business_id=%s", new_id, - shipment_no, + new_no, payload.business_id, ) 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( session: Session, shipment_id: int, diff --git a/backend/tests/integration/test_audit_retry.py b/backend/tests/integration/test_audit_retry.py new file mode 100644 index 0000000..9074953 --- /dev/null +++ b/backend/tests/integration/test_audit_retry.py @@ -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