修复：消除托盘与用户分配查询竞态

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

@@ -2,12 +2,13 @@ from datetime import datetime
 from typing import Any, Literal
 
 from pydantic import BaseModel
-from sqlalchemy import ColumnElement, Select, delete, func, or_, select, update
+from sqlalchemy import ColumnElement, Select, and_, delete, func, or_, select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm.attributes import InstrumentedAttribute
 
 from common.vo import PageModel
 from module_admin.entity.do.user_do import SysUser
+from module_business.domain.enums import PalletFlowState
 from module_business.entity.do.master_do import (
     BizCustomer,
     BizCustomerSite,
@@ -20,10 +21,13 @@ from module_business.entity.do.master_do import (
 )
 from module_business.entity.do.operation_do import (
     BizFillingRecord,
+    BizInboundItem,
     BizInboundOrder,
     BizPalletEvent,
     BizPalletInventory,
+    BizReturnItem,
     BizReturnOrder,
+    BizShipmentItem,
     BizShipmentOrder,
     BizUsageRecord,
 )
@@ -74,6 +78,15 @@ _REFERENCE_COLUMNS: dict[MasterResource, tuple[tuple[InstrumentedAttribute, ...]
     ),
 }
 
+_PALLET_REMOVAL_BLOCKING_COLUMNS: tuple[InstrumentedAttribute, ...] = (
+    BizInboundItem.pallet_id,
+    BizFillingRecord.pallet_id,
+    BizShipmentItem.pallet_id,
+    BizUsageRecord.pallet_id,
+    BizReturnItem.pallet_id,
+    BizPalletEvent.pallet_id,
+)
+
 
 class MasterDao:
     """菌种、客户、现场、产线和仓库主数据数据库操作层。"""
@@ -99,8 +112,13 @@ class MasterDao:
         model: type[Any],
         id_column: InstrumentedAttribute,
         resource_id: int,
+        *,
+        for_update: bool = False,
     ) -> Any | None:
-        result = await db.execute(select(model).where(id_column == resource_id))
+        statement = select(model).where(id_column == resource_id)
+        if for_update:
+            statement = statement.with_for_update()
+        result = await db.execute(statement)
         return result.scalar_one_or_none()
 
     @staticmethod
@@ -245,8 +263,20 @@ class MasterDao:
         return await cls._paginate(db, statement, query_object, is_page)
 
     @classmethod
-    async def get_site(cls, db: AsyncSession, site_id: int) -> BizCustomerSite | None:
-        return await cls._get(db, BizCustomerSite, BizCustomerSite.site_id, site_id)
+    async def get_site(
+        cls,
+        db: AsyncSession,
+        site_id: int,
+        *,
+        for_update: bool = False,
+    ) -> BizCustomerSite | None:
+        return await cls._get(
+            db,
+            BizCustomerSite,
+            BizCustomerSite.site_id,
+            site_id,
+            for_update=for_update,
+        )
 
     @classmethod
     async def find_site_by_code(
@@ -343,8 +373,20 @@ class MasterDao:
         return await cls._paginate(db, statement, query_object, is_page)
 
     @classmethod
-    async def get_warehouse(cls, db: AsyncSession, warehouse_id: int) -> BizWarehouse | None:
-        return await cls._get(db, BizWarehouse, BizWarehouse.warehouse_id, warehouse_id)
+    async def get_warehouse(
+        cls,
+        db: AsyncSession,
+        warehouse_id: int,
+        *,
+        for_update: bool = False,
+    ) -> BizWarehouse | None:
+        return await cls._get(
+            db,
+            BizWarehouse,
+            BizWarehouse.warehouse_id,
+            warehouse_id,
+            for_update=for_update,
+        )
 
     @classmethod
     async def find_warehouse_by_code(cls, db: AsyncSession, warehouse_code: str) -> BizWarehouse | None:
@@ -391,6 +433,37 @@ class MasterDao:
         result = await db.execute(statement)
         return result.scalar_one_or_none()
 
+    @classmethod
+    async def get_pallet_removability(
+        cls,
+        db: AsyncSession,
+        pallet_ids: list[int],
+    ) -> dict[int, bool]:
+        """批量判断托盘是否仍为未入库且没有任何业务引用。"""
+        removability = dict.fromkeys(pallet_ids, False)
+        if not pallet_ids:
+            return removability
+
+        has_unstocked_inventory = (
+            select(BizPalletInventory.pallet_id)
+            .where(
+                BizPalletInventory.pallet_id == BizPallet.pallet_id,
+                BizPalletInventory.flow_state == PalletFlowState.UNSTOCKED.value,
+            )
+            .exists()
+        )
+        has_no_business_references = tuple(
+            ~select(column).where(column == BizPallet.pallet_id).exists() for column in _PALLET_REMOVAL_BLOCKING_COLUMNS
+        )
+        statement = select(
+            BizPallet.pallet_id,
+            and_(has_unstocked_inventory, *has_no_business_references).label('can_remove'),
+        ).where(BizPallet.pallet_id.in_(pallet_ids))
+        result = await db.execute(statement)
+        for pallet_id, can_remove in result.all():
+            removability[pallet_id] = bool(can_remove)
+        return removability
+
     @classmethod
     async def find_pallet_by_code(cls, db: AsyncSession, pallet_code: str) -> BizPallet | None:
         return await cls._find(db, BizPallet, BizPallet.pallet_code, pallet_code)
@@ -484,6 +557,16 @@ class MasterDao:
         )
         return list(result.all())
 
