修复：补强使用并发幂等复核

shierfeng-fastapi-backend/module_business/dao/operation_dao.py

@@ -132,8 +132,12 @@ class OperationDao:
         key_column: InstrumentedAttribute,
         idempotency_key: str,
         *extra_conditions: ColumnElement[bool],
+        for_update: bool = False,
     ) -> Any | None:
-        result = await db.execute(select(model).where(key_column == idempotency_key, *extra_conditions))
+        statement = select(model).where(key_column == idempotency_key, *extra_conditions)
+        if for_update:
+            statement = statement.with_for_update().execution_options(populate_existing=True)
+        result = await db.execute(statement)
         return result.scalar_one_or_none()
 
     @staticmethod
@@ -373,6 +377,8 @@ class OperationDao:
         db: AsyncSession,
         idempotency_key: str,
         site_ids: list[int] | None = None,
+        *,
+        for_update: bool = False,
     ) -> BizUsageRecord | None:
         return await cls._get_by_key(
             db,
@@ -380,6 +386,7 @@ class OperationDao:
             BizUsageRecord.idempotency_key,
             idempotency_key,
             *cls._usage_scope_conditions(site_ids),
+            for_update=for_update,
         )
 
     @classmethod

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

@@ -169,6 +169,7 @@ class UsageService:
                 query_db,
                 page_object.idempotency_key,
                 site_ids=site_ids,
+                for_update=False,
             )
             if recovered is None:
                 raise ServiceException(message='使用失败，数据冲突')
@@ -344,6 +345,7 @@ class UsageService:
                 query_db,
                 page_object.idempotency_key,
                 site_ids=site_ids,
+                for_update=False,
             )
             if existing is not None:
                 if not cls._is_exact_idempotent_result(existing, page_object, request_hash):
@@ -385,12 +387,38 @@ class UsageService:
                 cls._raise_pallet_not_accessible(current_user)
             pallet = locked_pallets[0]
 
+            filling = (
+                await OperationDao.get_active_filling_for_update(
+                    query_db,
+                    inventory.active_filling_id,
+                )
+                if inventory.active_filling_id is not None
+                else None
+            )
+            locked_existing = await OperationDao.get_usage_by_idempotency_key(
+                query_db,
+                page_object.idempotency_key,
+                site_ids=site_ids,
+                for_update=True,
+            )
+            if locked_existing is not None:
+                if locked_existing.site_id != inventory.site_id or not cls._is_exact_idempotent_result(
+                    locked_existing,
+                    page_object,
+                    request_hash,
+                ):
+                    raise ServiceException(message='使用失败，幂等键已被其他请求使用')
+                await cls._lock_and_reauthorize_usage_site(
+                    query_db,
+                    locked_existing.site_id,
+                    current_user,
+                )
+                usage_id = locked_existing.usage_id
+                await query_db.rollback()
+                return cls._success(usage_id)
+
             if inventory.active_filling_id is None:
                 raise ServiceException(message='托盘不存在活动装填周期')
-            filling = await OperationDao.get_active_filling_for_update(
-                query_db,
-                inventory.active_filling_id,
-            )
             if filling is None:
                 raise ServiceException(message='托盘活动装填周期不存在或已关闭')
 

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

@@ -133,7 +133,12 @@ class TestDaoPublicContract(unittest.TestCase):
                 'site_ids',
             ),
             'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'),
-            'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'),
+            'get_usage_by_idempotency_key': (
+                'db',
+                'idempotency_key',
+                'site_ids',
+                'for_update',
+            ),
             'get_return_by_request_idempotency_key': ('db', 'idempotency_key'),
             'get_return_by_confirm_idempotency_key': ('db', 'idempotency_key'),
             'get_inbound_by_id': ('db', 'inbound_id', 'warehouse_ids'),
@@ -225,6 +230,13 @@ class TestDaoPublicContract(unittest.TestCase):
                 signature = inspect.signature(getattr(OperationDao, method_name))
                 self.assertIsNone(signature.parameters['site_ids'].default)
 
