Add a throttling-aware scheduler

docs/topics/throttling.rst

@@ -459,8 +459,7 @@ path as a string):
 
     THROTTLING_MANAGER = "myproject.throttling.MyThrottlingManager"
 
-
-.. _multi-throttling-scopes:
+.. _multiple-throttling-scopes:
 
 Handling of multiple throttling scopes
 --------------------------------------
@@ -468,7 +467,6 @@ Handling of multiple throttling scopes
 When a request has multiple throttling scopes, it is not sent until all of its
 throttling scopes allow it.
 
-
 .. _throttling-quotas:
 
 Throttling quotas
@@ -505,7 +503,6 @@ Everything else being equal, :class:`~scrapy.pqueues.ScrapyPriorityQueue` will
 prioritize requests that consume a higher portion of the available throttling
 quota, to minimize the risk of those requests getting stuck.
 
-
 .. _custom-throttling-scope-managers:
 
 Customizing throttling scope managers
@@ -603,6 +600,28 @@ exponential backoff, similar to a fixed-window rate limiter:
         def is_idle(self, now, max_idle):
             return self.active == 0
 
+.. _throttling-aware-scheduler:
+
+Throttling-aware scheduling
+===========================
+
+By default, throttling is enforced at the engine, where a request waiting on
+its :ref:`throttling scopes <throttling-scopes>` holds a concurrency slot. In a
+crawl that mixes heavily-throttled scopes with unthrottled ones, this can let
+throttled requests starve unthrottled ones that could be sent right away
+(**head-of-line blocking**; Scrapy logs a warning, see
+:setting:`DELAYED_REQUESTS_WARN_THRESHOLD`).
+
+:class:`~scrapy.core.scheduler.ThrottlingAwareScheduler` avoids this. To enable
+it:
+
+.. code-block:: python
+    :caption: ``settings.py``
+
+    SCHEDULER = "scrapy.core.scheduler.ThrottlingAwareScheduler"
+    SCHEDULER_PRIORITY_QUEUE = "scrapy.pqueues.ThrottlingAwarePriorityQueue"
+
+.. autoclass:: scrapy.core.scheduler.ThrottlingAwareScheduler
 
 .. _throttling-examples:
 
@@ -822,8 +841,8 @@ window (:setting:`BACKOFF_WINDOW`). You can use :ref:`throttling quotas
 
 -   Implement a :ref:`throttling manager <custom-throttling-scopes>` that:
 
-    -   Sets a ``cost`` throttling scope on each request to some estimation based
-        e.g. on request URL parameters:
+    -   Sets a ``cost`` throttling scope on each request to some estimation
+        based e.g. on request URL parameters:
 
         .. code-block:: python
 
@@ -871,6 +890,35 @@ window (:setting:`BACKOFF_WINDOW`). You can use :ref:`throttling quotas
     This will allow you to spend up to 100.0 units of cost per time window
     (default: 60 seconds) before throttling kicks in.
 
+.. _throttling-per-ip:
+
+Per-IP concurrency limiting
+---------------------------
+
+A concurrency limit keyed by IP is just a throttling scope whose id is the
+request's IP, with a ``concurrency`` limit. A request then carries two scopes,
+its domain and its IP, and is only sent when **both** allow it (see
+:ref:`multiple-throttling-scopes`).
+
+-   Implement a :ref:`throttling manager <custom-throttling-scopes>` that adds
+    the request's IP as a second scope:
+
+    .. code-block:: python
+
+        import socket
+
+        from scrapy.throttling import ThrottlingManager, add_scope, scope_cache
+        from scrapy.utils.asyncio import run_in_thread
+        from scrapy.utils.httpobj import urlparse_cached
+
+
+        class IPThrottlingManager(ThrottlingManager):
+            @scope_cache
+            async def get_scopes(self, request):
+                scopes = await super().get_scopes(request)
+                host = urlparse_cached(request).hostname
+                address = await run_in_thread(socket.gethostbyname, host)
+                return add_scope(scopes, address)
 
 .. _throttling-settings:
 
@@ -1024,6 +1072,8 @@ API
 
 .. autoclass:: scrapy.throttling.ThrottlingScopeManager
 
+.. autoclass:: scrapy.pqueues.ThrottlingAwarePriorityQueue
+
 .. autoclass:: scrapy.throttling.ThrottlingScopeConfig
 
 .. autoclass:: scrapy.throttling.BackoffConfig

scrapy/core/engine.py