+    @classmethod
+    async def list_warehouse_assigned_user_ids(cls, db: AsyncSession, warehouse_id: int) -> list[int]:
+        result = await db.execute(select(BizUserWarehouse.user_id).where(BizUserWarehouse.warehouse_id == warehouse_id))
+        return list(result.scalars().all())
+
+    @classmethod
+    async def list_site_assigned_user_ids(cls, db: AsyncSession, site_id: int) -> list[int]:
+        result = await db.execute(select(BizUserSite.user_id).where(BizUserSite.site_id == site_id))
+        return list(result.scalars().all())
+
     @classmethod
     async def count_references(
         cls,

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

@@ -1,7 +1,7 @@
 import secrets
 from datetime import datetime
 from io import BytesIO
-from typing import Any, Literal
+from typing import Literal
 
 import qrcode
 from sqlalchemy.exc import IntegrityError
@@ -11,7 +11,6 @@ from common.vo import CrudResponseModel, PageModel
 from exceptions.exception import ServiceException
 from module_business.dao.access_dao import AccessDao
 from module_business.dao.master_dao import MasterDao
-from module_business.domain.enums import PalletFlowState
 from module_business.entity.do.master_do import BizPallet
 from module_business.entity.do.operation_do import BizPalletInventory
 from module_business.entity.vo.master_vo import (
@@ -58,18 +57,6 @@ class PalletService:
             raise ServiceException(message='托盘ID不能重复')
         return sorted(pallet_ids)
 
-    @classmethod
-    async def _can_remove(cls, db: AsyncSession, pallet_id: int) -> bool:
-        inventory = await MasterDao.get_pallet_inventory(db, pallet_id)
-        if inventory is None or inventory.flow_state != PalletFlowState.UNSTOCKED.value:
-            return False
-        return await MasterDao.count_pallet_events(db, pallet_id) == 0
-
-    @classmethod
-    async def _to_pallet_model(cls, db: AsyncSession, record: Any) -> PalletModel:
-        model = PalletModel.model_validate(record)
-        return model.model_copy(update={'can_remove': await cls._can_remove(db, model.pallet_id)})
-
     @classmethod
     async def _lock_pallet_inventory(cls, db: AsyncSession, pallet_id: int) -> tuple[BizPallet, BizPalletInventory]:
         """按库存、托盘顺序加锁，避免与流转事务形成反序死锁。"""
@@ -91,17 +78,25 @@ class PalletService:
         is_page: bool = True,
     ) -> PageModel | list[PalletModel]:
         result = await MasterDao.list_pallet(db, query_object, is_page)
+        records = result.rows if isinstance(result, PageModel) else result
+        models = [PalletModel.model_validate(record) for record in records]
+        removability = await MasterDao.get_pallet_removability(
+            db,
+            [model.pallet_id for model in models],
+        )
+        rows = [model.model_copy(update={'can_remove': removability.get(model.pallet_id, False)}) for model in models]
         if isinstance(result, PageModel):
-            rows = [await cls._to_pallet_model(db, row) for row in result.rows]
             return result.model_copy(update={'rows': rows})
-        return [await cls._to_pallet_model(db, row) for row in result]
+        return rows
 
     @classmethod
     async def detail_pallet(cls, db: AsyncSession, pallet_id: int) -> PalletModel:
         pallet = await MasterDao.get_pallet(db, pallet_id)
         if pallet is None:
             raise ServiceException(message='托盘不存在')
-        return await cls._to_pallet_model(db, pallet)
+        model = PalletModel.model_validate(pallet)
+        removability = await MasterDao.get_pallet_removability(db, [pallet_id])
+        return model.model_copy(update={'can_remove': removability.get(pallet_id, False)})
 
     @classmethod
     async def add_pallet(
@@ -179,11 +174,11 @@ class PalletService:
         try:
             pallet_ids = cls._extract_pallet_ids(value)
             for pallet_id in pallet_ids:
-                _, inventory = await cls._lock_pallet_inventory(db, pallet_id)
-                if inventory.flow_state != PalletFlowState.UNSTOCKED.value:
-                    raise ServiceException(message='托盘未处于未入库状态，不能删除')
-                if await MasterDao.count_pallet_events(db, pallet_id) > 0:
-                    raise ServiceException(message='托盘已有流转事件，不能删除')
+                await cls._lock_pallet_inventory(db, pallet_id)
+
+            removability = await MasterDao.get_pallet_removability(db, pallet_ids)
+            if not all(removability.get(pallet_id, False) for pallet_id in pallet_ids):
+                raise ServiceException(message='托盘状态或关联数据不允许删除')
 
             for pallet_id in pallet_ids:
                 await MasterDao.delete_pallet_inventory(db, pallet_id)
@@ -208,19 +203,16 @@ class PalletService:
             raise ServiceException(message=f'{_ASSIGNMENT_LABELS[resource]}不存在')
 
         users = await MasterDao.list_assignable_users(db)
-        result = []
-        association_method = getattr(AccessDao, f'list_user_{resource}_ids')
-        for user in users:
-            assigned_ids = await association_method(db, user.user_id)
-            result.append(
+        assigned_user_ids = set(await getattr(MasterDao, f'list_{resource}_assigned_user_ids')(db, resource_id))
+        return [
             BusinessUserOptionModel(
                 user_id=user.user_id,
                 user_name=user.user_name,
                 nick_name=user.nick_name,
-                    assigned=resource_id in assigned_ids,
+                assigned=user.user_id in assigned_user_ids,
             )
-            )
-        return result
+            for user in users
+        ]
 
     @classmethod
     async def list_warehouse_users(
@@ -248,7 +240,14 @@ class PalletService:
     ) -> CrudResponseModel:
         label = _ASSIGNMENT_LABELS[resource]
         try:
-            if await getattr(MasterDao, f'get_{resource}')(db, resource_id) is None:
+            if (
+                await getattr(MasterDao, f'get_{resource}')(
+                    db,
+                    resource_id,
+                    for_update=True,
+                )
+                is None
+            ):
                 raise ServiceException(message=f'{label}不存在')
             users = await MasterDao.list_assignable_users_by_ids(db, assignment.user_ids)
             if {user.user_id for user in users} != set(assignment.user_ids):

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

@@ -5,10 +5,12 @@ from datetime import datetime
 from types import SimpleNamespace
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import call as mock_call
 
 from fastapi import FastAPI
 from fastapi.routing import APIRoute
 from pydantic import ValidationError
+from sqlalchemy.dialects import mysql, postgresql
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -23,8 +25,22 @@ from module_business.controller import site_controller as site_controller_module
 from module_business.controller import warehouse_controller as warehouse_controller_module
 from module_business.dao.access_dao import AccessDao
 from module_business.dao.master_dao import MasterDao
-from module_business.entity.do.master_do import BizPallet
-from module_business.entity.do.operation_do import BizPalletInventory
+from module_business.entity.do.master_do import (
+    BizCustomerSite,
+    BizPallet,
+    BizUserSite,
+    BizUserWarehouse,
+    BizWarehouse,
+)
+from module_business.entity.do.operation_do import (
+    BizFillingRecord,
+    BizInboundItem,
+    BizPalletEvent,
+    BizPalletInventory,
+    BizReturnItem,
+    BizShipmentItem,
+    BizUsageRecord,
+)
 from module_business.entity.vo.master_vo import (
     BusinessUserOptionModel,
     DeletePalletModel,
@@ -122,6 +138,57 @@ class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase):
         self.assertIn("biz_pallet.asset_status = '0'", sql)
         self.assertIn('ORDER BY biz_pallet.pallet_id', sql)
 
+    async def test_pallet_removability_uses_one_complete_business_reference_query(self) -> None:
+        query_result = MagicMock()
+        query_result.all.return_value = [(2, True), (3, False)]
+        self.db.execute.return_value = query_result
+
+        actual = await MasterDao.get_pallet_removability(self.db, [1, 2, 3])
+
+        business_reference_columns = (
+            BizInboundItem.pallet_id,
+            BizFillingRecord.pallet_id,
+            BizShipmentItem.pallet_id,
+            BizUsageRecord.pallet_id,
+            BizReturnItem.pallet_id,
+            BizPalletEvent.pallet_id,
+        )
+        expected_references = {(column.class_.__tablename__, column.key) for column in business_reference_columns}
+        metadata_references = {
+            (table.name, column.name)
+            for table in BizPallet.metadata.tables.values()
+            if table.name != BizPalletInventory.__tablename__
+            for column in table.columns
+            if any(foreign_key.target_fullname == 'biz_pallet.pallet_id' for foreign_key in column.foreign_keys)
+        }
+        self.assertEqual(expected_references, metadata_references)
+        self.assertEqual({1: False, 2: True, 3: False}, actual)
+        self.db.execute.assert_awaited_once()
+
+        statement = self.db.execute.await_args.args[0]
+        for dialect in (mysql.dialect(), postgresql.dialect()):
+            with self.subTest(dialect=dialect.name):
+                sql = ' '.join(str(statement.compile(dialect=dialect, compile_kwargs={'literal_binds': True})).split())
+                self.assertIn('FROM biz_pallet', sql)
+                self.assertIn(
+                    'EXISTS (SELECT biz_pallet_inventory.pallet_id FROM biz_pallet_inventory '
+                    'WHERE biz_pallet_inventory.pallet_id = biz_pallet.pallet_id '
+                    "AND biz_pallet_inventory.flow_state = 'UNSTOCKED')",
+                    sql,
+                )
+                self.assertEqual(6, sql.count('NOT (EXISTS'))
+                for table_name, _ in expected_references:
+                    self.assertIn(
+                        f'NOT (EXISTS (SELECT {table_name}.pallet_id FROM {table_name} '
+                        f'WHERE {table_name}.pallet_id = biz_pallet.pallet_id))',
+                        sql,
+                    )
+
+    async def test_empty_pallet_removability_skips_database_query(self) -> None:
+        self.assertEqual({}, await MasterDao.get_pallet_removability(self.db, []))
+
+        self.db.execute.assert_not_awaited()
+
     async def test_add_pallet_and_inventory_only_flush_current_transaction(self) -> None:
         pallet = BizPallet(
             pallet_code='P-001',
@@ -208,6 +275,82 @@ class TestPalletMasterDao(unittest.IsolatedAsyncioTestCase):
         self.assertIn('biz_pallet.pallet_id = 9', compile_sql(statement))
         self.assertIn('FOR UPDATE', compile_sql(statement))
 
+    async def test_assignment_resource_write_reads_use_row_lock(self) -> None:
+        cases = (
+            (MasterDao.get_warehouse, BizWarehouse, 'warehouse_id', 7),
+            (MasterDao.get_site, BizCustomerSite, 'site_id', 9),
+        )
+        for method, model, id_field, resource_id in cases:
+            with self.subTest(resource=model.__tablename__):
+                self.db.reset_mock()
+                query_result = MagicMock()
+                record = SimpleNamespace()
+                query_result.scalar_one_or_none.return_value = record
+                self.db.execute.return_value = query_result
+
+                actual = await method(self.db, resource_id, for_update=True)
+
+                self.assertIs(record, actual)
+                statement = self.db.execute.await_args.args[0]
+                sql = compile_sql(statement)
+                self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
+                self.assertIn(f'{model.__tablename__}.{id_field} = {resource_id}', sql)
+                self.assertIn('FOR UPDATE', sql)
+
+    async def test_assignment_resource_reads_are_unlocked_by_default(self) -> None:
+        cases = (
+            (MasterDao.get_warehouse, BizWarehouse, 7),
+            (MasterDao.get_site, BizCustomerSite, 9),
+        )
+        for method, model, resource_id in cases:
+            with self.subTest(resource=model.__tablename__):
+                self.db.reset_mock()
+                query_result = MagicMock()
+                query_result.scalar_one_or_none.return_value = SimpleNamespace()
+                self.db.execute.return_value = query_result
+
+                await method(self.db, resource_id)
+
+                statement = self.db.execute.await_args.args[0]
+                self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
+                self.assertNotIn('FOR UPDATE', compile_sql(statement))
+
+    async def test_assigned_user_id_queries_target_exact_resource_association(self) -> None:
+        cases = (
+            (
+                MasterDao.list_warehouse_assigned_user_ids,
+                BizUserWarehouse,
+                BizUserWarehouse.user_id,
+                'warehouse_id',
+                7,
+            ),
+            (
+                MasterDao.list_site_assigned_user_ids,
+                BizUserSite,
+                BizUserSite.user_id,
+                'site_id',
+                9,
+            ),
+        )
+        for method, model, selected_column, id_field, resource_id in cases:
+            with self.subTest(resource=model.__tablename__):
+                self.db.reset_mock()
+                query_result = MagicMock()
+                query_result.scalars.return_value.all.return_value = [2, 5]
+                self.db.execute.return_value = query_result
+
+                actual = await method(self.db, resource_id)
+
+                self.assertEqual([2, 5], actual)
+                self.db.execute.assert_awaited_once()
+                statement = self.db.execute.await_args.args[0]
+                self.assertEqual([selected_column], list(statement.selected_columns))
+                self.assertTrue(statement.get_final_froms()[0].compare(model.__table__))
+                self.assertIn(
+                    f'{model.__tablename__}.{id_field} = {resource_id}',
+                    compile_sql(statement),
+                )
+
     async def test_assignable_user_queries_filter_disabled_and_deleted_rows(self) -> None:
         list_result = MagicMock()
         list_result.all.return_value = ()
@@ -386,49 +529,56 @@ class TestPalletQueryAndEditService(unittest.IsolatedAsyncioTestCase):
     def setUp(self) -> None:
         self.db = MagicMock(spec=AsyncSession)
 
-    async def test_list_computes_can_remove_from_inventory_and_events(self) -> None:
+    async def test_list_computes_can_remove_with_one_bulk_query(self) -> None:
         page = PageModel[Any](
-            rows=[make_pallet_record(1), make_pallet_record(2)],
+            rows=[make_pallet_record(pallet_id) for pallet_id in range(1, 1001)],
             pageNum=1,
-            pageSize=10,
-            total=2,
+            pageSize=1000,
+            total=1000,
             hasNext=False,
         )
         query = PalletPageQueryModel()
+        list_pallet = AsyncMock(return_value=page)
+        removability = AsyncMock(return_value={pallet_id: pallet_id % 2 == 1 for pallet_id in range(1, 1001)})
+        get_inventory = AsyncMock(return_value=SimpleNamespace(flow_state='UNSTOCKED'))
+        count_events = AsyncMock(return_value=0)
         with (
-            patch.object(MasterDao, 'list_pallet', new=AsyncMock(return_value=page)),
-            patch.object(
-                MasterDao,
-                'get_pallet_inventory',
-                new=AsyncMock(
-                    side_effect=[
-                        SimpleNamespace(flow_state='UNSTOCKED'),
-                        SimpleNamespace(flow_state='UNSTOCKED'),
-                    ]
-                ),
-            ),
-            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=[0, 1])),
+            patch.object(MasterDao, 'list_pallet', new=list_pallet),
+            patch.object(MasterDao, 'get_pallet_removability', new=removability, create=True),
+            patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory),
+            patch.object(MasterDao, 'count_pallet_events', new=count_events),
         ):
             actual = await PalletService.list_pallet(self.db, query, is_page=True)
 
         self.assertIsInstance(actual, PageModel)