+        usage_key_signature = inspect.signature(OperationDao.get_usage_by_idempotency_key)
+        usage_key_lock = usage_key_signature.parameters.get('for_update')
+        if usage_key_lock is None:
+            self.fail('get_usage_by_idempotency_key 缺少 for_update 参数')
+        self.assertIs(inspect.Parameter.KEYWORD_ONLY, usage_key_lock.kind)
+        self.assertFalse(usage_key_lock.default)
+
         return_list_signature = inspect.signature(OperationDao.get_return_list)
         self.assertEqual(('db', 'query_object', 'is_page'), tuple(return_list_signature.parameters))
         self.assertIs(True, return_list_signature.parameters['is_page'].default)
@@ -943,6 +955,54 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
         self.assertNotIn('site_id IN', sql)
         self.assertNotIn('WHERE false', sql)
 
+    async def test_usage_idempotency_current_read_locks_refreshes_and_preserves_scope(self) -> None:
+        self.assertIn(
+            'for_update',
+            inspect.signature(OperationDao.get_usage_by_idempotency_key).parameters,
+        )
+        scalar_result = MagicMock()
+        scalar_result.scalar_one_or_none.return_value = None
+        self.db.execute.return_value = scalar_result
+
+        cases = (
+            ([11, 9, 11], 'biz_usage_record.site_id IN (9, 11)'),
+            ([], None),
+            (None, None),
+        )
+        for site_ids, expected_scope in cases:
+            with self.subTest(site_ids=site_ids):
+                self.db.reset_mock()
+                self.db.execute.return_value = scalar_result
+                await OperationDao.get_usage_by_idempotency_key(
+                    self.db,
+                    'usage-key',
+                    site_ids=site_ids,
+                    for_update=True,
+                )
+                statement = self.db.execute.await_args.args[0]
+
+                self.assertTrue(statement.get_execution_options().get('populate_existing'))
+                for dialect in (mysql.dialect(), postgresql.dialect()):
+                    sql = compile_sql(statement, dialect)
+                    self.assertIn('FOR UPDATE', sql)
+                    if expected_scope is not None:
+                        self.assertIn(expected_scope, sql)
+                    elif site_ids == []:
+                        self.assertTrue('false' in sql.lower() or '0 = 1' in sql, sql)
+                    else:
+                        self.assertNotIn('site_id IN', sql)
+
+        self.db.reset_mock()
+        self.db.execute.return_value = scalar_result
+        await OperationDao.get_usage_by_idempotency_key(
+            self.db,
+            'usage-key',
+            site_ids=[9],
+        )
+        unlocked_statement = self.db.execute.await_args.args[0]
+        self.assertNotIn('FOR UPDATE', compile_sql(unlocked_statement, mysql.dialect()))
+        self.assertFalse(unlocked_statement.get_execution_options().get('populate_existing', False))
+
     def test_time_conditions_include_fractional_last_second_with_exclusive_next_day(self) -> None:
         lower, upper = OperationDao._time_conditions(
             BizUsageRecord.occurred_at,

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

@@ -712,6 +712,17 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
             if for_update:
                 actions.append('site-mapping-lock')
 
+        async def get_usage_by_key(
+            _db: AsyncSession,
+            _idempotency_key: str,
+            _site_ids: list[int] | None = None,
+            *,
+            for_update: bool = False,
+            **_kwargs: Any,
+        ) -> None:
+            if for_update:
+                actions.append('usage-current-read')
+
         async def add_usage(_db: AsyncSession, usage: BizUsageRecord) -> BizUsageRecord:
             actions.append('usage-flush')
             usage.usage_id = 61
@@ -740,7 +751,11 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         with (
             patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
             patch.object(AccessService, 'assert_site_access', new=access),
-            patch.object(OperationDao, 'get_usage_by_idempotency_key', new=AsyncMock(return_value=None)),
+            patch.object(
+                OperationDao,
+                'get_usage_by_idempotency_key',
+                new=AsyncMock(side_effect=get_usage_by_key),
+            ),
             patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)),
             patch.object(InventoryDao, 'get_inventory_for_update', new=AsyncMock(side_effect=lock_inventory)),
             patch.object(InventoryDao, 'get_pallets_by_ids', new=AsyncMock(side_effect=lock_pallets)),
@@ -760,6 +775,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
                 'inventory-lock',
                 'pallet-lock',
                 'filling-lock',
+                'usage-current-read',
                 'site-lock',
                 'site-mapping-lock',
                 'line-lock',
@@ -1065,6 +1081,58 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
             request_hash=request_hash or UsageService._build_request_hash(self.page),
         )
 
+    def _patch_concurrent_retry_path(
+        self,
+        *,
+        inventory: BizPalletInventory,
+        recovered: BizUsageRecord,
+        site_scope: list[int] | None = None,
+        site_status: str = '0',
+        revoke_mapping: bool = False,
+    ) -> tuple[ExitStack, dict[str, AsyncMock], list[str]]:
+        actions: list[str] = []
+        locked_site = make_site(status=site_status)
+
+        async def read_key(*_args: Any, **kwargs: Any) -> BizUsageRecord | None:
+            for_update = kwargs.get('for_update', False)
+            actions.append('usage-current-read' if for_update else 'usage-snapshot-read')
+            return recovered if for_update else None
+
+        async def lock_inventory(*_args: Any) -> BizPalletInventory:
+            actions.append('inventory-lock')
+            return inventory
+
+        async def lock_pallet(*_args: Any) -> list[BizPallet]:
+            actions.append('pallet-lock')
+            return [self.core.pallet]
+
+        async def lock_filling(*_args: Any) -> BizFillingRecord:
+            actions.append('filling-lock')
+            return self.core.filling
+
+        async def lock_site(*_args: Any, **_kwargs: Any) -> BizCustomerSite:
+            actions.append('site-lock')
+            return locked_site
+
+        async def check_access(*_args: Any, **kwargs: Any) -> None:
+            if kwargs.get('for_update', False):
+                actions.append('site-mapping-lock')
+                if revoke_mapping:
+                    raise PermissionException(data='', message='无权访问该客户现场')
+
+        stack, mocks = self.core._patch_record_path(
+            inventory=inventory,
+            site=locked_site,
+            site_scope=site_scope,
+            key_read=AsyncMock(side_effect=read_key),
+            access=AsyncMock(side_effect=check_access),
+        )
+        mocks['inventory_lock'].side_effect = lock_inventory
+        mocks['pallet_lock'].side_effect = lock_pallet
+        mocks['filling_lock'].side_effect = lock_filling
+        mocks['site_lock'].side_effect = lock_site
+        return stack, mocks, actions
+
     async def test_exact_retry_rechecks_current_mapping_and_returns_without_mutation(self) -> None:
         existing = self._exact_usage()
         inventory_lock = AsyncMock()
@@ -1084,7 +1152,12 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
         ):
             result = await UsageService.record_usage_services(self.db, self.page, self.user)
 
-        get_key.assert_awaited_once_with(self.db, 'usage-key', site_ids=[9])
+        get_key.assert_awaited_once_with(
+            self.db,
+            'usage-key',
+            site_ids=[9],
+            for_update=False,
+        )
         site_lock.assert_awaited_once_with(self.db, 9, for_update=True)
         access.assert_awaited_once_with(self.db, self.user, 9, for_update=True)
         inventory_lock.assert_not_awaited()
@@ -1123,10 +1196,186 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
         self.db.commit.assert_not_awaited()
         self.db.rollback.assert_awaited_once()
 
