修复：消除使用幂等缺失键间隙锁

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

@@ -132,12 +132,8 @@ class OperationDao:
         key_column: InstrumentedAttribute,
         idempotency_key: str,
         *extra_conditions: ColumnElement[bool],
-        for_update: bool = False,
     ) -> Any | None:
-        statement = select(model).where(key_column == idempotency_key, *extra_conditions)
+        result = await db.execute(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
@@ -377,8 +373,6 @@ 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,
@@ -386,7 +380,6 @@ 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,7 +169,6 @@ class UsageService:
                 query_db,
                 page_object.idempotency_key,
                 site_ids=site_ids,
-                for_update=False,
             )
             if recovered is None:
                 raise ServiceException(message='使用失败，数据冲突')
@@ -340,17 +339,16 @@ class UsageService:
         try:
             request_hash = cls._build_request_hash(page_object)
             user_id, user_name = cls._current_identity(current_user)
-            site_ids = await cls._site_scope(query_db, current_user)
+            preflight_site_ids = await cls._site_scope(query_db, current_user)
             existing = await OperationDao.get_usage_by_idempotency_key(
                 query_db,
                 page_object.idempotency_key,
-                site_ids=site_ids,
+                site_ids=preflight_site_ids,
-                for_update=False,
             )
             if existing is not None:
                 if not cls._is_exact_idempotent_result(existing, page_object, request_hash):
                     raise ServiceException(message='使用失败，幂等键已被其他请求使用')
-                if site_ids is not None and existing.site_id not in site_ids:
+                if preflight_site_ids is not None and existing.site_id not in preflight_site_ids:
                     cls._raise_usage_not_accessible(current_user)
                 await cls._lock_and_reauthorize_usage_site(
                     query_db,
@@ -367,10 +365,12 @@ class UsageService:
             )
             if scanned_pallet is None:
                 cls._raise_pallet_not_accessible(current_user)
+            scanned_pallet_id = int(scanned_pallet.pallet_id)
+            await query_db.rollback()
 
             inventory = await InventoryDao.get_inventory_for_update(
                 query_db,
-                scanned_pallet.pallet_id,
+                scanned_pallet_id,
             )
             if inventory is None:
                 cls._raise_pallet_not_accessible(current_user, '托盘库存不存在')
@@ -378,6 +378,7 @@ class UsageService:
                 cls._raise_pallet_not_accessible(current_user, '托盘不在客户现场')
             expected_site_id = inventory.site_id
             await cls._assert_pallet_site_access(query_db, current_user, inventory.site_id)
+            locked_site_ids = await cls._site_scope(query_db, current_user)
 
             locked_pallets = await InventoryDao.get_pallets_by_ids(
                 query_db,
@@ -398,8 +399,7 @@ class UsageService:
             locked_existing = await OperationDao.get_usage_by_idempotency_key(
                 query_db,
                 page_object.idempotency_key,
-                site_ids=site_ids,
+                site_ids=locked_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(

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

@@ -133,12 +133,7 @@ class TestDaoPublicContract(unittest.TestCase):
                 'site_ids',
             ),
             'get_filling_by_idempotency_key': ('db', 'idempotency_key', 'warehouse_ids'),
-            'get_usage_by_idempotency_key': (
+            'get_usage_by_idempotency_key': ('db', 'idempotency_key', 'site_ids'),
-                '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'),
@@ -230,12 +225,7 @@ 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)
+        self.assertNotIn('for_update', inspect.signature(OperationDao._get_by_key).parameters)
-        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))
@@ -955,11 +945,7 @@ 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:
+    async def test_usage_idempotency_reads_never_lock_and_preserve_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
@@ -977,14 +963,13 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
                     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'))
+                self.assertFalse(statement.get_execution_options().get('populate_existing', False))
                 for dialect in (mysql.dialect(), postgresql.dialect()):
                     sql = compile_sql(statement, dialect)
-                    self.assertIn('FOR UPDATE', sql)
+                    self.assertNotIn('FOR UPDATE', sql)
                     if expected_scope is not None:
                         self.assertIn(expected_scope, sql)
                     elif site_ids == []:
@@ -992,17 +977,6 @@ class TestOperationDaoReads(unittest.IsolatedAsyncioTestCase):
                     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