+        self.assertEqual(1000, len(actual.rows))
         self.assertTrue(all(isinstance(row, PalletModel) for row in actual.rows))
-        self.assertEqual([True, False], [row.can_remove for row in actual.rows])
+        self.assertEqual(
+            [pallet_id % 2 == 1 for pallet_id in range(1, 1001)],
+            [row.can_remove for row in actual.rows],
+        )
+        list_pallet.assert_awaited_once_with(self.db, query, True)
+        removability.assert_awaited_once_with(self.db, list(range(1, 1001)))
+        get_inventory.assert_not_awaited()
+        count_events.assert_not_awaited()
 
     async def test_detail_computes_false_when_inventory_is_not_unstocked(self) -> None:
+        removability = AsyncMock(return_value={3: False})
+        get_inventory = AsyncMock(return_value=SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE'))
+        count_events = AsyncMock(return_value=0)
         with (
             patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(3))),
-            patch.object(
-                MasterDao,
-                'get_pallet_inventory',
-                new=AsyncMock(return_value=SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE')),
-            ),
-            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)),
+            patch.object(MasterDao, 'get_pallet_removability', new=removability, create=True),
+            patch.object(MasterDao, 'get_pallet_inventory', new=get_inventory),
+            patch.object(MasterDao, 'count_pallet_events', new=count_events),
         ):
             actual = await PalletService.detail_pallet(self.db, pallet_id=3)
 
         self.assertIsInstance(actual, PalletModel)
         self.assertFalse(actual.can_remove)
+        removability.assert_awaited_once_with(self.db, [3])
+        get_inventory.assert_not_awaited()
+        count_events.assert_not_awaited()
 
     async def test_detail_rejects_missing_pallet(self) -> None:
         with (
@@ -539,6 +689,10 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
             actions.append(f'count-events-{pallet_id}')
             return 0
 
+        async def get_removability(_db: AsyncSession, pallet_ids: list[int]) -> dict[int, bool]:
+            actions.append(f'get-removability-{pallet_ids}')
+            return dict.fromkeys(pallet_ids, True)
+
         async def delete_inventory(_db: AsyncSession, pallet_id: int) -> None:
             actions.append(f'delete-inventory-{pallet_id}')
 
@@ -553,6 +707,12 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
             patch.object(MasterDao, 'get_pallet', new=AsyncMock(side_effect=get_pallet)),
             patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(side_effect=get_inventory)),
             patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(side_effect=count_events)),
+            patch.object(
+                MasterDao,
+                'get_pallet_removability',
+                new=AsyncMock(side_effect=get_removability),
+                create=True,
+            ),
             patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock(side_effect=delete_inventory)),
             patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=delete_pallet)),
         ):
@@ -562,10 +722,9 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
             [
                 'get-inventory-4-True',
                 'get-pallet-4-True',
-                'count-events-4',
                 'get-inventory-5-True',
                 'get-pallet-5-True',
-                'count-events-5',
+                'get-removability-[4, 5]',
                 'delete-inventory-4',
                 'delete-pallet-4',
                 'delete-inventory-5',
@@ -578,32 +737,26 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
         self.db.rollback.assert_not_awaited()
         self.assertTrue(result.is_success)
 
-    async def test_remove_rejects_invalid_state_before_any_delete(self) -> None:
+    async def test_remove_rejects_missing_pallet_or_inventory(self) -> None:
         cases = (
-            ('托盘不存在', None, None, 0),
-            ('托盘库存不存在', make_pallet_record(4), None, 0),
-            (
-                '托盘未处于未入库状态，不能删除',
-                make_pallet_record(4),
-                SimpleNamespace(flow_state='EMPTY_IN_WAREHOUSE'),
-                0,
-            ),
-            (
-                '托盘已有流转事件，不能删除',
-                make_pallet_record(4),
-                SimpleNamespace(flow_state='UNSTOCKED'),
-                1,
-            ),
+            ('托盘不存在', None),
+            ('托盘库存不存在', make_pallet_record(4)),
         )
-        for message, pallet, inventory, event_count in cases:
+        for message, pallet in cases:
             with self.subTest(message=message):
                 self.db.reset_mock()
                 delete_inventory = AsyncMock()
                 delete_pallet = AsyncMock()
+                removability = AsyncMock()
                 with (
                     patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=pallet)),
-                    patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=inventory)),
-                    patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=event_count)),
+                    patch.object(MasterDao, 'get_pallet_inventory', new=AsyncMock(return_value=None)),
+                    patch.object(
+                        MasterDao,
+                        'get_pallet_removability',
+                        new=removability,
+                        create=True,
+                    ),
                     patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory),
                     patch.object(MasterDao, 'delete_pallet', new=delete_pallet),
                     self.assertRaises(ServiceException) as raised,
@@ -611,12 +764,55 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
                     await PalletService.remove_pallet(self.db, 4)
 
                 self.assertEqual(message, raised.exception.message)
+                removability.assert_not_awaited()
+                delete_inventory.assert_not_awaited()
+                delete_pallet.assert_not_awaited()
+                self.db.commit.assert_not_awaited()
+                self.db.rollback.assert_awaited_once()
+
+    async def test_remove_rejects_any_ineligible_pallet_before_any_delete(self) -> None:
+        delete_inventory = AsyncMock()
+        delete_pallet = AsyncMock()
+        removability = AsyncMock(return_value={4: True, 5: False})
+        with (
+            patch.object(
+                MasterDao,
+                'get_pallet',
+                new=AsyncMock(side_effect=[make_pallet_record(4), make_pallet_record(5)]),
+            ),
+            patch.object(
+                MasterDao,
+                'get_pallet_inventory',
+                new=AsyncMock(
+                    side_effect=[
+                        SimpleNamespace(flow_state='UNSTOCKED'),
+                        SimpleNamespace(flow_state='UNSTOCKED'),
+                    ]
+                ),
+            ),
+            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)),
+            patch.object(
+                MasterDao,
+                'get_pallet_removability',
+                new=removability,
+                create=True,
+            ),
+            patch.object(MasterDao, 'delete_pallet_inventory', new=delete_inventory),
+            patch.object(MasterDao, 'delete_pallet', new=delete_pallet),
+            self.assertRaises(ServiceException) as raised,
+        ):
+            await PalletService.remove_pallet(self.db, DeletePalletModel(palletIds=[5, 4]))
+
+        self.assertEqual('托盘状态或关联数据不允许删除', raised.exception.message)
+        removability.assert_awaited_once_with(self.db, [4, 5])
         delete_inventory.assert_not_awaited()
         delete_pallet.assert_not_awaited()
         self.db.commit.assert_not_awaited()
         self.db.rollback.assert_awaited_once()
 
     async def test_remove_integrity_error_rolls_back_with_stable_failure(self) -> None:
