Fix loadscope-family schedulers hanging the run after a worker crash

changelog/784.bugfix.rst

@@ -0,0 +1 @@
+Fixed ``--dist=loadscope``, ``loadfile`` and ``loadgroup`` hanging indefinitely after a worker crash. The crashed worker's completed work units were returned to the work queue and could later be assigned as empty commands that a node can never report against, and a node re-scheduled a single test never started it, since a worker holds its last test in reserve until it is sent further work or told to shut down. The crashed test itself is also no longer re-executed by the replacement worker: it is reported as crashed, matching ``--dist=load``.

src/xdist/scheduler/loadscope.py

@@ -187,6 +187,11 @@ def remove_node(self, node: WorkerController) -> str | None:
             for nodeid, completed in work_unit.items():
                 if not completed:
                     crashitem = nodeid
+                    # The crashed test is reported as failed by
+                    # handle_crashitem, so mark it complete lest it be
+                    # rescheduled and crash the next node too. This matches
+                    # LoadScheduling, which pops the crashed item.
+                    work_unit[nodeid] = True
                     break
             else:
                 continue
@@ -196,8 +201,16 @@ def remove_node(self, node: WorkerController) -> str | None:
                 "Unable to identify crashitem on a workload with pending items"
             )
 
-        # Made uncompleted work unit available again
-        self.workqueue.update(workload)
+        # Make uncompleted work units available again. Work units in which
+        # every test has completed are dropped rather than re-queued: assigning
+        # one to a node would send it an empty "runtests" command, after which
+        # the node would never produce the test report that drives scheduling
+        # onwards, hanging the run (#784).
+        self.workqueue.update(
+            (scope, work_unit)
+            for scope, work_unit in workload.items()
+            if not all(work_unit.values())
+        )
 
         for node in self.assigned_work:
             self._reschedule(node)
@@ -320,20 +333,32 @@ def _reschedule(self, node: WorkerController) -> None:
         if node.shutting_down:
             return
 
-        # Check that more work is available
-        if not self.workqueue:
-            node.shutdown()
-            return
+        while True:
+            # Check that more work is available
+            if not self.workqueue:
+                node.shutdown()
+                return
 
-        self.log("Number of units waiting for node:", len(self.workqueue))
+            self.log("Number of units waiting for node:", len(self.workqueue))
 
-        # Check that the node is almost depleted of work
-        # 2: Heuristic of minimum tests to enqueue more work
-        if self._pending_of(self.assigned_work[node]) > 2:
-            return
+            # Check that the node is almost depleted of work
+            # 2: Heuristic of minimum tests to enqueue more work
+            if self._pending_of(self.assigned_work[node]) > 2:
+                return
+
+            # Pop one unit of work and assign it
+            self._assign_work_unit(node)
 
