Test loadscope recovery after worker failure

changelog/1378.bugfix

@@ -0,0 +1 @@
+Avoid re-queueing completed loadscope work units when a worker crashes.

src/xdist/remote.py

@@ -261,7 +261,14 @@ def pytest_collection_modifyitems(
                     gnames.add(str(name))
                 if not gnames:
                     continue
-                item._nodeid = f"{item.nodeid}@{'_'.join(sorted(gnames))}"
+                nodeid = f"{item.nodeid}@{'_'.join(sorted(gnames))}"
+                item_id = getattr(item, "id", None)
+                if item_id is not None and hasattr(item_id, "parse"):
+                    # pytest 9.2 stores the structured node ID in ``_id``;
+                    # assigning ``_nodeid`` is ignored there.
+                    item._id = item_id.parse(nodeid)  # type: ignore[attr-defined]
+                else:
+                    item._nodeid = nodeid
 
     @pytest.hookimpl
     def pytest_collection_finish(self, session: pytest.Session) -> None:

testing/test_dsession.py

@@ -15,6 +15,7 @@
 from xdist.report import report_collection_diff
 from xdist.scheduler import EachScheduling
 from xdist.scheduler import LoadScheduling
+from xdist.scheduler import LoadScopeScheduling
 from xdist.scheduler import WorkStealingScheduling
 from xdist.workermanage import WorkerController
 
@@ -290,6 +291,36 @@ def pytest_collectreport(self, report: pytest.CollectReport) -> None:
         assert "Different tests were collected between" in rep.longrepr
 
 
+class TestLoadScopeScheduling:
+    def test_remove_node_does_not_requeue_completed_scopes(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        config = pytester.parseconfig("--tx=2*popen")
+        sched = LoadScopeScheduling(config)
+        node1, node2 = MockNode(), MockNode()
+        sched.add_node(node1)
+        sched.add_node(node2)
+        collection = [f"test_{i}.py::test_case" for i in range(6)]
+        sched.add_node_collection(node1, collection)
+        sched.add_node_collection(node2, collection)
+        sched.schedule()
+
+        sched.mark_test_complete(node1, 0)
+        assert any(
+            completed
+            for work_unit in sched.assigned_work[node1].values()
+            for completed in work_unit.values()
+        )
+
+        crashitem = sched.remove_node(node1)
+
+        assert crashitem is not None
+        assert all(
+            any(not completed for completed in work_unit.values())
+            for work_unit in sched.workqueue.values()
+        )
+
+
 class TestWorkStealingScheduling:
     def test_ideal_case(self, pytester: pytest.Pytester) -> None:
         config = pytester.parseconfig("--tx=2*popen")