@@ -31,6 +31,7 @@ from scrapy.exceptions import (
 from scrapy.http import Request, Response
 from scrapy.utils.asyncio import (
     AsyncioLoopingCall,
+    call_later,
     create_looping_call,
     is_asyncio_available,
 )
@@ -57,6 +58,7 @@ if TYPE_CHECKING:
     from scrapy.settings import BaseSettings, Settings
     from scrapy.signalmanager import SignalManager
     from scrapy.spiders import Spider
+    from scrapy.utils.asyncio import CallLaterResult
 
 
 logger = logging.getLogger(__name__)
@@ -132,6 +134,14 @@ class ExecutionEngine:
         # Requests currently held by the throttling manager, waiting for their
         # scopes to allow them through to the downloader.
         self._throttling_waiting: set[Request] = set()
+        # Number of in-flight asynchronous enqueue operations (see
+        # ``_enqueue_request_async``), so the spider is not considered idle
+        # while a request is still on its way into the scheduler.
+        self._scheduling: int = 0
+        # A coalesced wakeup timer, armed when a throttling-aware scheduler
+        # reports that all pending requests are time-blocked (see
+        # ``next_request_delay``).
+        self._throttling_wakeup: CallLaterResult | None = None
         self._delayed_requests_warn_threshold: int = self.settings.getint(
             "DELAYED_REQUESTS_WARN_THRESHOLD"
         )
@@ -269,6 +279,7 @@ class ExecutionEngine:
 
     def pause(self) -> None:
         self.paused = True
+        self._cancel_throttling_wakeup()
 
     def unpause(self) -> None:
         self.paused = False
@@ -338,13 +349,43 @@ class ExecutionEngine:
         if self._slot is None or self._slot.closing is not None or self.paused:
             return
 
+        self._cancel_throttling_wakeup()
+
         while not self.needs_backout():
             if not self._start_scheduled_request():
                 break
 
+        self._maybe_arm_throttling_wakeup()
+
         if self.spider_is_idle() and self._slot.close_if_idle:
             self._spider_idle()
 
+    def _cancel_throttling_wakeup(self) -> None:
+        if self._throttling_wakeup is not None:
+            self._throttling_wakeup.cancel()
+            self._throttling_wakeup = None
+
+    def _maybe_arm_throttling_wakeup(self) -> None:
+        """Arm a single coalesced wakeup when the scheduler is throttling-aware
+        and reports that every pending request is time-blocked.
+
+        Concurrency-blocked states do not need this: a freed slot already
+        re-runs the loop from :meth:`_download`'s ``finally``. Only time-based
+        gates (delay, backoff, quota windows) need a timer, since nothing else
+        would re-run the loop while they are closed.
+        """
+        assert self._slot is not None
+        scheduler = self._slot.scheduler
+        delay_fn = getattr(scheduler, "next_request_delay", None)
+        if delay_fn is None or not scheduler.has_pending_requests():
+            return
+        delay = delay_fn()
+        if delay is None:
+            return
+        self._throttling_wakeup = call_later(
+            max(0.0, delay), self._slot.nextcall.schedule
+        )
+
     def needs_backout(self) -> bool:
         """Returns ``True`` if no more requests can be sent at the moment, or
         ``False`` otherwise.
@@ -382,10 +423,18 @@ class ExecutionEngine:
         if len(self._throttling_waiting) < self._delayed_requests_warn_threshold:
             return
         self._delayed_requests_warned = True
+        recommendation = ""
+        # A throttling-aware scheduler holds throttled requests in the
+        # scheduler instead of in _throttling_waiting, so it does not hit this
+        # path; recommend it only when it is not already in use.
+        scheduler = self._slot.scheduler if self._slot is not None else None
+        if scheduler is None or not hasattr(scheduler, "next_request_delay"):
+            recommendation = (
+                " Consider switching to scrapy.core.scheduler.ThrottlingAwareScheduler."
+            )
         logger.warning(
-            f"There are {len(self._throttling_waiting)} requests held back by throttling. This may "
-            "indicate a throttling configuration that is too aggressive or a "
-            "site that is rate-limiting heavily. See DELAYED_REQUESTS_WARN_THRESHOLD.",
+            f"There are {len(self._throttling_waiting)} requests held back by "
+            f"throttling. See DELAYED_REQUESTS_WARN_THRESHOLD.{recommendation}",
             extra={"spider": self.spider},
         )
 
@@ -464,6 +513,8 @@ class ExecutionEngine:
             return False
         if self._throttling_waiting:  # requests held by the throttling manager
             return False
+        if self._scheduling:  # requests still on their way into the scheduler
+            return False
         if self._start is not None:  # not all start requests are handled
             return False
         return not self._slot.scheduler.has_pending_requests()
@@ -476,6 +527,7 @@ class ExecutionEngine:
         self._slot.nextcall.schedule()  # type: ignore[union-attr]
 
     def _schedule_request(self, request: Request) -> None:
+        assert self._slot is not None  # typing
         request_scheduled_result = self.signals.send_catch_log(
             signals.request_scheduled,
             request=request,
@@ -485,11 +537,35 @@ class ExecutionEngine:
         for _, result in request_scheduled_result:
             if isinstance(result, Failure) and isinstance(result.value, IgnoreRequest):
                 return
-        if not self._slot.scheduler.enqueue_request(request):  # type: ignore[union-attr]
+        scheduler = self._slot.scheduler
+        if hasattr(scheduler, "enqueue_request_async"):
+            self._scheduling += 1
+            _schedule_coro(self._enqueue_request_async(request))
+            return
+        if not scheduler.enqueue_request(request):
             self.signals.send_catch_log(
                 signals.request_dropped, request=request, spider=self.spider
             )
 
+    async def _enqueue_request_async(self, request: Request) -> None:
+        assert self._slot is not None
+        try:
+            stored = await self._slot.scheduler.enqueue_request_async(request)  # type: ignore[attr-defined]
+            if not stored:
+                self.signals.send_catch_log(
+                    signals.request_dropped, request=request, spider=self.spider
+                )
+        except Exception:
+            logger.error(
+                "Error while enqueuing request",
+                exc_info=True,
+                extra={"spider": self.spider},
+            )
+        finally:
+            self._scheduling -= 1
+        if self._slot is not None:
+            self._slot.nextcall.schedule()
+
     def download(self, request: Request) -> Deferred[Response]:
         """Return a Deferred which fires with a Response as result, only downloader middlewares are applied"""
         warnings.warn(
@@ -675,6 +751,8 @@ class ExecutionEngine:
             "Closing spider (%(reason)s)", {"reason": reason}, extra={"spider": spider}
         )
 
+        self._cancel_throttling_wakeup()
+
         try:
             await self._slot.close()
         except Exception:

scrapy/core/scheduler.py

@@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, cast
 from twisted.internet.defer import Deferred  # noqa: TC002
 
 from scrapy.spiders import Spider  # noqa: TC001
+from scrapy.throttling import iter_scopes
 from scrapy.utils.job import job_dir
 from scrapy.utils.misc import build_from_crawler, load_object
 
@@ -25,6 +26,7 @@ if TYPE_CHECKING:
     from scrapy.http.request import Request
     from scrapy.pqueues import ScrapyPriorityQueue
     from scrapy.statscollectors import StatsCollector
+    from scrapy.throttling import ThrottlingManagerProtocol
 
 
 logger = logging.getLogger(__name__)
@@ -496,3 +498,109 @@ class Scheduler(BaseScheduler):
     def _write_dqs_state(self, dqdir: str, state: Any) -> None:
         with Path(dqdir, "active.json").open("w", encoding="utf-8") as f:
             json.dump(state, f)
+
+
+class ThrottlingAwareScheduler(Scheduler):
+    """A :class:`Scheduler` that only ever hands the engine requests whose
+    :ref:`throttling scopes <throttling-scopes>` allow them to be sent **right
+    now**.
+
+    This avoids the head-of-line blocking that the default scheduler can suffer
+    when a crawl mixes heavily-throttled scopes with unthrottled ones: with the
+    default scheduler, requests taken from the scheduler in order wait at the
+    throttling gate
+    (:meth:`~scrapy.throttling.ThrottlingManagerProtocol.acquire`) while
+    holding a concurrency slot, so enough throttled requests waiting on a clock
+    can fill :setting:`CONCURRENT_REQUESTS` and starve unthrottled requests
+    that could be sent right away. Requests held back by this scheduler instead
+    occupy neither concurrency slots nor memory (beyond what is needed to track
+    each distinct scope set), so they cannot starve other scopes.
+
+    When several requests could be sent at the same time, the one with the
+    highest request :attr:`~scrapy.Request.priority` is sent first; ties are
+    broken by preferring the least-busy scopes.
+
+    It requires :setting:`SCHEDULER_PRIORITY_QUEUE` to be set to
+    :class:`~scrapy.pqueues.ThrottlingAwarePriorityQueue` (or a compatible
+    subclass), and the configured :setting:`SCHEDULER_DISK_QUEUE` /
+    :setting:`SCHEDULER_MEMORY_QUEUE` to support ``peek``.
+    """
+
+    def open(self, spider: Spider) -> Deferred[None] | None:
+        result = super().open(spider)
+        if not hasattr(self.mqs, "next_request_delay"):
+            raise ValueError(
+                f"{type(self).__name__} requires SCHEDULER_PRIORITY_QUEUE to be "
+                f"set to a throttling-aware priority queue such as "
+                f"scrapy.pqueues.ThrottlingAwarePriorityQueue, but the "
+                f"configured one ({type(self.mqs).__name__}) is not."
+            )
+        assert self.crawler is not None
+        assert self.crawler.throttler is not None
+        self._throttler: ThrottlingManagerProtocol = self.crawler.throttler
+        return result
+
+    def enqueue_request(self, request: Request) -> bool:
+        raise RuntimeError(
+            "ThrottlingAwareScheduler requires the asynchronous enqueue path; "
+            "enqueue_request_async() is used by the engine instead of "
+            "enqueue_request()."
+        )
+
+    async def enqueue_request_async(self, request: Request) -> bool:
+        """Resolve the :ref:`throttling scope set <throttling-scopes>` of
+        *request* and store it under the queue for that scope set.
+
+        This is the asynchronous counterpart of
+        :meth:`Scheduler.enqueue_request`; scope resolution
+        (:meth:`~scrapy.throttling.ThrottlingManagerProtocol.get_scopes`) is
+        asynchronous, which is why enqueuing has to be too.
+        """
+        if not request.dont_filter and self.df.request_seen(request):
+            self.df.log(request, self.spider)
+            return False
+        scope_set = frozenset(iter_scopes(await self._throttler.get_scopes(request)))
+        assert self.stats is not None
+        if self._dqpush_throttling(request, scope_set):
+            self.stats.inc_value("scheduler/enqueued/disk")
+        else:
+            self.mqs.push(request, scope_set)  # type: ignore[call-arg]
+            self.stats.inc_value("scheduler/enqueued/memory")
+        self.stats.inc_value("scheduler/enqueued")
+        return True
+
+    def _dqpush_throttling(self, request: Request, scope_set: frozenset[str]) -> bool:
+        if self.dqs is None:
+            return False
+        try:
+            self.dqs.push(request, scope_set)  # type: ignore[call-arg]
+        except ValueError as e:  # non serializable request
+            if self.logunser:
+                logger.warning(
+                    "Unable to serialize request: %(request)s - reason:"
+                    " %(reason)s - no more unserializable requests will be"
+                    " logged (stats being collected)",
+                    {"request": request, "reason": e},
+                    exc_info=True,
+                    extra={"spider": self.spider},
+                )
+                self.logunser = False
+            assert self.stats is not None
+            self.stats.inc_value("scheduler/unserializable")
+            return False
+        return True
+
+    def next_request_delay(self) -> float | None:
+        """Return the minimum number of seconds until some pending request
+        becomes sendable because a time-based throttling gate opens, or
+        ``None`` if no pending request is time-blocked.
+
+        The engine uses this to arm a single wakeup timer when
+        :meth:`next_request` returns ``None`` while requests are still
+        pending."""
+        delays = [
+            delay
+            for pq in (self.mqs, self.dqs)
+            if pq is not None and (delay := pq.next_request_delay()) is not None  # type: ignore[attr-defined]
+        ]
+        return min(delays) if delays else None

scrapy/pqueues.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import logging
 from typing import TYPE_CHECKING, Protocol, cast
 
@@ -15,6 +16,7 @@ if TYPE_CHECKING:
     from scrapy import Request
     from scrapy.core.downloader import Downloader
     from scrapy.crawler import Crawler
+    from scrapy.throttling import ScopeID, ThrottlingManagerProtocol
 
 logger = logging.getLogger(__name__)
 
@@ -437,3 +439,179 @@ class DownloaderAwarePriorityQueue:
 
     def __contains__(self, slot: str) -> bool:
         return slot in self.pqueues
+
+
+def _scope_set_key(scope_set: frozenset[ScopeID]) -> str:
+    """Return a reversible, JSON-safe string key for *scope_set*.
+
+    Used both as the in-memory dict key encoding for the on-disk state and to
+    derive the per-scope-set subdirectory name. The encoding is
+    order-independent (the scope ids are sorted)."""
+    return json.dumps(sorted(scope_set))
+
+
+def _scope_set_from_key(key: str) -> frozenset[ScopeID]:
+    return frozenset(json.loads(key))
+
+
+class ThrottlingAwarePriorityQueue:
+    """Priority queue that partitions requests by their full :ref:`throttling
+    scope set <throttling-scopes>` and only ever pops a request that can be
+    sent right now.
+
+    The downstream queue class must support ``peek``.
+
+    Disk persistence
+    ================
+
+    .. warning:: The files that this class generates on disk are an
+        implementation detail, and may change without a warning in a future
+        version of Scrapy. Do not rely on the following information for
+        anything other than debugging purposes.
+
+    When instantiated with a non-empty *key* argument, *key* is used as a
+    persistence directory, and inside it this class creates a subdirectory per
+    scope set, named from a path-safe, order-independent encoding of its scope
+    ids.
+    """
+
+    @classmethod
+    def from_crawler(
+        cls,
+        crawler: Crawler,
+        downstream_queue_cls: type[QueueProtocol],
+        key: str,
+        startprios: dict[str, Iterable[int]] | None = None,
+        *,
+        start_queue_cls: type[QueueProtocol] | None = None,
+    ) -> Self:
+        return cls(
+            crawler,
+            downstream_queue_cls,
+            key,
+            startprios,
+            start_queue_cls=start_queue_cls,
+        )
+
+    def __init__(
+        self,
+        crawler: Crawler,
+        downstream_queue_cls: type[QueueProtocol],
+        key: str,
+        slot_startprios: dict[str, Iterable[int]] | None = None,
+        *,
+        start_queue_cls: type[QueueProtocol] | None = None,
+    ):
+        if slot_startprios and not isinstance(slot_startprios, dict):
+            raise ValueError(
+                "ThrottlingAwarePriorityQueue accepts ``slot_startprios`` as a "
+                f"dict; {slot_startprios.__class__!r} instance is passed. Most "
+                "likely, it means the state is created by an incompatible "
+                "priority queue. Only a crawl started with the same priority "
+                "queue class can be resumed."
+            )
+
+        assert crawler.throttler is not None
+        self._throttler: ThrottlingManagerProtocol = crawler.throttler
+        self.downstream_queue_cls: type[QueueProtocol] = downstream_queue_cls
+        self._start_queue_cls: type[QueueProtocol] | None = start_queue_cls
+        self.key: str = key
+        self.crawler: Crawler = crawler
+
+        # scope set -> priority queue
+        self.pqueues: dict[frozenset[ScopeID], ScrapyPriorityQueue] = {}
+        if slot_startprios:
+            for set_key, startprios in slot_startprios.items():
+                scope_set = _scope_set_from_key(set_key)
+                self.pqueues[scope_set] = self.pqfactory(scope_set, startprios)
+
+    def pqfactory(
+        self, scope_set: frozenset[ScopeID], startprios: Iterable[int] = ()
+    ) -> ScrapyPriorityQueue:
+        return ScrapyPriorityQueue(
+            self.crawler,
+            self.downstream_queue_cls,
+            self.key + "/" + _path_safe(_scope_set_key(scope_set)),
+            startprios,
+            start_queue_cls=self._start_queue_cls,
+        )
+
+    def push(self, request: Request, scope_set: frozenset[ScopeID]) -> None:
+        if scope_set not in self.pqueues:
+            self.pqueues[scope_set] = self.pqfactory(scope_set)
+        self.pqueues[scope_set].push(request)
+
+    def _select(
+        self,
+    ) -> tuple[frozenset[ScopeID], ScrapyPriorityQueue] | None:
+        """Return the sendable ``(scope_set, queue)`` pair to pop from, or
+        ``None`` if no queue can be popped from right now.
+
+        Among the sendable queues (those whose scope set can be sent right now),
+        the one whose head has the highest request priority is chosen; ties are
+        broken by ascending load (the maximum
+        :meth:`~scrapy.throttling.ThrottlingManagerProtocol.scope_load` over the
+        scopes of the queue), i.e. by preferring the least-busy scopes.
+        """
+        best_sort_key: tuple[int, float] | None = None
+        best: tuple[frozenset[ScopeID], ScrapyPriorityQueue] | None = None
+        for scope_set, queue in self.pqueues.items():
+            head = queue.peek()
+            if head is None or not self._throttler.is_ready(head):
+                continue
+            load = max(
+                (self._throttler.scope_load(scope_id) for scope_id in scope_set),
+                default=0.0,
+            )
+            sort_key = (queue.priority(head), load)
+            if best_sort_key is None or sort_key < best_sort_key:
+                best_sort_key = sort_key
+                best = (scope_set, queue)
+        return best
+
+    def pop(self) -> Request | None:
+        selected = self._select()
+        if selected is None:
+            return None
+        scope_set, queue = selected
+        request = queue.pop()
+        if request is not None:
+            self._throttler.reserve(request)
+        if len(queue) == 0:
+            del self.pqueues[scope_set]
+        return request
+
+    def peek(self) -> Request | None:
+        selected = self._select()
+        if selected is None:
+            return None
+        return selected[1].peek()
+
+    def next_request_delay(self) -> float | None:
+        delay: float | None = None
+        for queue in self.pqueues.values():
+            head = queue.peek()
+            if head is None:
+                continue
+            if self._throttler.is_ready(head):
+                return 0.0
+            head_delay = self._throttler.time_until_ready(head)
+            if head_delay is None:
+                continue
+            if delay is None or head_delay < delay:
+                delay = head_delay
+        return delay
+
+    def close(self) -> dict[str, list[int]]:
+        active = {
+            _scope_set_key(scope_set): queue.close()
+            for scope_set, queue in self.pqueues.items()
+        }
+        self.pqueues.clear()
+        return active
+
+    def __len__(self) -> int:
+        return sum(len(x) for x in self.pqueues.values()) if self.pqueues else 0
+
+    def __contains__(self, scope_set: frozenset[ScopeID]) -> bool:
+        return scope_set in self.pqueues

scrapy/throttling.py

@@ -317,6 +317,48 @@ class ThrottlingManagerProtocol(Protocol):
         enforce a concurrency limit can let other requests through.
         """
 
+    def is_ready(self, request: Request) -> bool:
+        """Return whether every scope of *request* allows it to be sent right
+        now, i.e. all time-based gates (delay, backoff, quota window) are open
+        *and* a concurrency slot is free in every scope.
+
+        This is the synchronous, non-blocking counterpart of :meth:`acquire`,
+        used by a :ref:`throttling-aware scheduler <throttling-aware-scheduler>`
+        to decide whether a request can be dequeued now. It assumes the scopes
+        of *request* have already been resolved (e.g. by an earlier
+        :meth:`get_scopes` call at enqueue time).
+        """
+
+    def reserve(self, request: Request) -> None:
+        """Claim a send for *request*: record the send on every one of its
+        scopes and mark *request* as reserved, so that a later :meth:`acquire`
+        for it returns immediately without reserving again.
+
+        A :ref:`throttling-aware scheduler <throttling-aware-scheduler>` calls
+        this when it decides to dequeue *request* (after :meth:`is_ready`
+        returned ``True``). The reservation is released by :meth:`release`.
+        """
+
+    def time_until_ready(self, request: Request) -> float | None:
+        """Return the number of seconds until every time-based gate of
+        *request* would be open, or ``None`` if no time-based gate is currently
+        blocking it (only a concurrency slot could be).
+
+        Used by a :ref:`throttling-aware scheduler
+        <throttling-aware-scheduler>` to schedule a wakeup when all pending
+        requests are time-blocked.
+        """
+
+    def scope_load(self, scope_id: str) -> float:
+        """Return the current load of the scope identified by *scope_id*: its
+        active sends divided by its concurrency limit (or by the global
+        :setting:`CONCURRENT_REQUESTS` when the scope has no explicit limit).
+
+        Used by a :ref:`throttling-aware scheduler
+        <throttling-aware-scheduler>` to balance dequeuing across scopes,
+        preferring the least-loaded ones.
+        """
+
     async def process_response(self, response: Response) -> None:
         """Update the throttling state based on *response*."""
 
@@ -352,10 +394,14 @@ def scope_cache(f: _GetScopesMethod) -> _GetScopesMethod:
             async def get_scopes(self, request):
                 return urlparse_cached(request).netloc
     """
-    cache: WeakKeyDictionary[Request, RequestScopes] = WeakKeyDictionary()
 
     @wraps(f)
     async def wrapper(self: Any, request: Request) -> RequestScopes:
+        cache: WeakKeyDictionary[Request, RequestScopes] | None = self.__dict__.get(
+            "_scope_cache"
+        )
+        if cache is None:
+            cache = self.__dict__["_scope_cache"] = WeakKeyDictionary()
         if request in cache:
             return cache[request]
         scopes = await f(self, request)
@@ -405,6 +451,7 @@ class ThrottlingManager:
             "THROTTLING_SCOPES"
         )
         self._scope_managers: dict[ScopeID, ThrottlingScopeManagerProtocol] = {}
+        self._global_concurrency: int = crawler.settings.getint("CONCURRENT_REQUESTS")
         self._last_eviction: float | None = None
         # Concurrency slots reserved by acquire(), to be released once the
         # request finishes downloading.
@@ -414,11 +461,37 @@ class ThrottlingManager:
 
     @scope_cache
     async def get_scopes(self, request: Request) -> RequestScopes:
+        return self._resolve_scopes_sync(request)
+
+    def _resolve_scopes_sync(self, request: Request) -> RequestScopes:
+        """Best-effort synchronous scope resolution.
+
+        It mirrors :meth:`get_scopes`, and is used by the synchronous
+        readiness methods (:meth:`is_ready`, :meth:`reserve`,
+        :meth:`time_until_ready`) when the cached result of an earlier
+        :meth:`get_scopes` call is not available (e.g. for a request restored
+        from disk). Subclasses whose :meth:`get_scopes` cannot be resolved
+        synchronously should rely on the enqueue-time cache instead.
+        """
         scopes = request.meta.get("throttling_scopes")
         if scopes is not None:
             return cast("RequestScopes", scopes)
         return urlparse_cached(request).netloc
 
+    def _cached_scope_values(
+        self, request: Request
+    ) -> list[tuple[ScopeID, float | None]]:
+        """Return the ``(scope_id, quota_amount)`` pairs of *request*, reading
+        the cache populated by :meth:`get_scopes` and falling back to
+        :meth:`_resolve_scopes_sync`."""
+        cache: WeakKeyDictionary[Request, RequestScopes] | None = self.__dict__.get(
+            "_scope_cache"
+        )
+        scopes = cache.get(request) if cache is not None else None
+        if scopes is None:
+            scopes = self._resolve_scopes_sync(request)
+        return list(iter_scope_values(scopes))
+
     async def get_initial_backoff(self) -> BackoffData:
         return None
 
@@ -472,6 +545,10 @@ class ThrottlingManager:
         return manager
 
     async def acquire(self, request: Request) -> None:
+        # A throttling-aware scheduler reserves the request before handing it
+        # to the engine, so there is nothing left to wait for or record here.
+        if request in self._reserved:
+            return
         now = time.monotonic()
         self._maybe_evict(now)
         await self._apply_request_delay(request)
@@ -518,6 +595,44 @@ class ThrottlingManager:
         for manager, _ in managers:
             manager.record_done()
 
+    # -- Synchronous readiness API (used by a throttling-aware scheduler) ------
+
+    def is_ready(self, request: Request) -> bool:
+        now = time.monotonic()
+        for scope_id, value in self._cached_scope_values(request):
+            manager = self._get_scope_manager(scope_id)
+            if manager.can_send(now=now, amount=value) > 0:
+                return False
+            if manager.concurrency_blocked():
+                return False
+        return True
+
+    def reserve(self, request: Request) -> None:
+        managers = [
+            (self._get_scope_manager(scope_id), value)
+            for scope_id, value in self._cached_scope_values(request)
+        ]
+        for manager, value in managers:
+            manager.record_sent(amount=value)
+        self._reserved[request] = managers
+
+    def time_until_ready(self, request: Request) -> float | None:
+        now = time.monotonic()
+        wait = 0.0
+        for scope_id, value in self._cached_scope_values(request):
+            manager = self._get_scope_manager(scope_id)
+            wait = max(wait, manager.can_send(now=now, amount=value))
+        return wait if wait > 0 else None
+
+    def scope_load(self, scope_id: ScopeID) -> float:
+        manager = self._get_scope_manager(scope_id)
+        active: int = getattr(manager, "_active", 0)
+        concurrency: int | None = getattr(manager, "_concurrency", None)
+        limit = concurrency if concurrency is not None else self._global_concurrency
+        if not limit:
+            return 0.0
+        return active / limit
+
     async def _wait_for_slot(self, managers: list[Any]) -> None:
         """Block until any of *managers* frees a concurrency slot.
 
@@ -749,13 +864,13 @@ class ThrottlingScopeManagerProtocol(Protocol):
         Return ``False`` when no concurrency limit is enforced.
         """
 
-    def slot_event(self) -> Deferred:
+    def slot_event(self) -> Deferred[None]:
         """Return a :class:`~twisted.internet.defer.Deferred` that fires when
         a concurrency slot next becomes available (e.g. when
         :meth:`record_done` is called or the limit is raised via
         :meth:`set_concurrency`)."""
 
-    def discard_slot_event(self, event: Deferred) -> None:
+    def discard_slot_event(self, event: Deferred[None]) -> None:
         """Cancel a pending slot event returned by :meth:`slot_event`.
 
         Called by :class:`ThrottlingManager` when the wait ends without the

tests/test_pqueues.py

@@ -6,12 +6,18 @@ import queuelib
 
 from scrapy.core.downloader import Downloader
 from scrapy.http.request import Request
-from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue
+from scrapy.pqueues import (
+    DownloaderAwarePriorityQueue,
+    ScrapyPriorityQueue,
+    ThrottlingAwarePriorityQueue,
+)
 from scrapy.spiders import Spider
 from scrapy.squeues import FifoMemoryQueue, PickleFifoDiskQueue
+from scrapy.throttling import iter_scopes
 from scrapy.utils.misc import build_from_crawler, load_object
 from scrapy.utils.test import get_crawler
 from tests.test_scheduler import MockDownloader
+from tests.utils.decorators import coroutine_test
 
 
 class TestPriorityQueue:
@@ -309,3 +315,107 @@ def test_pop_order(input_, output):
         actual_output_urls.append(request.url)
 
     assert actual_output_urls == expected_output_urls
+
+
+class TestThrottlingAwarePriorityQueue:
+    def _queue(self, crawler, key=""):
+        return build_from_crawler(
+            ThrottlingAwarePriorityQueue,
+            crawler,
+            downstream_queue_cls=FifoMemoryQueue,
+            key=key,
+        )
+
+    async def _push(self, queue, crawler, request):
+        scope_set = frozenset(iter_scopes(await crawler.throttler.get_scopes(request)))
+        queue.push(request, scope_set)
+
+    @coroutine_test
+    async def test_partitions_by_scope_set(self):
+        crawler = get_crawler(Spider)
+        queue = self._queue(crawler)
+        await self._push(queue, crawler, Request("http://a.com/1"))
+        await self._push(queue, crawler, Request("http://a.com/2"))
+        await self._push(queue, crawler, Request("http://b.com/1"))
+        # Two distinct scope sets -> two internal queues, three requests.
+        assert len(queue.pqueues) == 2
+        assert len(queue) == 3
+
+    @coroutine_test
+    async def test_pop_skips_blocked_scope(self):
+        crawler = get_crawler(
+            Spider,
+            settings_dict={
+                "THROTTLING_SCOPES": {"slow.com": {"delay": 1000.0}},
+                "RANDOMIZE_DOWNLOAD_DELAY": False,
+            },
+        )
+        queue = self._queue(crawler)
+        await self._push(queue, crawler, Request("http://slow.com/1"))
+        await self._push(queue, crawler, Request("http://slow.com/2"))
+        await self._push(queue, crawler, Request("http://fast.com/1"))
+        # The first slow request is sendable (no delay accrued yet); after it is
+        # popped (and reserved), the second slow request is blocked, but the
+        # fast one is still served.
+        popped = [queue.pop(), queue.pop(), queue.pop()]
+        urls = [r.url if r else None for r in popped]
+        assert "http://fast.com/1" in urls
+        assert "http://slow.com/1" in urls
+        # The blocked second slow request stays in the queue.
+        assert None in urls
+        assert len(queue) == 1
+        delay = queue.next_request_delay()
+        assert delay is not None
+        assert delay == pytest.approx(1000.0, abs=1.0)
+
+    @coroutine_test
+    async def test_least_loaded_first(self):
+        crawler = get_crawler(
+            Spider,
+            settings_dict={
+                "THROTTLING_SCOPES": {
+                    "a.com": {"concurrency": 4},
+                    "b.com": {"concurrency": 4},
+                }
+            },
+        )
+        queue = self._queue(crawler)
+        # Make a.com busier than b.com.
+        busy = Request("http://a.com/0")
+        await crawler.throttler.get_scopes(busy)
+        crawler.throttler.reserve(busy)
+        await self._push(queue, crawler, Request("http://a.com/1"))
+        await self._push(queue, crawler, Request("http://b.com/1"))
+        # Equal priority, so the lower-load scope (b.com) is served first.
+        assert queue.pop().url == "http://b.com/1"
+
+    @coroutine_test
+    async def test_priority_beats_load(self):
+        crawler = get_crawler(
+            Spider,
+            settings_dict={
+                "THROTTLING_SCOPES": {
+                    "a.com": {"concurrency": 4},
+                    "b.com": {"concurrency": 4},
+                }
+            },
+        )
+        queue = self._queue(crawler)
+        # Make a.com busier than b.com.
+        busy = Request("http://a.com/0")
+        await crawler.throttler.get_scopes(busy)
+        crawler.throttler.reserve(busy)
+        # The a.com request has higher priority, and a.com still has room, so it
+        # is served first despite a.com being the busier scope.
+        await self._push(queue, crawler, Request("http://a.com/1", priority=10))
+        await self._push(queue, crawler, Request("http://b.com/1"))
+        assert queue.pop().url == "http://a.com/1"
+
+    @coroutine_test
+    async def test_empty_and_close(self):
+        crawler = get_crawler(Spider)
+        queue = self._queue(crawler)
+        assert queue.pop() is None
+        assert queue.next_request_delay() is None
+        await self._push(queue, crawler, Request("http://a.com/1"))
+        assert queue.close() != {}

tests/test_scheduler.py

@@ -10,7 +10,7 @@ from unittest.mock import Mock
 import pytest
 
 from scrapy.core.downloader import Downloader
-from scrapy.core.scheduler import BaseScheduler, Scheduler
+from scrapy.core.scheduler import BaseScheduler, Scheduler, ThrottlingAwareScheduler
 from scrapy.crawler import Crawler
 from scrapy.exceptions import ScrapyDeprecationWarning
 from scrapy.http import Request
@@ -406,3 +406,175 @@ class TestIncompatibility:
                 ValueError, match="does not support CONCURRENT_REQUESTS_PER_IP"
             ):
                 self._incompatible()
