修复：支持已确认退回的并发幂等

shierfeng-fastapi-backend/module_business/service/return_service.py

@@ -446,14 +446,13 @@ class ReturnService:
         pallet_ids: list[int],
         source_site_id: int,
         current_user: CurrentUserModel,
+        inventories: list[BizPalletInventory],
     ) -> tuple[
         list[BizPalletInventory],
         list[BizPallet],
         dict[int, BizFillingRecord | None],
-        list[int] | None,
     ]:
         ordered_ids = cls._validate_pallet_ids(pallet_ids)
-        inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids)
         inventory_by_id = {inventory.pallet_id: inventory for inventory in inventories}
         if set(inventory_by_id) != set(ordered_ids):
             cls._raise_pallet_not_accessible(current_user, '存在不存在或库存缺失的托盘')
@@ -461,13 +460,11 @@ class ReturnService:
         ordered_inventories = [inventory_by_id[pallet_id] for pallet_id in ordered_ids]
         if any(inventory.site_id != source_site_id for inventory in ordered_inventories):
             cls._raise_pallet_not_accessible(current_user, '同一退回单托盘必须属于同一现场')
-        try:
+        if any(
-            await AccessService.assert_site_access(query_db, current_user, source_site_id)
+            inventory.warehouse_id is not None or inventory.flow_state not in cls._REQUESTABLE_STATES
-        except PermissionException:
+            for inventory in ordered_inventories
-            cls._raise_pallet_not_accessible(current_user)
+        ):
-        fresh_site_ids = await cls._site_scope(query_db, current_user)
+            raise ServiceException(message='托盘当前状态不允许发起退回')
-        if fresh_site_ids is not None and source_site_id not in fresh_site_ids:
-            cls._raise_pallet_not_accessible(current_user)
 
         pallets = await InventoryDao.get_pallets_by_ids(query_db, ordered_ids)
         pallet_by_id = {pallet.pallet_id: pallet for pallet in pallets}
@@ -485,7 +482,7 @@ class ReturnService:
         )
         for filling_id in filling_ids:
             filling_by_id[filling_id] = await OperationDao.get_active_filling_for_update(query_db, filling_id)
-        return ordered_inventories, ordered_pallets, filling_by_id, fresh_site_ids
+        return ordered_inventories, ordered_pallets, filling_by_id
 
     @classmethod
     def _validate_locked_request_pallets(
@@ -618,7 +615,7 @@ class ReturnService:
             await query_db.rollback()
 
     @classmethod
-    async def request_return_services(  # noqa: PLR0915
+    async def request_return_services(  # noqa: PLR0912, PLR0915
         cls,
         query_db: AsyncSession,
         page_object: ReturnRequestModel,
@@ -672,12 +669,18 @@ class ReturnService:
                 await query_db.rollback()
                 return cls._success('申请成功', return_id)
 
-            inventories, pallets, filling_by_id, fresh_site_ids = await cls._lock_request_pallets(
+            inventories = await InventoryDao.get_inventories_for_update(query_db, ordered_ids)
-                query_db,
+            try:
-                ordered_ids,
+                await AccessService.assert_site_access(
-                page_object.source_site_id,
+                    query_db,
-                current_user,
+                    current_user,
-            )
+                    page_object.source_site_id,
+                )
+            except PermissionException:
+                cls._raise_pallet_not_accessible(current_user)
+            fresh_site_ids = await cls._site_scope(query_db, current_user)
+            if fresh_site_ids is not None and page_object.source_site_id not in fresh_site_ids:
+                cls._raise_pallet_not_accessible(current_user)
             fresh_warehouse_ids = None if fresh_site_ids is None else []
             fresh_existing = await OperationDao.get_return_by_request_idempotency_key(
                 query_db,
@@ -693,6 +696,13 @@ class ReturnService:
                 await query_db.rollback()
                 return cls._success('申请成功', fresh_existing_id)
 
+            inventories, pallets, filling_by_id = await cls._lock_request_pallets(
+                query_db,
+                ordered_ids,
+                page_object.source_site_id,
+                current_user,
+                inventories,
+            )
             await cls._lock_new_request_route(query_db, page_object, current_user)
             cls._validate_locked_request_pallets(
                 inventories,

shierfeng-fastapi-backend/tests/module_business/test_return_service.py

@@ -696,16 +696,18 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
 
         key_read = AsyncMock(side_effect=read_key)
         scope_read = AsyncMock(side_effect=read_scope)
+        pallet_locks = AsyncMock(side_effect=lock_pallets)
+        filling_locks = AsyncMock(side_effect=lock_filling)
         self.db.rollback.side_effect = rollback
         with (
             patch.object(AccessDao, 'list_user_site_ids', new=scope_read),
             patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
             patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(side_effect=lock_inventories)),
-            patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)),
+            patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks),
             patch.object(
                 OperationDao,
                 'get_active_filling_for_update',
-                new=AsyncMock(side_effect=lock_filling),
+                new=filling_locks,
             ),
             patch.object(AccessService, 'assert_site_access', new=AsyncMock(side_effect=check_access)),
             patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=lock_site)),
@@ -728,8 +730,6 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
                 'inventory-lock',
                 'access-fresh',
                 'scope-fresh',
-                'pallet-lock',
-                'filling-lock',
                 'key-fresh',
                 'site-lock',
                 'mapping-lock',
