From a0ac70992f2d6d6d0a02d87becf83758f0ff6129 Mon Sep 17 00:00:00 2001 From: Sisyphus Date: Sat, 27 Jun 2026 12:47:03 +0800 Subject: [PATCH 1/3] fix: worker restart collection mismatch in loadscope/loadgroup scheduler Fix two related issues when a worker crashes and is restarted: 1. **Issue #1189**: When a worker crashes, its entry in registered_collections was not cleaned up. This caused collection_is_completed to remain True, and replacement workers whose collections didn't match the original would have their collections silently dropped, leading to KeyError when trying to assign work. 2. **Issue #1323**: When a worker crashed, ALL work units (including completed ones) were added back to the workqueue. When these completed work units were assigned to a new worker, nodeids_indexes would be empty, causing the worker to hang waiting for work that never arrives. Changes: - remove_node(): Now removes the crashed node from registered_collections - remove_node(): Only requeues work units that have pending (uncompleted) items - add_node_collection(): Always registers collection even if it doesn't match, allowing _check_nodes_have_same_collection() to report the mismatch properly Fixes #1189, #1323 --- src/xdist/scheduler/loadscope.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/src/xdist/scheduler/loadscope.py b/src/xdist/scheduler/loadscope.py index 73162dcd..8fbe8fbd 100644 --- a/src/xdist/scheduler/loadscope.py +++ b/src/xdist/scheduler/loadscope.py @@ -179,6 +179,10 @@ def remove_node(self, node: WorkerController) -> str | None: node has no more pending items. """ workload = self.assigned_work.pop(node) + # Also remove from registered_collections to allow replacement workers + # to properly register their collections (#1189) + self.registered_collections.pop(node, None) + if not self._pending_of(workload): return None @@ -196,8 +200,14 @@ 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) + # Only requeue work units that have pending items (#1323) + # Completed work units should not be requeued as they would + # result in empty nodeids_indexes causing workers to hang + pending_work = OrderedDict() + for scope, work_unit in workload.items(): + if not all(work_unit.values()): + pending_work[scope] = work_unit + self.workqueue.update(pending_work) for node in self.assigned_work: self._reschedule(node) @@ -231,7 +241,9 @@ def add_node_collection( self.collection, collection, other_node.gateway.id, node.gateway.id ) self.log(msg) - return + # Still register collection so the node can participate + # _check_nodes_have_same_collection() will report the mismatch + # when schedule() is called (#1189) self.registered_collections[node] = list(collection) From 0a31c8675195a28df608dff84f699f4fcfbba0d7 Mon Sep 17 00:00:00 2001 From: C1-BA-B1-F3 Date: Thu, 20 Aug 2026 21:03:49 +0800 Subject: [PATCH 2/3] test: cover loadscope worker restart collection handling Add unit tests for clearing registered collections on node removal, avoiding requeue of completed work units, and still registering mismatched late collections so replacement workers can participate. --- testing/test_dsession.py | 106 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 680b7ae0..4a02a2a1 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -16,6 +16,7 @@ from xdist.scheduler import EachScheduling from xdist.scheduler import LoadScheduling from xdist.scheduler import WorkStealingScheduling +from xdist.scheduler import LoadScopeScheduling from xdist.workermanage import WorkerController @@ -632,3 +633,108 @@ 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_clears_registered_collection( + self, pytester: pytest.Pytester + ) -> None: + """Replacement workers must be able to re-register collections (#1189).""" + config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope") + sched = LoadScopeScheduling(config) + node1, node2 = MockNode(), MockNode() + node1.gateway.id = "gw0" + node2.gateway.id = "gw1" + sched.add_node(node1) + sched.add_node(node2) + + collection = [ + "test_a.py::test_1", + "test_a.py::test_2", + "test_b.py::test_1", + ] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + assert sched.collection_is_completed + sched.schedule() + + # Give node1 unfinished work, then crash it. + assert node1 in sched.registered_collections + crashitem = sched.remove_node(node1) + assert crashitem is not None + assert node1 not in sched.registered_collections + + # A replacement worker with the same collection should register cleanly. + replacement = MockNode() + replacement.gateway.id = "gw0-replaced" + sched.add_node(replacement) + sched.add_node_collection(replacement, collection) + assert replacement in sched.registered_collections + assert sched.registered_collections[replacement] == collection + + def test_remove_node_does_not_requeue_completed_work_units( + self, pytester: pytest.Pytester + ) -> None: + """Completed scopes must not be requeued on crash (#1323).""" + config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope") + sched = LoadScopeScheduling(config) + node1, node2 = MockNode(), MockNode() + node1.gateway.id = "gw0" + node2.gateway.id = "gw1" + sched.add_node(node1) + sched.add_node(node2) + + collection = [ + "test_a.py::test_1", + "test_a.py::test_2", + "test_b.py::test_1", + ] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + sched.schedule() + + # Force a deterministic assigned workload on node1: + # one fully completed scope and one pending scope. + sched.assigned_work[node1] = { + "test_a.py": { + "test_a.py::test_1": True, + "test_a.py::test_2": True, + }, + "test_b.py": { + "test_b.py::test_1": False, + }, + } + # Ensure workqueue is empty before crash so we can observe requeueing. + sched.workqueue.clear() + + crashitem = sched.remove_node(node1) + assert crashitem == "test_b.py::test_1" + assert "test_b.py" in sched.workqueue + assert "test_a.py" not in sched.workqueue + # Pending scope should still contain the unfinished test marker. + assert sched.workqueue["test_b.py"]["test_b.py::test_1"] is False + + def test_add_node_collection_registers_mismatch_after_schedule( + self, pytester: pytest.Pytester + ) -> None: + """Mismatched collections are still registered after scheduling (#1189).""" + config = pytester.parseconfig("--tx=2*popen", "--dist=loadscope") + sched = LoadScopeScheduling(config) + node1, node2 = MockNode(), MockNode() + node1.gateway.id = "gw0" + node2.gateway.id = "gw1" + sched.add_node(node1) + sched.add_node(node2) + + collection = ["test_a.py::test_1", "test_b.py::test_1"] + sched.add_node_collection(node1, collection) + sched.add_node_collection(node2, collection) + sched.schedule() + + late = MockNode() + late.gateway.id = "gw2" + mismatched = ["test_a.py::test_1", "test_c.py::test_1"] + sched.add_node(late) + sched.add_node_collection(late, mismatched) + assert late in sched.registered_collections + assert sched.registered_collections[late] == mismatched From f5174bc63e75a4090b700132bb57f40e7777bc0d Mon Sep 17 00:00:00 2001 From: "pre-commit-ci[bot]" <66853113+pre-commit-ci[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 13:04:11 +0000 Subject: [PATCH 3/3] [pre-commit.ci] auto fixes from pre-commit.com hooks for more information, see https://pre-commit.ci --- testing/test_dsession.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/testing/test_dsession.py b/testing/test_dsession.py index 4a02a2a1..11e53b1f 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -15,8 +15,8 @@ from xdist.report import report_collection_diff from xdist.scheduler import EachScheduling from xdist.scheduler import LoadScheduling -from xdist.scheduler import WorkStealingScheduling from xdist.scheduler import LoadScopeScheduling +from xdist.scheduler import WorkStealingScheduling from xdist.workermanage import WorkerController