@@ -6,7 +6,7 @@ import json
 import unittest
 from contextlib import ExitStack
 from datetime import datetime
-from typing import Any, get_type_hints
+from typing import Any, NoReturn, get_type_hints
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from fastapi import FastAPI
@@ -610,6 +610,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         *,
         site_scope: list[int] | None | AsyncMock = None,
         key_read: BizUsageRecord | None | AsyncMock = None,
+        scanned_pallet: Any | None = None,
         pallet: BizPallet | None = None,
         inventory: BizPalletInventory | None = None,
         filling: BizFillingRecord | None = None,
@@ -627,6 +628,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
             scope_read = AsyncMock(return_value=[9] if site_scope is None else site_scope)
         usage_key_read = key_read if isinstance(key_read, AsyncMock) else AsyncMock(return_value=key_read)
         selected_pallet = pallet or self.pallet
+        selected_scanned_pallet = selected_pallet if scanned_pallet is None else scanned_pallet
         selected_inventory = inventory or self.inventory
         selected_filling = filling or self.filling
         selected_site = self.site if site is None else site
@@ -640,6 +642,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         usage_write = add_usage or AsyncMock(side_effect=assign_usage_id)
         filling_write = update_filling or AsyncMock()
         event_write = add_event or AsyncMock()
+        scan_pallet = AsyncMock(return_value=selected_scanned_pallet)
         inventory_lock = AsyncMock(return_value=selected_inventory)
         pallet_lock = AsyncMock(return_value=[selected_pallet])
         filling_lock = AsyncMock(return_value=selected_filling)
@@ -649,9 +652,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         stack.enter_context(patch.object(AccessDao, 'list_user_site_ids', new=scope_read))
         stack.enter_context(patch.object(AccessService, 'assert_site_access', new=site_access))
         stack.enter_context(patch.object(OperationDao, 'get_usage_by_idempotency_key', new=usage_key_read))
-        stack.enter_context(
+        stack.enter_context(patch.object(InventoryDao, 'get_pallet_by_qr_token', new=scan_pallet))
-            patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=selected_pallet))
-        )
         stack.enter_context(patch.object(InventoryDao, 'get_inventory_for_update', new=inventory_lock))
         stack.enter_context(patch.object(InventoryDao, 'get_pallets_by_ids', new=pallet_lock))
         stack.enter_context(patch.object(OperationDao, 'get_active_filling_for_update', new=filling_lock))
@@ -665,6 +666,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         return stack, {
             'scope_read': scope_read,
             'key_read': usage_key_read,
+            'scan_pallet': scan_pallet,
             'access': site_access,
             'inventory_lock': inventory_lock,
             'pallet_lock': pallet_lock,
@@ -681,6 +683,21 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
     ) -> None:
         actions: list[str] = []
         captured: dict[str, Any] = {}
+        scope_call_count = 0
+        key_call_count = 0
+
+        async def read_site_scope(*_args: Any) -> list[int]:
+            nonlocal scope_call_count
+            scope_call_count += 1
+            actions.append('preflight-scope' if scope_call_count == 1 else 'fresh-scope')
+            return [9]
+
+        async def scan_pallet(*_args: Any) -> BizPallet:
+            actions.append('qr-snapshot-read')
+            return self.pallet
+
+        async def rollback() -> None:
+            actions.append('rollback')
 
         async def lock_inventory(*_args: Any) -> BizPalletInventory:
             actions.append('inventory-lock')
@@ -711,17 +728,16 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         ) -> None:
             if for_update:
                 actions.append('site-mapping-lock')
+            else:
+                actions.append('site-access')
 
         async def get_usage_by_key(
-            _db: AsyncSession,
+            *_args: Any,
-            _idempotency_key: str,
-            _site_ids: list[int] | None = None,
-            *,
-            for_update: bool = False,
             **_kwargs: Any,
         ) -> None:
-            if for_update:
+            nonlocal key_call_count
-                actions.append('usage-current-read')
+            key_call_count += 1
+            actions.append('usage-preflight-read' if key_call_count == 1 else 'usage-fresh-read')
 
         async def add_usage(_db: AsyncSession, usage: BizUsageRecord) -> BizUsageRecord:
             actions.append('usage-flush')