+
+
+_THROTTLING_AWARE_PQ = "scrapy.pqueues.ThrottlingAwarePriorityQueue"
+
+
+class TestThrottlingAwareScheduler:
+    def _crawler(self, settings_dict: dict[str, Any] | None = None) -> Crawler:
+        settings = {
+            "SCHEDULER_PRIORITY_QUEUE": _THROTTLING_AWARE_PQ,
+            "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter",
+            **(settings_dict or {}),
+        }
+        return get_crawler(Spider, settings)
+
+    def _scheduler(self, crawler: Crawler) -> ThrottlingAwareScheduler:
+        spider = Spider(name="spider")
+        crawler.spider = spider
+        scheduler = ThrottlingAwareScheduler.from_crawler(crawler)
+        scheduler.open(spider)
+        return scheduler
+
+    @coroutine_test
+    async def test_enqueue_async_and_dequeue(self) -> None:
+        scheduler = self._scheduler(self._crawler())
+        assert await scheduler.enqueue_request_async(Request("http://a.com/1")) is True
+        assert scheduler.has_pending_requests()
+        assert len(scheduler) == 1
+        request = scheduler.next_request()
+        assert request is not None
+        assert request.url == "http://a.com/1"
+        assert scheduler.next_request() is None
+        assert not scheduler.has_pending_requests()
+        scheduler.close("finished")
+
+    def test_sync_enqueue_raises(self) -> None:
+        scheduler = self._scheduler(self._crawler())
+        with pytest.raises(RuntimeError, match="asynchronous enqueue path"):
+            scheduler.enqueue_request(Request("http://a.com/1"))
+        scheduler.close("finished")
+
+    def test_requires_throttling_aware_priority_queue(self) -> None:
+        crawler = self._crawler(
+            {"SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ScrapyPriorityQueue"}
+        )
+        spider = Spider(name="spider")
+        crawler.spider = spider
+        scheduler = ThrottlingAwareScheduler.from_crawler(crawler)
+        with pytest.raises(ValueError, match="throttling-aware priority queue"):
+            scheduler.open(spider)
+
+    @coroutine_test
+    async def test_delay_blocks_and_reports_delay(self) -> None:
+        crawler = self._crawler(
+            {
+                "THROTTLING_SCOPES": {"slow.com": {"delay": 1000.0}},
+                "RANDOMIZE_DOWNLOAD_DELAY": False,
+            },
+        )
+        scheduler = self._scheduler(crawler)
+        await scheduler.enqueue_request_async(Request("http://slow.com/1"))
+        await scheduler.enqueue_request_async(Request("http://slow.com/2"))
+        # The first request is sendable; the second is blocked by the delay.
+        first = scheduler.next_request()
+        assert first is not None
+        assert scheduler.next_request() is None
+        assert scheduler.has_pending_requests()
+        assert scheduler.next_request_delay() == pytest.approx(1000.0, abs=1.0)
+        scheduler.close("finished")
+
+    @coroutine_test
+    async def test_no_delay_when_only_concurrency_blocked(self) -> None:
+        crawler = self._crawler(
+            {"THROTTLING_SCOPES": {"slow.com": {"concurrency": 1}}},
+        )
+        scheduler = self._scheduler(crawler)
+        await scheduler.enqueue_request_async(Request("http://slow.com/1"))
+        await scheduler.enqueue_request_async(Request("http://slow.com/2"))
+        assert scheduler.next_request() is not None
+        assert scheduler.next_request() is None
+        # A purely concurrency-blocked state has no time-based wakeup.
+        assert scheduler.next_request_delay() is None
+        scheduler.close("finished")
+
+    @coroutine_test
+    async def test_resume_from_disk(self, tmp_path: Path) -> None:
+        settings = {"JOBDIR": str(tmp_path)}
+        scheduler = self._scheduler(self._crawler(settings))
+        await scheduler.enqueue_request_async(Request("http://a.com/1"))
+        await scheduler.enqueue_request_async(Request("http://b.com/1"))
+        scheduler.close("shutdown")
+
+        scheduler2 = self._scheduler(self._crawler(settings))
+        assert len(scheduler2) == 2
+        urls = set()
+        while (request := scheduler2.next_request()) is not None:
+            urls.add(request.url)
+        assert urls == {"http://a.com/1", "http://b.com/1"}
+        scheduler2.close("finished")
+
+
+class TestIntegrationWithThrottlingAwareScheduler:
+    @inline_callbacks_test
+    def test_integration(self):
+        crawler = get_crawler(
+            spidercls=StartUrlsSpider,
+            settings_dict={
+                "SCHEDULER": "scrapy.core.scheduler.ThrottlingAwareScheduler",
+                "SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ThrottlingAwarePriorityQueue",
+                "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter",
+            },
+        )
+        with MockServer() as mockserver:
+            url = mockserver.url("/status?n=200", is_secure=False)
+            start_urls = [url] * 6
+            yield crawler.crawl(start_urls)
+            assert crawler.stats.get_value("downloader/response_count") == len(
+                start_urls
+            )
+
+    @inline_callbacks_test
+    def test_integration_follow_requests(self):
+        # Exercises the engine's asynchronous enqueue path for requests yielded
+        # from a callback (not just start requests), and the idle guard that
+        # keeps the spider open while an enqueue is in flight.
+        class FollowSpider(Spider):
+            name = "follow"
+
+            def __init__(self, base_url, **kwargs):
+                self.base_url = base_url
+                super().__init__(**kwargs)
+
+            async def start(self):
+                yield Request(self.base_url + "/status?n=200")
+
+            def parse(self, response):
+                if b"follow" not in response.url.encode():
+                    yield Request(response.url + "&follow=1", callback=self.parse)
+
+        with MockServer() as mockserver:
+            base_url = mockserver.url("", is_secure=False).rstrip("/")
+            crawler = get_crawler(
+                spidercls=FollowSpider,
+                settings_dict={
+                    "SCHEDULER": "scrapy.core.scheduler.ThrottlingAwareScheduler",
+                    "SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ThrottlingAwarePriorityQueue",
+                    "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter",
+                },
+            )
+            yield crawler.crawl(base_url=base_url)
+            assert crawler.stats.get_value("downloader/response_count") == 2
+
+    @inline_callbacks_test
+    def test_integration_with_delay(self):
+        # A small per-scope delay forces the engine to arm the throttling wakeup
+        # timer between requests; the crawl must still complete.
+        crawler = get_crawler(
+            spidercls=StartUrlsSpider,
+            settings_dict={
+                "SCHEDULER": "scrapy.core.scheduler.ThrottlingAwareScheduler",
+                "SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ThrottlingAwarePriorityQueue",
+                "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter",
+                "RANDOMIZE_DOWNLOAD_DELAY": False,
+                "THROTTLING_SCOPES": {"127.0.0.1": {"delay": 0.05}},
+            },
+        )
+        with MockServer() as mockserver:
+            url = mockserver.url("/status?n=200", is_secure=False)
+            start_urls = [url] * 4
+            yield crawler.crawl(start_urls)
+            assert crawler.stats.get_value("downloader/response_count") == len(
+                start_urls
+            )