@@ -739,6 +739,8 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
         )
         self.assertEqual(2, key_read.await_count)
         self.assertEqual(2, scope_read.await_count)
+        pallet_locks.assert_not_awaited()
+        filling_locks.assert_not_awaited()
         add_order.assert_not_awaited()
         add_items.assert_not_awaited()
         add_event.assert_not_awaited()
@@ -779,6 +781,109 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
         self.db.commit.assert_not_awaited()
         self.assertEqual(2, self.db.rollback.await_count)
 
+    async def test_concurrent_confirmed_winner_returns_before_inventory_business_validation(self) -> None:
+        winner = make_return_order(
+            status=ReturnStatus.CONFIRMED.value,
+            request_hash=ReturnService._build_request_hash(self.page),
+        )
+        relocated = [
+            make_inventory(
+                2,
+                site_id=None,
+                warehouse_id=7,
+                line_id=None,
+                state=PalletFlowState.FILLED_IN_WAREHOUSE.value,
+            ),
+            make_inventory(
+                5,
+                site_id=None,
+                warehouse_id=7,
+                line_id=None,
+                remaining=0,
+                filling_id=None,
+                state=PalletFlowState.EMPTY_IN_WAREHOUSE.value,
+            ),
+        ]
+        key_read = AsyncMock(side_effect=[None, winner])
+        pallet_locks = AsyncMock()
+        filling_locks = AsyncMock()
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=relocated)),
+            patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks),
+            patch.object(OperationDao, 'get_active_filling_for_update', new=filling_locks),
+            patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
+            patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order,
+            patch.object(InventoryDao, 'add_event', new=AsyncMock()) as add_event,
+        ):
+            try:
+                result = await ReturnService.request_return_services(self.db, self.page, self.user)
+            except PermissionException as exc:
+                self.fail(f'已确认赢家在 fresh key 查询前被库存位置提前隐藏：{exc.message}')
+
+        self.assertEqual(13, result.result)
+        self.assertEqual(2, key_read.await_count)
+        pallet_locks.assert_not_awaited()
+        filling_locks.assert_not_awaited()
+        add_order.assert_not_awaited()
+        add_event.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
+    async def test_concurrent_confirmed_winner_different_payload_is_stable_conflict(self) -> None:
+        winner = make_return_order(
+            status=ReturnStatus.CONFIRMED.value,
+            request_hash=ReturnService._build_request_hash(self.page),
+        )
+        conflict_page = self.page.model_copy(update={'target_warehouse_id': 8})
+        relocated = [
+            make_inventory(
+                2,
+                site_id=None,
+                warehouse_id=7,
+                line_id=None,
+                state=PalletFlowState.FILLED_IN_WAREHOUSE.value,
+            ),
+            make_inventory(
+                5,
+                site_id=None,
+                warehouse_id=7,
+                line_id=None,
+                remaining=0,
+                filling_id=None,
+                state=PalletFlowState.EMPTY_IN_WAREHOUSE.value,
+            ),
+        ]
+        key_read = AsyncMock(side_effect=[None, winner])
+        pallet_locks = AsyncMock()
+        filling_locks = AsyncMock()
+        with (
+            patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+            patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
+            patch.object(OperationDao, 'get_return_by_request_idempotency_key', new=key_read),
+            patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=relocated)),
+            patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_locks),
+            patch.object(OperationDao, 'get_active_filling_for_update', new=filling_locks),
+            patch.object(OperationDao, 'add_return_order', new=AsyncMock()) as add_order,
+        ):
+            try:
+                await ReturnService.request_return_services(self.db, conflict_page, self.user)
+            except Exception as exc:
+                raised = exc
+            else:
+                raised = None
+
+        self.assertIsInstance(raised, ServiceException)
+        self.assertEqual('申请退回失败，幂等键已被其他请求使用', getattr(raised, 'message', None))
+        self.assertEqual(2, key_read.await_count)
+        pallet_locks.assert_not_awaited()
+        filling_locks.assert_not_awaited()
+        add_order.assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
     async def test_concurrent_winner_uses_fresh_expanded_site_scope(self) -> None:
         winner = make_return_order(request_hash=ReturnService._build_request_hash(self.page))
         inventories = [
@@ -903,12 +1008,14 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
             with self.subTest(case=name):
                 db = MagicMock(spec=AsyncSession)
                 pallet_locks = AsyncMock(return_value=[make_pallet(2), make_pallet(5)])
+                scope_read = AsyncMock(return_value=[9])
+                key_read = AsyncMock(return_value=None)
                 with (
-                    patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
+                    patch.object(AccessDao, 'list_user_site_ids', new=scope_read),
                     patch.object(
                         OperationDao,
                         'get_return_by_request_idempotency_key',
-                        new=AsyncMock(return_value=None),
+                        new=key_read,
                     ),
                     patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
                     patch.object(
@@ -926,6 +1033,8 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
                         raised = None
                 self.assertIsInstance(raised, PermissionException)
                 messages.append(getattr(raised, 'message', '未抛出异常'))
+                self.assertEqual(2, scope_read.await_count)
+                self.assertEqual(2, key_read.await_count)
                 pallet_locks.assert_not_awaited()
                 db.commit.assert_not_awaited()
                 db.rollback.assert_awaited()