+        removability = AsyncMock(return_value={4: True})
+        count_events = AsyncMock(return_value=0)
         with (
             patch.object(MasterDao, 'get_pallet', new=AsyncMock(return_value=make_pallet_record(4))),
             patch.object(
@@ -624,7 +820,13 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
                 'get_pallet_inventory',
                 new=AsyncMock(return_value=SimpleNamespace(flow_state='UNSTOCKED')),
             ),
-            patch.object(MasterDao, 'count_pallet_events', new=AsyncMock(return_value=0)),
+            patch.object(MasterDao, 'count_pallet_events', new=count_events),
+            patch.object(
+                MasterDao,
+                'get_pallet_removability',
+                new=removability,
+                create=True,
+            ),
             patch.object(MasterDao, 'delete_pallet_inventory', new=AsyncMock()),
             patch.object(MasterDao, 'delete_pallet', new=AsyncMock(side_effect=integrity_error())),
             self.assertRaises(ServiceException) as raised,
@@ -632,6 +834,8 @@ class TestPalletRemovalService(unittest.IsolatedAsyncioTestCase):
             await PalletService.remove_pallet(self.db, 4)
 
         self.assertEqual('删除托盘失败，托盘状态或关联数据已变化', raised.exception.message)
+        removability.assert_awaited_once_with(self.db, [4])
+        count_events.assert_not_awaited()
         self.db.rollback.assert_awaited_once()
         self.db.commit.assert_not_awaited()
 
@@ -659,25 +863,26 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase):
         ]
 
     async def test_assignment_get_returns_only_safe_fields_with_assigned_marker(self) -> None:
-        cases = (
-            ('warehouse', 7, [7], []),
-            ('site', 9, [9], []),
-        )
-        for resource, resource_id, first_ids, second_ids in cases:
+        for resource, resource_id in (('warehouse', 7), ('site', 9)):
             with self.subTest(resource=resource):
-                association_method = f'list_user_{resource}_ids'
+                get_resource = AsyncMock(return_value=SimpleNamespace())
+                list_users = AsyncMock(return_value=self.users)
+                list_assigned_ids = AsyncMock(return_value=[2])
+                list_user_resource_ids = AsyncMock()
                 with (
                     patch.object(
                         MasterDao,
                         f'get_{resource}',
-                        new=AsyncMock(return_value=SimpleNamespace()),
+                        new=get_resource,
                     ),
-                    patch.object(MasterDao, 'list_assignable_users', new=AsyncMock(return_value=self.users)),
+                    patch.object(MasterDao, 'list_assignable_users', new=list_users),
                     patch.object(
-                        AccessDao,
-                        association_method,
-                        new=AsyncMock(side_effect=[first_ids, second_ids]),
-                    ) as list_ids,
+                        MasterDao,
+                        f'list_{resource}_assigned_user_ids',
+                        new=list_assigned_ids,
+                        create=True,
+                    ),
+                    patch.object(AccessDao, f'list_user_{resource}_ids', new=list_user_resource_ids),
                 ):
                     actual = await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id)
 
@@ -689,24 +894,57 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase):
                     [option.model_dump(by_alias=True) for option in actual],
                 )
                 self.assertTrue(all(isinstance(option, BusinessUserOptionModel) for option in actual))