@@ -745,18 +761,21 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
             actions.append('commit')
             captured['usage'].__dict__.pop('usage_id', None)
 
+        self.db.rollback.side_effect = rollback
         self.db.commit.side_effect = commit
+        scope_read = AsyncMock(side_effect=read_site_scope)
+        key_read = AsyncMock(side_effect=get_usage_by_key)
         access = AsyncMock(side_effect=assert_site_access)
         update_filling = AsyncMock()
         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(AccessService, 'assert_site_access', new=access),
             patch.object(
                 OperationDao,
                 'get_usage_by_idempotency_key',
-                new=AsyncMock(side_effect=get_usage_by_key),
+                new=key_read,
             ),
-            patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(return_value=self.pallet)),
+            patch.object(InventoryDao, 'get_pallet_by_qr_token', new=AsyncMock(side_effect=scan_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)),
             patch.object(OperationDao, 'get_active_filling_for_update', new=AsyncMock(side_effect=lock_filling)),
@@ -772,10 +791,16 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
 
         self.assertEqual(
             [
+                'preflight-scope',
+                'usage-preflight-read',
+                'qr-snapshot-read',
+                'rollback',
                 'inventory-lock',
+                'site-access',
+                'fresh-scope',
                 'pallet-lock',
                 'filling-lock',
-                'usage-current-read',
+                'usage-fresh-read',
                 'site-lock',
                 'site-mapping-lock',
                 'line-lock',
@@ -785,6 +810,11 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
             ],
             actions,
         )
+        self.assertEqual(2, scope_read.await_count)
+        self.assertEqual(
+            [{'site_ids': [9]}, {'site_ids': [9]}],
+            [call.kwargs for call in key_read.await_args_list],
+        )
         self.assertEqual([False, True], [call.kwargs.get('for_update', False) for call in access.await_args_list])
         self.assertEqual([9, 9], [call.args[2] for call in access.await_args_list])
         self.assertEqual(4, self.inventory.current_line_id)
@@ -829,7 +859,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         self.assertEqual('usage-key', event.idempotency_key)
         self.assertEqual(usage.request_hash, event.request_hash)
         self.db.commit.assert_awaited_once_with()
-        self.db.rollback.assert_not_awaited()
+        self.db.rollback.assert_awaited_once_with()
         self.assertTrue(result.is_success)
         self.assertEqual(61, result.result)
 
@@ -869,7 +899,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(0, event.after_remaining_packages)
         self.assertEqual(4, event.after_line_id)
         self.db.commit.assert_awaited_once_with()
-        self.db.rollback.assert_not_awaited()
+        self.db.rollback.assert_awaited_once_with()
         mocks['add_usage'].assert_awaited_once()
 
     async def test_same_line_can_continue_but_different_line_is_rejected(self) -> None:
@@ -900,7 +930,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         mocks['add_usage'].assert_not_awaited()
         mocks['add_event'].assert_not_awaited()
         self.db.commit.assert_not_awaited()