tests/test_throttling.py

@@ -540,6 +540,91 @@ class TestThrottlingScopeManager:
         assert scope._concurrency == 1
 
 
+class TestThrottlingManagerReadiness:
+    """The synchronous readiness API used by a throttling-aware scheduler."""
+
+    @coroutine_test
+    async def test_is_ready_unconstrained_scope(self):
+        manager = _manager()
+        request = Request("http://example.com/a")
+        # A scope with no configured delay/concurrency/quota is always ready.
+        assert await manager.get_scopes(request) == "example.com"
+        assert manager.is_ready(request) is True
+
+    @coroutine_test
+    async def test_is_ready_without_cached_scopes(self):
+        # is_ready falls back to synchronous resolution when get_scopes was not
+        # called first (e.g. for a request restored from disk).
+        manager = _manager()
+        request = Request("http://example.com/a")
+        assert manager.is_ready(request) is True
+
+    @coroutine_test
+    async def test_reserve_blocks_delay_scope(self):
+        manager = _manager(
+            {
+                "THROTTLING_SCOPES": {"example.com": {"delay": 100.0}},
+                "RANDOMIZE_DOWNLOAD_DELAY": False,
+            }
+        )
+        first = Request("http://example.com/1")
+        second = Request("http://example.com/2")
+        await manager.get_scopes(first)
+        await manager.get_scopes(second)
+        assert manager.is_ready(first) is True
+        manager.reserve(first)
+        # The base delay now blocks any further request for the scope.
+        assert manager.is_ready(second) is False
+        assert manager.time_until_ready(second) == pytest.approx(100.0, abs=1.0)
+
+    @coroutine_test
+    async def test_reserve_blocks_on_concurrency(self):
+        manager = _manager({"THROTTLING_SCOPES": {"example.com": {"concurrency": 1}}})
+        first = Request("http://example.com/1")
+        second = Request("http://example.com/2")
+        await manager.get_scopes(first)
+        await manager.get_scopes(second)
+        manager.reserve(first)
+        assert manager.is_ready(second) is False
+        # Pure concurrency blocking is not time-gated.
+        assert manager.time_until_ready(second) is None
+        manager.release(first)
+        assert manager.is_ready(second) is True
+
+    @coroutine_test
+    async def test_acquire_noop_when_reserved(self):
+        manager = _manager(
+            {
+                "THROTTLING_SCOPES": {"example.com": {"delay": 100.0}},
+                "RANDOMIZE_DOWNLOAD_DELAY": False,
+            }
+        )
+        request = Request("http://example.com/1")
+        await manager.get_scopes(request)
+        manager.reserve(request)
+        # A reserved request fast-paths through acquire() without re-recording
+        # the send or waiting for the delay.
+        await manager.acquire(request)
+        scope = manager._get_scope_manager("example.com")
+        assert scope._active == 1  # reserve recorded exactly one send
+
+    @coroutine_test
+    async def test_scope_load(self):
+        manager = _manager({"THROTTLING_SCOPES": {"example.com": {"concurrency": 4}}})
+        assert manager.scope_load("example.com") == 0.0
+        request = Request("http://example.com/1")
+        await manager.get_scopes(request)
+        manager.reserve(request)
+        assert manager.scope_load("example.com") == pytest.approx(0.25)
+
+    def test_scope_load_falls_back_to_global_concurrency(self):
+        manager = _manager({"CONCURRENT_REQUESTS": 8})
+        # A scope with no explicit concurrency limit uses CONCURRENT_REQUESTS.
+        request = Request("http://example.com/1")
+        manager.reserve(request)
+        assert manager.scope_load("example.com") == pytest.approx(1 / 8)
+
+
 class TestThrottlingIntegration:
     @coroutine_test
     async def test_backoff_recorded_on_429(self, mockserver):