From 33228faffc5376366c95558e95869ed671483871 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=84=A6=E9=BE=99=E8=A8=80?= Date: Sun, 26 Jul 2026 21:43:55 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E8=A1=A5=E5=BC=BA?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E5=B9=B6=E5=8F=91=E5=B9=82=E7=AD=89=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../module_business/dao/operation_dao.py | 9 +- .../module_business/service/usage_service.py | 36 ++- .../module_business/test_inventory_dao.py | 62 +++- .../module_business/test_usage_service.py | 265 +++++++++++++++++- 4 files changed, 359 insertions(+), 13 deletions(-) diff --git a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py b/shierfeng-fastapi-backend/module_business/dao/operation_dao.py index ef329b2..59c5a14 100644 --- a/shierfeng-fastapi-backend/module_business/dao/operation_dao.py +++ b/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 diff --git a/shierfeng-fastapi-backend/module_business/service/usage_service.py b/shierfeng-fastapi-backend/module_business/service/usage_service.py index e25d7bb..7ba1861 100644 --- a/shierfeng-fastapi-backend/module_business/service/usage_service.py +++ b/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='托盘活动装填周期不存在或已关闭') diff --git a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py b/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py index c7503ac..c1d9f49 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_inventory_dao.py +++ b/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, diff --git a/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py b/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py index cbf75bb..3ee75e0 100644 --- a/shierfeng-fastapi-backend/tests/module_business/test_usage_service.py +++ b/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([9], key_read.await_args_list[1].kwargs['site_ids']) + 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([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()