+    async def test_concurrent_exact_retry_wins_before_stale_inventory_rejection(self) -> None:
+        cases = (
+            (
+                'first-request-emptied-pallet',
+                make_inventory(
+                    current_line_id=4,
+                    active_filling_id=None,
+                    remaining_packages=0,
+                    flow_state=PalletFlowState.AT_SITE_EMPTY.value,
+                    version=4,
+                ),
+                False,
+                0,
+                '1',
+            ),
+            (
+                'first-request-left-less-than-original-usage',
+                make_inventory(
+                    current_line_id=4,
+                    remaining_packages=2,
+                    flow_state=PalletFlowState.AT_SITE_PARTIAL.value,
+                    version=4,
+                ),
+                True,
+                2,
+                '0',
+            ),
+        )
+        for name, inventory, expect_filling_lock, remaining_after, site_status in cases:
+            with self.subTest(case=name):
+                self.db.reset_mock()
+                recovered = self._exact_usage()
+                recovered.remaining_packages_after = remaining_after
+                stack, mocks, actions = self._patch_concurrent_retry_path(
+                    inventory=inventory,
+                    recovered=recovered,
+                    site_status=site_status,
+                )
+                key_read = mocks['key_read']
+                inventory_before = (
+                    inventory.active_filling_id,
+                    inventory.current_line_id,
+                    inventory.remaining_packages,
+                    inventory.flow_state,
+                    inventory.version,
+                )
+
+                with stack:
+                    try:
+                        result = await UsageService.record_usage_services(
+                            self.db,
+                            self.page,
+                            self.user,
+                        )
+                    except ServiceException as exc:
+                        self.fail(f'精确并发重试被旧库存状态提前拒绝：{exc.message}')
+
+                expected_actions = [
+                    'usage-snapshot-read',
+                    'inventory-lock',
+                    'pallet-lock',
+                ]
+                if expect_filling_lock:
+                    expected_actions.append('filling-lock')
+                expected_actions.extend(
+                    [
+                        'usage-current-read',
+                        'site-lock',
+                        'site-mapping-lock',
+                    ]
+                )
+                self.assertEqual(expected_actions, actions)
+                self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
+                self.assertTrue(all(call.kwargs['site_ids'] == [9] for call in key_read.await_args_list))
+                if expect_filling_lock:
+                    mocks['filling_lock'].assert_awaited_once()
+                else:
+                    mocks['filling_lock'].assert_not_awaited()
+                mocks['site_lock'].assert_awaited_once_with(self.db, 9, for_update=True)
+                mocks['line_lock'].assert_not_awaited()
+                mocks['add_usage'].assert_not_awaited()
+                mocks['add_event'].assert_not_awaited()
+                mocks['update_filling'].assert_not_awaited()
+                self.db.commit.assert_not_awaited()
+                self.db.rollback.assert_awaited_once()
+                self.assertEqual(
+                    inventory_before,
+                    (
+                        inventory.active_filling_id,
+                        inventory.current_line_id,
+                        inventory.remaining_packages,
+                        inventory.flow_state,
+                        inventory.version,
+                    ),
+                )
+                self.assertEqual(61, result.result)
+
+    async def test_concurrent_exact_retry_rechecks_current_mapping_before_success(self) -> None:
+        inventory = make_inventory(
+            current_line_id=4,
+            active_filling_id=None,
+            remaining_packages=0,
+            flow_state=PalletFlowState.AT_SITE_EMPTY.value,
+            version=4,
+        )
+        recovered = self._exact_usage()
+        stack, mocks, _ = self._patch_concurrent_retry_path(
+            inventory=inventory,
+            recovered=recovered,
+            revoke_mapping=True,
+        )
+        key_read = mocks['key_read']
+        access = mocks['access']
+        with stack:
+            try:
+                await UsageService.record_usage_services(self.db, self.page, self.user)
+            except PermissionException as exc:
+                self.assertEqual('无权访问该使用记录', exc.message)
+            except ServiceException as exc:
+                self.fail(f'撤权并发重试未进入 current mapping 复核：{exc.message}')
+            else:
+                self.fail('current mapping 已撤权时不得返回幂等成功')
+
+        self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
+        mocks['site_lock'].assert_awaited_once_with(self.db, 9, for_update=True)
+        self.assertEqual([False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list])
+        mocks['line_lock'].assert_not_awaited()
+        mocks['add_usage'].assert_not_awaited()
+        mocks['add_event'].assert_not_awaited()
+        mocks['update_filling'].assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.db.rollback.assert_awaited_once()
+
+    async def test_concurrent_current_read_rejects_hash_or_actual_site_conflict(self) -> None:
+        request_hash = UsageService._build_request_hash(self.page)
+        conflicts = (
+            ('different-request-hash', make_usage(request_hash='different-hash')),
+            ('different-actual-site', make_usage(site_id=10, request_hash=request_hash)),
+        )
+        for name, recovered in conflicts:
+            with self.subTest(case=name):
+                self.db.reset_mock()
+                inventory = make_inventory(
+                    current_line_id=4,
+                    active_filling_id=None,
+                    remaining_packages=0,
+                    flow_state=PalletFlowState.AT_SITE_EMPTY.value,
+                    version=4,
+                )
+                stack, mocks, _ = self._patch_concurrent_retry_path(
+                    inventory=inventory,
+                    recovered=recovered,
+                    site_scope=[9, 10],
+                )
+                key_read = mocks['key_read']
+                with stack:
+                    try:
+                        await UsageService.record_usage_services(self.db, self.page, self.user)
+                    except ServiceException as exc:
+                        self.assertEqual(
+                            '使用失败，幂等键已被其他请求使用',
+                            exc.message,
+                        )
+                    else:
+                        self.fail('锁后命中的摘要或实际现场冲突不得返回成功')
+
+                self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
+                self.assertTrue(all(call.kwargs['site_ids'] == [9, 10] for call in key_read.await_args_list))
+                mocks['site_lock'].assert_not_awaited()
+                mocks['line_lock'].assert_not_awaited()
+                mocks['add_usage'].assert_not_awaited()
+                mocks['add_event'].assert_not_awaited()
+                mocks['update_filling'].assert_not_awaited()
+                self.db.commit.assert_not_awaited()
+                self.db.rollback.assert_awaited_once()
+
     async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None:
         recovered = self._exact_usage()
         scope = AsyncMock(side_effect=[[9], [9]])
-        key_read = AsyncMock(side_effect=[None, recovered])
+        key_read = AsyncMock(side_effect=[None, None, recovered])
         add_usage = AsyncMock(side_effect=integrity_error())
         stack, mocks = self.core._patch_record_path(
             site_scope=scope,
@@ -1138,8 +1387,9 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual(61, result.result)
         self.assertEqual(2, scope.await_count)
-        self.assertEqual(2, key_read.await_count)
+        self.assertEqual(3, key_read.await_count)
-        self.assertEqual([9], key_read.await_args_list[1].kwargs['site_ids'])
+        self.assertEqual([False, True, False], [call.kwargs['for_update'] for call in key_read.await_args_list])
+        self.assertEqual([9], key_read.await_args_list[2].kwargs['site_ids'])
         self.assertEqual(2, self.db.rollback.await_count)
         self.db.commit.assert_not_awaited()
         mocks['add_event'].assert_not_awaited()
@@ -1148,7 +1398,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
         for recovered in (self._exact_usage(request_hash='different-hash'), None):
             with self.subTest(recovered=recovered is not None):
                 self.db.reset_mock()
-                key_read = AsyncMock(side_effect=[None, recovered])
+                key_read = AsyncMock(side_effect=[None, None, recovered])
                 stack, _ = self.core._patch_record_path(
                     key_read=key_read,
                     add_usage=AsyncMock(side_effect=integrity_error()),
@@ -1171,7 +1421,8 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
             await UsageService.record_usage_services(self.db, self.page, self.user)
 
         self.assertEqual(2, scope.await_count)
-        self.assertEqual(1, key_read.await_count)
+        self.assertEqual(2, key_read.await_count)
+        self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
         self.assertEqual(2, self.db.rollback.await_count)
         mocks['add_event'].assert_not_awaited()
         self.db.commit.assert_not_awaited()