-        # Pop one unit of work and assign it
-        self._assign_work_unit(node)
+            # A worker holds its final pending test in reserve, not starting
+            # it until further work or a shutdown notice arrives, so a node
+            # must hold at least two pending tests to make progress (see the
+            # note about #277 in .schedule()). Keep assigning until it does,
+            # since only a test report drives scheduling onwards and a node
+            # left below that threshold never produces one; if the workqueue
+            # runs dry first, the loop shuts the node down, which is what
+            # releases its reserved test (#784).
+            if self._pending_of(self.assigned_work[node]) >= 2:
+                return
 
     def schedule(self) -> None:
         """Initiate distribution of the test collection.

testing/acceptance_test.py

@@ -1032,6 +1032,37 @@ def test(i): os._exit(1)
         )
         assert "INTERNALERROR" not in res.stdout.str()
 
+    def test_loadfile_crashed_worker(self, pytester: pytest.Pytester) -> None:
+        """A crashed worker's workload is rescheduled without hanging the
+        run (#784).
+
+        The worker crashes after completing the whole of test_a.py. Its
+        re-queued workload must not include the completed test_a.py work
+        unit (an empty command the replacement worker could never report
+        against), the crashed test must not run a second time, and the
+        replacement worker must be sent enough tests to make progress, since
+        a worker holds its last test in reserve until told what follows.
+        """
+        pytester.makepyfile(
+            test_a="""
+                def test_pass_1(): pass
+                def test_pass_2(): pass
+            """,
+            test_b="""
+                import os
+                def test_crash(): os._exit(1)
+                def test_after(): pass
+            """,
+        )
+        res = pytester.runpytest_subprocess("-n1", "--dist=loadfile", "-v", timeout=120)
+        res.stdout.fnmatch_lines(
+            [
+                "replacing crashed worker gw*",
+                "worker*crashed while running*test_crash*",
+                "*1 failed*3 passed*",
+            ]
+        )
+
     def test_max_worker_restart_die(self, pytester: pytest.Pytester) -> None:
         f = pytester.makepyfile(
             """

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
 
@@ -632,3 +633,85 @@ def test_get_workers_status_line(
     status_and_items: Sequence[tuple[WorkerStatus, int]], expected: str
 ) -> None:
     assert get_workers_status_line(status_and_items) == expected
+
+
+class TestLoadScopeScheduling:
+    def test_remove_node_requeues_only_pending_work_units(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        """A crashed node's completed work units must not return to the
+        workqueue: assigning one to a node sends an empty "runtests" command
+        against which the node can never report, hanging the run (#784). The
+        crashed test itself must not be re-queued either, since it is
+        reported as failed by handle_crashitem."""
+        config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope")
+        sched = LoadScopeScheduling(config)
+        node1, node2 = MockNode(), MockNode()
+        sched.add_node(node1)
+        sched.add_node(node2)
+        collection = [f"test_{m}.py::test_{i}" for m in "abcdef" for i in (1, 2)]
+        sched.add_node_collection(node1, collection)
+        sched.add_node_collection(node2, collection)
+        sched.schedule()
+        # node1 was assigned test_a.py and test_c.py.
+        assert node1.sent == [0, 1, 4, 5]
+
+        # node1 completes the whole of test_a.py, picking up test_e.py.
+        sched.mark_test_complete(node1, 0)
+        sched.mark_test_complete(node1, 1)
+        assert node1.sent == [0, 1, 4, 5, 8, 9]
+
+        # node1 crashes in test_c.py::test_1.
+        crashitem = sched.remove_node(node1)
+        assert crashitem == "test_c.py::test_1"
+
+        # The completed test_a.py unit is gone for good, the units with
+        # pending tests are re-queued, and the crashed test is marked
+        # completed so that it is not run a second time.
+        assert list(sched.workqueue.keys()) == ["test_f.py", "test_c.py", "test_e.py"]
+        for work_unit in sched.workqueue.values():
+            assert not all(work_unit.values())
+        assert sched.workqueue["test_c.py"] == {
+            "test_c.py::test_1": True,
+            "test_c.py::test_2": False,
+        }
+
+    def test_node_is_topped_up_until_it_can_report(
+        self, pytester: pytest.Pytester
+    ) -> None:
+        """A node must be given at least two pending tests, because a worker
+        does not start its last test until it is sent further work or told to
+        shut down. A replacement node given a single-test work unit would
+        otherwise never report, and a report is the only thing that drives
+        scheduling onwards (#784)."""
+        config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope")
+        sched = LoadScopeScheduling(config)
+        node1, node2 = MockNode(), MockNode()
+        sched.add_node(node1)
+        sched.add_node(node2)
+        collection = [
+            "test_a.py::test_1",
+            "test_a.py::test_2",
+            "test_a.py::test_3",
+            "test_b.py::test_1",
+            "test_b.py::test_2",
+            "test_b.py::test_3",
+            "test_c.py::test_1",
+            "test_d.py::test_1",
+            "test_e.py::test_1",
+        ]
+        sched.add_node_collection(node1, collection)
+        sched.add_node_collection(node2, collection)
+        sched.schedule()
+        assert node1.sent == [0, 1, 2]
+        assert node2.sent == [3, 4, 5]
+
+        # A replacement node arrives while single-test scopes are queued.
+        node3 = MockNode()
+        sched.add_node(node3)
+        sched.add_node_collection(node3, collection)
+        sched.schedule()
+
+        # One test is not enough to make progress: it must receive two.
+        assert node3.sent == [6, 7]
+        assert not node3.shutting_down