Fix DownloaderAwarePriorityQueue tie-breaking across slots (#7351)

scrapy/pqueues.py

@@ -350,10 +350,38 @@ class DownloaderAwarePriorityQueue:
         self.crawler: Crawler = crawler
 
         self.pqueues: dict[str, ScrapyPriorityQueue] = {}  # slot -> priority queue
+        self._last_selected_slot: str | None = None
         if slot_startprios:
             for slot, startprios in slot_startprios.items():
                 self.pqueues[slot] = self.pqfactory(slot, startprios)
 
+    def _next_slot(self, stats: list[tuple[int, str]], *, update_state: bool) -> str:
+        last = self._last_selected_slot
+        min_active: int | None = None
+        best_slot: str | None = None
+        best_slot_after_last: str | None = None
+        for active, slot in stats:
+            if min_active is None or active < min_active:
+                min_active = active
+                best_slot = slot
+                best_slot_after_last = None
+                if last is not None and slot > last:
+                    best_slot_after_last = slot
+            elif active == min_active:
+                if best_slot is None or slot < best_slot:
+                    best_slot = slot
+                if (
+                    last is not None
+                    and slot > last
+                    and (best_slot_after_last is None or slot < best_slot_after_last)
+                ):
+                    best_slot_after_last = slot
+        assert best_slot is not None
+        slot = best_slot_after_last if best_slot_after_last is not None else best_slot
+        if update_state:
+            self._last_selected_slot = slot
+        return slot
+
     def pqfactory(
         self, slot: str, startprios: Iterable[int] = ()
     ) -> ScrapyPriorityQueue:
@@ -371,7 +399,7 @@ class DownloaderAwarePriorityQueue:
         if not stats:
             return None
 
-        slot = min(stats)[1]
+        slot = self._next_slot(stats, update_state=True)
         queue = self.pqueues[slot]
         request = queue.pop()
         if len(queue) == 0:
@@ -395,7 +423,7 @@ class DownloaderAwarePriorityQueue:
         stats = self._downloader_interface.stats(self.pqueues)
         if not stats:
             return None
-        slot = min(stats)[1]
+        slot = self._next_slot(stats, update_state=False)
         queue = self.pqueues[slot]
         return queue.peek()
 

tests/test_pqueues.py

@@ -4,6 +4,7 @@ from unittest.mock import Mock
 import pytest
 import queuelib
 
+from scrapy.core.downloader import Downloader
 from scrapy.http.request import Request
 from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue
 from scrapy.spiders import Spider
@@ -158,6 +159,54 @@ class TestDownloaderAwarePriorityQueue:
         assert self.queue.pop().url == req3.url
         assert self.queue.peek() is None
 
+    def test_tie_breaking_rotates_slots(self):
+        # No active downloads are tracked in the downloader, so every slot has
+        # the same score and tie-breaking must not starve a slot.
+        req_a1 = Request("https://example.org/a1")
+        req_a1.meta[Downloader.DOWNLOAD_SLOT] = "slot-a"
+        req_b1 = Request("https://example.org/b1")
+        req_b1.meta[Downloader.DOWNLOAD_SLOT] = "slot-b"
+        req_a2 = Request("https://example.org/a2")
+        req_a2.meta[Downloader.DOWNLOAD_SLOT] = "slot-a"
+        req_b2 = Request("https://example.org/b2")
+        req_b2.meta[Downloader.DOWNLOAD_SLOT] = "slot-b"
+
+        for request in (req_a1, req_b1, req_a2, req_b2):
+            self.queue.push(request)
+
+        slots = [
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+        ]
+
+        assert slots == ["slot-a", "slot-b", "slot-a", "slot-b"]
+
+    def test_tie_breaking_keeps_rotation_after_selected_slot_is_deleted(self):
+        # If the selected slot becomes empty, rotation should continue from
+        # that slot marker to avoid restarting from the smallest slot.
+        req_a1 = Request("https://example.org/a1")
+        req_a1.meta[Downloader.DOWNLOAD_SLOT] = "slot-a"
+        req_a2 = Request("https://example.org/a2")
+        req_a2.meta[Downloader.DOWNLOAD_SLOT] = "slot-a"
+        req_b1 = Request("https://example.org/b1")
+        req_b1.meta[Downloader.DOWNLOAD_SLOT] = "slot-b"
+        req_c1 = Request("https://example.org/c1")
+        req_c1.meta[Downloader.DOWNLOAD_SLOT] = "slot-c"
+
+        for request in (req_a1, req_a2, req_b1, req_c1):
+            self.queue.push(request)
+
+        slots = [
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+            self.queue.pop().meta[Downloader.DOWNLOAD_SLOT],
+        ]
+
+        assert slots == ["slot-a", "slot-b", "slot-c", "slot-a"]
+
 
 @pytest.mark.parametrize(
     ("input_", "output"),