修复：完善旧数据纳入事务读取

shierfeng-fastapi-backend/module_business/domain/test_batch_graph.py

@@ -214,18 +214,29 @@ class TestBatchGraph:
         return edges
 
     @staticmethod
-    async def _existing_ids(db: AsyncSession, table_name: str, ids: set[int]) -> set[int]:
+    async def _existing_ids(
+        db: AsyncSession,
+        table_name: str,
+        ids: set[int],
+        *,
+        for_update: bool,
+    ) -> set[int]:
         if not ids:
             return set()
         primary_key = BATCH_MODELS[table_name][1]
-        result = await db.execute(select(primary_key).where(primary_key.in_(sorted(ids))).order_by(primary_key))
+        statement = select(primary_key).where(primary_key.in_(sorted(ids))).order_by(primary_key)
+        if for_update:
+            statement = statement.with_for_update()
+        result = await db.execute(statement)
         return {row[0] for row in result.all()}
 
     @classmethod
-    async def compute_closure(
+    async def _compute_closure(
         cls,
         db: AsyncSession,
         roots: list[AdoptionRootModel],
+        *,
+        for_update: bool,
     ) -> BatchSelection:
         selected: dict[str, set[int]] = {table_name: set() for table_name in BATCH_MODELS}
         requested: dict[str, set[int]] = defaultdict(set)
@@ -234,7 +245,12 @@ class TestBatchGraph:
 
         queue: deque[str] = deque()
         for table_name in sorted(requested):
-            existing = await cls._existing_ids(db, table_name, requested[table_name])
+            existing = await cls._existing_ids(
+                db,
+                table_name,
+                requested[table_name],
+                for_update=for_update,
+            )
             selected[table_name].update(existing)
             if existing:
                 queue.append(table_name)
@@ -251,7 +267,10 @@ class TestBatchGraph:
                 statement = select(edge.result_column).where(edge.filter_column.in_(sorted(pending_ids)))
                 if edge.source_type is not None:
                     statement = statement.where(BizPalletEvent.source_type == edge.source_type)
-                result = await db.execute(statement.order_by(edge.result_column))
+                statement = statement.order_by(edge.result_column)
+                if for_update:
+                    statement = statement.with_for_update()
+                result = await db.execute(statement)
                 related_ids = {row[0] for row in result.all() if row[0] is not None}
                 new_ids = related_ids - selected[edge.target_table]
                 if new_ids:
@@ -259,3 +278,20 @@ class TestBatchGraph:
                     queue.append(edge.target_table)
 
         return {table_name: frozenset(selected[table_name]) for table_name in BATCH_MODELS}
+
+    @classmethod
+    async def compute_closure(
+        cls,
+        db: AsyncSession,
+        roots: list[AdoptionRootModel],
+    ) -> BatchSelection:
+        return await cls._compute_closure(db, roots, for_update=False)
+
+    @classmethod
+    async def compute_locked_closure(
+        cls,
+        db: AsyncSession,
+        roots: list[AdoptionRootModel],
+    ) -> BatchSelection:
+        """使用锁定读计算执行阶段闭包，避免复用一致性快照。"""
+        return await cls._compute_closure(db, roots, for_update=True)