-                self.assertEqual(
-                    [(self.db, 2), (self.db, 5)],
-                    [call.args for call in list_ids.await_args_list],
-                )
+                get_resource.assert_awaited_once_with(self.db, resource_id)
+                list_users.assert_awaited_once_with(self.db)
+                list_assigned_ids.assert_awaited_once_with(self.db, resource_id)
+                list_user_resource_ids.assert_not_awaited()
+
+    async def test_assignment_get_uses_three_fixed_queries_for_1000_users(self) -> None:
+        users = [
+            SimpleNamespace(user_id=user_id, user_name=f'user-{user_id}', nick_name=None) for user_id in range(1, 1001)
+        ]
+        assigned_ids = [2, 500, 1000]
+        for resource, resource_id in (('warehouse', 7), ('site', 9)):
+            with self.subTest(resource=resource):
+                get_resource = AsyncMock(return_value=SimpleNamespace())
+                list_users = AsyncMock(return_value=users)
+                list_assigned = AsyncMock(return_value=assigned_ids)
+                list_user_resource_ids = AsyncMock()
+                with (
+                    patch.object(MasterDao, f'get_{resource}', new=get_resource),
+                    patch.object(MasterDao, 'list_assignable_users', new=list_users),
+                    patch.object(
+                        MasterDao,
+                        f'list_{resource}_assigned_user_ids',
+                        new=list_assigned,
+                        create=True,
+                    ),
+                    patch.object(AccessDao, f'list_user_{resource}_ids', new=list_user_resource_ids),
+                ):
+                    actual = await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id)
+
+                self.assertEqual(1000, len(actual))
+                self.assertEqual(assigned_ids, [option.user_id for option in actual if option.assigned])
+                get_resource.assert_awaited_once_with(self.db, resource_id)
+                list_users.assert_awaited_once_with(self.db)
+                list_assigned.assert_awaited_once_with(self.db, resource_id)
+                list_user_resource_ids.assert_not_awaited()
 
     async def test_assignment_get_rejects_missing_resource_before_user_query(self) -> None:
         cases = (('warehouse', 7, '仓库不存在'), ('site', 9, '客户现场不存在'))
         for resource, resource_id, message in cases:
             with self.subTest(resource=resource):
+                get_resource = AsyncMock(return_value=None)
                 list_users = AsyncMock()
                 with (
-                    patch.object(MasterDao, f'get_{resource}', new=AsyncMock(return_value=None)),
+                    patch.object(MasterDao, f'get_{resource}', new=get_resource),
                     patch.object(MasterDao, 'list_assignable_users', new=list_users),
                     self.assertRaises(ServiceException) as raised,
                 ):
                     await getattr(PalletService, f'list_{resource}_users')(self.db, resource_id)
 
                 self.assertEqual(message, raised.exception.message)
+                get_resource.assert_awaited_once_with(self.db, resource_id)
                 list_users.assert_not_awaited()
 
     async def test_assignment_put_validates_users_replaces_and_commits(self) -> None:
@@ -714,19 +952,25 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase):
         for resource, resource_id in (('warehouse', 7), ('site', 9)):
             with self.subTest(resource=resource):
                 self.db.reset_mock()
-                replace = AsyncMock()
+                calls = MagicMock()
+                get_resource_mock = AsyncMock(return_value=SimpleNamespace())
+                list_users_mock = AsyncMock(return_value=self.users)
+                replace_mock = AsyncMock()
+                calls.attach_mock(get_resource_mock, 'get_resource')
+                calls.attach_mock(list_users_mock, 'list_users')
+                calls.attach_mock(replace_mock, 'replace')
                 with (
                     patch.object(
                         MasterDao,
                         f'get_{resource}',
-                        new=AsyncMock(return_value=SimpleNamespace()),
+                        new=get_resource_mock,
                     ),
                     patch.object(
                         MasterDao,
                         'list_assignable_users_by_ids',
-                        new=AsyncMock(return_value=self.users),
-                    ) as list_users,
-                    patch.object(AccessDao, f'replace_{resource}_users', new=replace),
+                        new=list_users_mock,
+                    ),
+                    patch.object(AccessDao, f'replace_{resource}_users', new=replace_mock),
                 ):
                     result = await getattr(PalletService, f'assign_{resource}_users')(
                         self.db,
@@ -734,8 +978,17 @@ class TestBusinessUserAssignmentService(unittest.IsolatedAsyncioTestCase):
                         assignment,
                     )
 
-                list_users.assert_awaited_once_with(self.db, [2, 5])
-                replace.assert_awaited_once_with(self.db, resource_id, [2, 5])
+                self.assertEqual(
+                    [
+                        mock_call.get_resource(self.db, resource_id, for_update=True),
+                        mock_call.list_users(self.db, [2, 5]),
+                        mock_call.replace(self.db, resource_id, [2, 5]),
+                    ],
+                    calls.mock_calls,
+                )
+                get_resource_mock.assert_awaited_once_with(self.db, resource_id, for_update=True)
+                list_users_mock.assert_awaited_once_with(self.db, [2, 5])
+                replace_mock.assert_awaited_once_with(self.db, resource_id, [2, 5])
                 self.db.commit.assert_awaited_once()
                 self.db.rollback.assert_not_awaited()
                 self.assertTrue(result.is_success)