修复:统一退回锁序并避免提交后访问

This commit is contained in:
焦龙言
2026-07-27 00:49:29 +08:00
parent 80c2288122
commit c9180b3832
2 changed files with 121 additions and 13 deletions
@@ -533,6 +533,16 @@ class ReturnService:
page_object: ReturnRequestModel, page_object: ReturnRequestModel,
current_user: CurrentUserModel, current_user: CurrentUserModel,
) -> None: ) -> None:
warehouse = await MasterDao.get_warehouse(
query_db,
page_object.target_warehouse_id,
for_update=True,
)
if warehouse is None:
raise ServiceException(message='目标仓库不存在')
if warehouse.status != '0':
raise ServiceException(message='目标仓库已停用')
site = await MasterDao.get_site(query_db, page_object.source_site_id, for_update=True) site = await MasterDao.get_site(query_db, page_object.source_site_id, for_update=True)
await AccessService.assert_site_access( await AccessService.assert_site_access(
query_db, query_db,
@@ -545,16 +555,6 @@ class ReturnService:
if site.status != '0': if site.status != '0':
raise ServiceException(message='来源客户现场已停用') raise ServiceException(message='来源客户现场已停用')
warehouse = await MasterDao.get_warehouse(
query_db,
page_object.target_warehouse_id,
for_update=True,
)
if warehouse is None:
raise ServiceException(message='目标仓库不存在')
if warehouse.status != '0':
raise ServiceException(message='目标仓库已停用')
@classmethod @classmethod
async def _lock_and_reauthorize_request_site( async def _lock_and_reauthorize_request_site(
cls, cls,
@@ -733,9 +733,10 @@ class ReturnService:
update_time=current_time, update_time=current_time,
) )
return_order = await OperationDao.add_return_order(query_db, return_order) return_order = await OperationDao.add_return_order(query_db, return_order)
return_id = int(return_order.return_id)
await OperationDao.add_return_items( await OperationDao.add_return_items(
query_db, query_db,
cls._build_return_items(return_order.return_id, inventories, user_name, current_time), cls._build_return_items(return_id, inventories, user_name, current_time),
) )
for inventory in inventories: for inventory in inventories:
@@ -751,7 +752,7 @@ class ReturnService:
pallet_id=inventory.pallet_id, pallet_id=inventory.pallet_id,
event_type=PalletEventType.RETURN_REQUESTED.value, event_type=PalletEventType.RETURN_REQUESTED.value,
source_type='RETURN', source_type='RETURN',
source_id=return_order.return_id, source_id=return_id,
before_state=before_state, before_state=before_state,
after_state=PalletFlowState.RETURNING.value, after_state=PalletFlowState.RETURNING.value,
before_warehouse_id=before_warehouse_id, before_warehouse_id=before_warehouse_id,
@@ -770,7 +771,7 @@ class ReturnService:
await InventoryDao.add_event(query_db, inventory, event) await InventoryDao.add_event(query_db, inventory, event)
await query_db.commit() await query_db.commit()
return cls._success('申请成功', return_order.return_id) return cls._success('申请成功', return_id)
except IntegrityError: except IntegrityError:
await query_db.rollback() await query_db.rollback()
return await cls._recover_request_integrity_error( return await cls._recover_request_integrity_error(
@@ -640,6 +640,113 @@ class ReturnRequestTransactionRedTest(unittest.IsolatedAsyncioTestCase):
idempotencyKey='request-key', idempotencyKey='request-key',
) )
async def test_new_request_route_locks_warehouse_before_site_and_mapping(self) -> None:
actions: list[str] = []
async def lock_warehouse(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
actions.append('warehouse-row')
return SimpleNamespace(status='0')
async def lock_site(*_args: Any, **_kwargs: Any) -> SimpleNamespace:
actions.append('site-row')
return SimpleNamespace(status='0')
async def lock_site_mapping(*_args: Any, **_kwargs: Any) -> None:
actions.append('site-mapping')
warehouse_access = AsyncMock()
with (
patch.object(MasterDao, 'get_warehouse', new=AsyncMock(side_effect=lock_warehouse)) as warehouse_read,
patch.object(MasterDao, 'get_site', new=AsyncMock(side_effect=lock_site)) as site_read,
patch.object(
AccessService, 'assert_site_access', new=AsyncMock(side_effect=lock_site_mapping)
) as site_access,
patch.object(AccessService, 'assert_warehouse_access', new=warehouse_access),
):
await ReturnService._lock_new_request_route(self.db, self.page, self.user)
self.assertEqual(['warehouse-row', 'site-row', 'site-mapping'], actions)
warehouse_read.assert_awaited_once_with(self.db, 7, for_update=True)
site_read.assert_awaited_once_with(self.db, 9, for_update=True)
site_access.assert_awaited_once_with(self.db, self.user, 9, for_update=True)
warehouse_access.assert_not_awaited()
async def test_request_caches_return_id_before_commit_expires_order(self) -> None:
class ExpiringReturnOrder:
def __init__(self) -> None:
self.expired = False
self.return_id_reads = 0
@property
def return_id(self) -> int:
self.return_id_reads += 1
if self.expired:
raise RuntimeError('commit 后访问了已过期的 return_order.return_id')
return 13
inventories = [
make_inventory(2),
make_inventory(5, remaining=0, state=PalletFlowState.AT_SITE_EMPTY.value),
]
probe = ExpiringReturnOrder()
captured_items: list[BizReturnItem] = []
captured_events: list[BizPalletEvent] = []
async def add_items(_db: AsyncSession, items: list[BizReturnItem]) -> list[BizReturnItem]:
captured_items.extend(items)
return items
async def add_event(
_db: AsyncSession,
_inventory: BizPalletInventory,
event: BizPalletEvent,
) -> BizPalletEvent:
captured_events.append(event)
return event
async def expire_order() -> None:
probe.expired = True
self.db.commit.side_effect = expire_order
warehouse_access = AsyncMock()
with (
patch.object(AccessDao, 'list_user_site_ids', new=AsyncMock(return_value=[9])),
patch.object(
OperationDao,
'get_return_by_request_idempotency_key',
new=AsyncMock(return_value=None),
),
patch.object(InventoryDao, 'get_inventories_for_update', new=AsyncMock(return_value=inventories)),
patch.object(
InventoryDao,
'get_pallets_by_ids',
new=AsyncMock(return_value=[make_pallet(2), make_pallet(5)]),
),
patch.object(
OperationDao,
'get_active_filling_for_update',
new=AsyncMock(return_value=make_filling(2)),
),
patch.object(AccessService, 'assert_site_access', new=AsyncMock()),
patch.object(AccessService, 'assert_warehouse_access', new=warehouse_access),
patch.object(MasterDao, 'get_site', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
patch.object(MasterDao, 'get_warehouse', new=AsyncMock(return_value=SimpleNamespace(status='0'))),
patch.object(OperationDao, 'add_return_order', new=AsyncMock(return_value=probe)),
patch.object(OperationDao, 'add_return_items', new=AsyncMock(side_effect=add_items)),
patch.object(InventoryDao, 'add_event', new=AsyncMock(side_effect=add_event)),
):
try:
result = await ReturnService.request_return_services(self.db, self.page, self.user)
except RuntimeError as exc:
self.fail(str(exc))
self.assertEqual(13, result.result)
self.assertEqual(1, probe.return_id_reads)
self.assertEqual([13, 13], [item.return_id for item in captured_items])
self.assertEqual([13, 13], [event.source_id for event in captured_events])
warehouse_access.assert_not_awaited()
self.db.commit.assert_awaited_once()
async def test_concurrent_same_request_returns_winner_after_inventory_wait(self) -> None: # noqa: PLR0915 async def test_concurrent_same_request_returns_winner_after_inventory_wait(self) -> None: # noqa: PLR0915
request_hash = ReturnService._build_request_hash(self.page) request_hash = ReturnService._build_request_hash(self.page)
winner = make_return_order(request_hash=request_hash) winner = make_return_order(request_hash=request_hash)