-        self.db.rollback.assert_awaited_once()
+        self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_line_must_exist_be_enabled_and_belong_to_locked_site(self) -> None:
         for line, message in (
@@ -919,7 +949,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
                 mocks['add_usage'].assert_not_awaited()
                 mocks['add_event'].assert_not_awaited()
                 self.db.commit.assert_not_awaited()
-                self.db.rollback.assert_awaited_once()
+                self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_only_site_usage_states_and_strict_current_quantity_are_allowed(self) -> None:
         cases = (
@@ -950,7 +980,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
                 mocks['add_usage'].assert_not_awaited()
                 mocks['add_event'].assert_not_awaited()
                 self.db.commit.assert_not_awaited()
-                self.db.rollback.assert_awaited_once()
+                self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_initial_actual_site_denial_hides_qr_before_secondary_locks(self) -> None:
         access = AsyncMock(side_effect=PermissionException(data='', message='无权访问该客户现场'))
@@ -965,7 +995,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         mocks['site_lock'].assert_not_awaited()
         mocks['line_lock'].assert_not_awaited()
         mocks['add_usage'].assert_not_awaited()
-        self.db.rollback.assert_awaited_once()
+        self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_current_mapping_denial_wins_over_disabled_site_without_leaking_status(self) -> None:
         async def access_check(
@@ -988,7 +1018,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         access.assert_awaited_with(self.db, self.user, 9, for_update=True)
         mocks['line_lock'].assert_not_awaited()
         mocks['add_usage'].assert_not_awaited()
-        self.db.rollback.assert_awaited_once()
+        self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_admin_bypasses_mapping_but_gets_precise_site_errors(self) -> None:
         for site, message in ((None, '客户现场不存在'), (make_site(status='1'), '客户现场已停用')):
@@ -1001,7 +1031,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
                     await UsageService.record_usage_services(self.db, self.page, make_user(1))
                 self.assertEqual(message, raised.exception.message)
                 mocks['add_usage'].assert_not_awaited()
-                self.db.rollback.assert_awaited_once()
+                self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_locked_qr_asset_and_active_filling_invariants_fail_before_usage_write(self) -> None:
         cases = (
@@ -1021,7 +1051,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
                 mocks['update_filling'].assert_not_awaited()
                 mocks['add_event'].assert_not_awaited()
                 self.db.commit.assert_not_awaited()
-                self.db.rollback.assert_awaited_once()
+                self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_invalid_locked_inventory_version_fails_before_usage_write(self) -> None:
         inventory = make_inventory(version=-1)
@@ -1032,7 +1062,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
         self.assertEqual('托盘库存版本无效', raised.exception.message)
         mocks['add_usage'].assert_not_awaited()
         mocks['add_event'].assert_not_awaited()
-        self.db.rollback.assert_awaited_once()
+        self.assertEqual(2, self.db.rollback.await_count)
 
     async def test_record_hides_missing_and_non_site_qr_but_admin_gets_precise_error(self) -> None:
         with (
@@ -1061,7 +1091,7 @@ class TestRecordUsageCore(unittest.IsolatedAsyncioTestCase):
             await UsageService.record_usage_services(self.db, self.page, self.user)
         self.assertEqual('无权访问该托盘', non_site.exception.message)
         mocks['pallet_lock'].assert_not_awaited()
-        self.db.rollback.assert_awaited_once()
+        self.assertEqual(2, self.db.rollback.await_count)
 
 
 @unittest.skipUnless(UsageService is not None, 'UsageService 尚未实现')
@@ -1086,17 +1116,40 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
         *,
         inventory: BizPalletInventory,
         recovered: BizUsageRecord,
-        site_scope: list[int] | None = None,
+        scope_sequence: tuple[list[int], list[int]] | None = None,
         site_status: str = '0',
         revoke_mapping: bool = False,
+        revoke_initial_access: bool = False,
     ) -> tuple[ExitStack, dict[str, AsyncMock], list[str]]:
         actions: list[str] = []
         locked_site = make_site(status=site_status)
+        scopes = scope_sequence or ([9], [9])
+        scope_call_count = 0
+        key_call_count = 0
+        rollback_count = 0
+
+        async def read_site_scope(*_args: Any) -> list[int]:
+            nonlocal scope_call_count
+            scope_call_count += 1
+            actions.append('preflight-scope' if scope_call_count == 1 else 'fresh-scope')
+            return scopes[scope_call_count - 1]
 
         async def read_key(*_args: Any, **kwargs: Any) -> BizUsageRecord | None:
-            for_update = kwargs.get('for_update', False)
+            nonlocal key_call_count
-            actions.append('usage-current-read' if for_update else 'usage-snapshot-read')
+            key_call_count += 1
-            return recovered if for_update else None
+            actions.append('usage-preflight-read' if key_call_count == 1 else 'usage-fresh-read')
+            if key_call_count == 1:
+                return None
+            return recovered if kwargs.get('site_ids') == scopes[1] else None
+
+        async def scan_pallet(*_args: Any) -> BizPallet:
+            actions.append('qr-snapshot-read')
+            return self.core.pallet
+
+        async def rollback() -> None:
+            nonlocal rollback_count
+            rollback_count += 1
+            actions.append('preflight-rollback' if rollback_count == 1 else 'result-rollback')
 
         async def lock_inventory(*_args: Any) -> BizPalletInventory:
             actions.append('inventory-lock')
@@ -1119,18 +1172,24 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                 actions.append('site-mapping-lock')
                 if revoke_mapping:
                     raise PermissionException(data='', message='无权访问该客户现场')
+            else:
+                actions.append('site-access')
+                if revoke_initial_access:
+                    raise PermissionException(data='', message='无权访问该客户现场')
 
         stack, mocks = self.core._patch_record_path(
             inventory=inventory,
             site=locked_site,
-            site_scope=site_scope,
+            site_scope=AsyncMock(side_effect=read_site_scope),
             key_read=AsyncMock(side_effect=read_key),
             access=AsyncMock(side_effect=check_access),
         )
+        mocks['scan_pallet'].side_effect = scan_pallet
         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
+        self.db.rollback.side_effect = rollback
         return stack, mocks, actions
 
     async def test_exact_retry_rechecks_current_mapping_and_returns_without_mutation(self) -> None:
@@ -1156,7 +1215,6 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
             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)
@@ -1254,21 +1312,31 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                         self.fail(f'精确并发重试被旧库存状态提前拒绝：{exc.message}')
 
                 expected_actions = [
-                    'usage-snapshot-read',
+                    'preflight-scope',
+                    'usage-preflight-read',
+                    'qr-snapshot-read',
+                    'preflight-rollback',
                     'inventory-lock',
+                    'site-access',
+                    'fresh-scope',
                     'pallet-lock',
                 ]
                 if expect_filling_lock:
                     expected_actions.append('filling-lock')
                 expected_actions.extend(
                     [
-                        'usage-current-read',
+                        'usage-fresh-read',
                         'site-lock',
                         'site-mapping-lock',
+                        'result-rollback',
                     ]
                 )
                 self.assertEqual(expected_actions, actions)
-                self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
+                self.assertEqual(
+                    [{'site_ids': [9]}, {'site_ids': [9]}],
+                    [call.kwargs for call in key_read.await_args_list],
+                )
+                self.assertEqual(2, mocks['scope_read'].await_count)
                 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()
@@ -1280,7 +1348,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                 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(2, self.db.rollback.await_count)
                 self.assertEqual(
                     inventory_before,
                     (
@@ -1293,6 +1361,144 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                 )
                 self.assertEqual(61, result.result)
 
+    async def test_concurrent_retry_uses_fresh_scope_granted_during_inventory_wait(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, actions = self._patch_concurrent_retry_path(
+            inventory=inventory,
+            recovered=recovered,
+            scope_sequence=([], [9]),
+        )
+        with stack:
+            try:
+                result = await UsageService.record_usage_services(self.db, self.page, self.user)
+            except ServiceException as exc:
+                self.fail(f'锁等待期间新增授权后未使用 fresh scope：{exc.message}')
+
+        self.assertEqual(
+            [
+                'preflight-scope',
+                'usage-preflight-read',
+                'qr-snapshot-read',
+                'preflight-rollback',
+                'inventory-lock',
+                'site-access',
+                'fresh-scope',
+                'pallet-lock',
+                'usage-fresh-read',
+                'site-lock',
+                'site-mapping-lock',
+                'result-rollback',
+            ],
+            actions,
+        )
+        self.assertEqual(
+            [{'site_ids': []}, {'site_ids': [9]}],
+            [call.kwargs for call in mocks['key_read'].await_args_list],
+        )
+        mocks['filling_lock'].assert_not_awaited()
+        mocks['line_lock'].assert_not_awaited()
+        mocks['add_usage'].assert_not_awaited()
+        mocks['add_event'].assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+        self.assertEqual(61, result.result)
+
+    async def test_preflight_grant_revoked_before_fresh_scope_stays_hidden(self) -> None:
+        recovered = self._exact_usage()
+        stack, mocks, actions = self._patch_concurrent_retry_path(
+            inventory=make_inventory(),
+            recovered=recovered,
+            revoke_initial_access=True,
+        )
+        with stack:
+            try:
+                await UsageService.record_usage_services(self.db, self.page, self.user)
+            except PermissionException as exc:
+                self.assertEqual('无权访问该托盘', exc.message)
+            else:
+                self.fail('锁事务开始前已撤权时不得继续 fresh scope 或返回成功')
+
+        self.assertEqual(
+            [
+                'preflight-scope',
+                'usage-preflight-read',
+                'qr-snapshot-read',
+                'preflight-rollback',
+                'inventory-lock',
+                'site-access',
+                'result-rollback',
+            ],
+            actions,
+        )
+        self.assertEqual(1, mocks['scope_read'].await_count)
+        self.assertEqual(1, mocks['key_read'].await_count)
+        mocks['pallet_lock'].assert_not_awaited()
+        mocks['filling_lock'].assert_not_awaited()
+        mocks['site_lock'].assert_not_awaited()
+        mocks['add_usage'].assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
+    async def test_preflight_rollback_expires_scanned_pallet_and_fresh_lock_rechecks_qr(self) -> None:
+        class ExpiringPalletProbe:
+            def __init__(self, pallet_id: int) -> None:
+                self.expired = False
+                self.pallet_id_reads = 0
+                self._pallet_id = pallet_id
+
+            @property
+            def pallet_id(self) -> int:
+                if self.expired:
+                    raise AssertionError('rollback 后禁止读取旧 scanned pallet ID')
+                self.pallet_id_reads += 1
+                return self._pallet_id
+
+            def expire(self) -> None:
+                self.expired = True
+
+            def __getattr__(self, name: str) -> NoReturn:
+                raise AssertionError(f'预检 scanned pallet 只允许读取纯整数 ID：{name}')
+
+        scanned = ExpiringPalletProbe(pallet_id=2)
+        fresh_locked_pallet = make_pallet(qr_token='fresh-qr-mismatch')
+        scope = AsyncMock(side_effect=[[9], [9]])
+        key_read = AsyncMock(return_value=None)
+        stack, mocks = self.core._patch_record_path(
+            scanned_pallet=scanned,
+            pallet=fresh_locked_pallet,
+            site_scope=scope,
+            key_read=key_read,
+        )
+
+        async def rollback() -> None:
+            scanned.expire()
+
+        self.db.rollback.side_effect = rollback
+        with stack, self.assertRaises(ServiceException) as raised:
+            await UsageService.record_usage_services(self.db, self.page, self.user)
+
+        self.assertEqual('托盘二维码与库存不一致', raised.exception.message)
+        self.assertTrue(scanned.expired)
+        self.assertEqual(1, scanned.pallet_id_reads)
+        mocks['inventory_lock'].assert_awaited_once_with(self.db, 2)
+        mocks['pallet_lock'].assert_awaited_once_with(self.db, [2])
+        self.assertEqual(2, scope.await_count)
+        self.assertEqual(
+            [{'site_ids': [9]}, {'site_ids': [9]}],
+            [call.kwargs for call in key_read.await_args_list],
+        )
+        mocks['add_usage'].assert_not_awaited()
+        mocks['add_event'].assert_not_awaited()
+        self.db.commit.assert_not_awaited()
+        self.assertEqual(2, self.db.rollback.await_count)
+
     async def test_concurrent_exact_retry_rechecks_current_mapping_before_success(self) -> None:
         inventory = make_inventory(
             current_line_id=4,
@@ -1319,7 +1525,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
             else:
                 self.fail('current mapping 已撤权时不得返回幂等成功')
 
-        self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
+        self.assertEqual(
+            [{'site_ids': [9]}, {'site_ids': [9]}],
+            [call.kwargs for call in key_read.await_args_list],
+        )
+        self.assertEqual(2, mocks['scope_read'].await_count)
         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()
@@ -1327,9 +1537,9 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
         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(2, self.db.rollback.await_count)
 
-    async def test_concurrent_current_read_rejects_hash_or_actual_site_conflict(self) -> None:
+    async def test_concurrent_fresh_key_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')),
@@ -1348,7 +1558,7 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                 stack, mocks, _ = self._patch_concurrent_retry_path(
                     inventory=inventory,
                     recovered=recovered,
-                    site_scope=[9, 10],
+                    scope_sequence=([9, 10], [9, 10]),
                 )
                 key_read = mocks['key_read']
                 with stack:
@@ -1362,7 +1572,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                     else:
                         self.fail('锁后命中的摘要或实际现场冲突不得返回成功')
 
-                self.assertEqual([False, True], [call.kwargs['for_update'] for call in key_read.await_args_list])
+                self.assertEqual(
+                    [{'site_ids': [9, 10]}, {'site_ids': [9, 10]}],
+                    [call.kwargs for call in key_read.await_args_list],
+                )
+                self.assertEqual(2, mocks['scope_read'].await_count)
                 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()
@@ -1370,11 +1584,11 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                 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(2, self.db.rollback.await_count)
 
     async def test_integrity_error_recovers_only_fresh_scoped_exact_result(self) -> None:
         recovered = self._exact_usage()
-        scope = AsyncMock(side_effect=[[9], [9]])
+        scope = AsyncMock(side_effect=[[9], [9], [9]])
         key_read = AsyncMock(side_effect=[None, None, recovered])
         add_usage = AsyncMock(side_effect=integrity_error())
         stack, mocks = self.core._patch_record_path(
@@ -1386,11 +1600,14 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
             result = await UsageService.record_usage_services(self.db, self.page, self.user)
 
         self.assertEqual(61, result.result)
-        self.assertEqual(2, scope.await_count)
+        self.assertEqual(3, scope.await_count)
         self.assertEqual(3, key_read.await_count)
-        self.assertEqual([False, True, False], [call.kwargs['for_update'] for call in key_read.await_args_list])
+        self.assertEqual(
+            [{'site_ids': [9]}, {'site_ids': [9]}, {'site_ids': [9]}],
+            [call.kwargs 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.assertEqual(3, self.db.rollback.await_count)
         self.db.commit.assert_not_awaited()
         mocks['add_event'].assert_not_awaited()
 
@@ -1406,11 +1623,15 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
                 with stack, self.assertRaises(ServiceException) as raised:
                     await UsageService.record_usage_services(self.db, self.page, self.user)
                 self.assertTrue('幂等' in raised.exception.message or '冲突' in raised.exception.message)
-                self.assertEqual(2, self.db.rollback.await_count)
+                self.assertEqual(
+                    [{'site_ids': [9]}, {'site_ids': [9]}, {'site_ids': [9]}],
+                    [call.kwargs for call in key_read.await_args_list],
+                )
+                self.assertEqual(3, self.db.rollback.await_count)
                 self.db.commit.assert_not_awaited()
 
     async def test_integrity_recovery_rechecks_fresh_scope_and_rejects_revoked_site(self) -> None:
-        scope = AsyncMock(side_effect=[[9], []])
+        scope = AsyncMock(side_effect=[[9], [9], []])
         key_read = AsyncMock(return_value=None)
         stack, mocks = self.core._patch_record_path(
             site_scope=scope,
@@ -1420,21 +1641,24 @@ class TestUsageIdempotency(unittest.IsolatedAsyncioTestCase):
         with stack, self.assertRaises(PermissionException):
             await UsageService.record_usage_services(self.db, self.page, self.user)
 
-        self.assertEqual(2, scope.await_count)
+        self.assertEqual(3, scope.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(
-        self.assertEqual(2, self.db.rollback.await_count)
+            [{'site_ids': [9]}, {'site_ids': [9]}],
+            [call.kwargs for call in key_read.await_args_list],
+        )
+        self.assertEqual(3, self.db.rollback.await_count)
         mocks['add_event'].assert_not_awaited()
         self.db.commit.assert_not_awaited()
 
-    async def test_unexpected_write_failure_rolls_back_once_and_never_commits(self) -> None:
+    async def test_unexpected_write_failure_rolls_back_lock_transaction_and_never_commits(self) -> None:
         stack, _ = self.core._patch_record_path(
             add_event=AsyncMock(side_effect=RuntimeError('event failed')),
         )
         with stack, self.assertRaises(RuntimeError):
             await UsageService.record_usage_services(self.db, self.page, self.user)
 
-        self.db.rollback.assert_awaited_once()
+        self.assertEqual(2, self.db.rollback.await_count)
         self.db.commit.assert_not_awaited()
 
     async def test_request_hash_failure_also_rolls_back(self) -> None: