WIP

docs/topics/components.rst

@@ -26,6 +26,7 @@ That includes the classes that you may assign to the following settings:
 - :setting:`SCHEDULER_START_MEMORY_QUEUE`
 - :setting:`SPIDER_MIDDLEWARES`
 - :setting:`THROTTLING_MANAGER`
+- :setting:`THROTTLING_SCOPE_MANAGER`
 
 Third-party Scrapy components may also let you define additional Scrapy
 components, usually configurable through :ref:`settings <topics-settings>`, to

docs/topics/settings.rst

@@ -861,6 +861,7 @@ Default:
         "scrapy.downloadermiddlewares.redirect.MetaRefreshMiddleware": 580,
         "scrapy.downloadermiddlewares.httpcompression.HttpCompressionMiddleware": 590,
         "scrapy.downloadermiddlewares.redirect.RedirectMiddleware": 600,
+        "scrapy.downloadermiddlewares.backoff.BackoffMiddleware": 630,
         "scrapy.downloadermiddlewares.cookies.CookiesMiddleware": 700,
         "scrapy.downloadermiddlewares.httpproxy.HttpProxyMiddleware": 750,
         "scrapy.downloadermiddlewares.stats.DownloaderStats": 850,
@@ -1255,6 +1256,7 @@ Default:
         "scrapy.extensions.feedexport.FeedExporter": 0,
         "scrapy.extensions.logstats.LogStats": 0,
         "scrapy.extensions.spiderstate.SpiderState": 0,
+        "scrapy.extensions.throttle.AutoThrottle": 0,
     }
 
 A dict containing the extensions available by default in Scrapy, and their

docs/topics/throttling.rst

@@ -140,12 +140,7 @@ A **backoff trigger** is a response whose status code is in
 :setting:`BACKOFF_HTTP_CODES` or a download exception whose type is in
 :setting:`BACKOFF_EXCEPTIONS`. On each trigger:
 
-#.  If the response carries a :ref:`Retry-After or RateLimit-Reset
-    <rate-limiting-headers>` value, that value (capped at
-    :setting:`BACKOFF_MAX_DELAY`) is honored as a hard minimum: no request is
-    sent for the scope until it elapses.
-
-#.  Otherwise the delay grows exponentially:
+#.  The delay grows exponentially:
 
     .. code-block:: text
 
@@ -154,6 +149,13 @@ A **backoff trigger** is a response whose status code is in
     :setting:`BACKOFF_JITTER` is then applied so that requests that backed off
     together do not retry in lockstep.
 
+#.  If the response carries a :ref:`Retry-After or RateLimit-Reset
+    <rate-limiting-headers>` value, the scope is *also* held back until that
+    time (capped at :setting:`BACKOFF_MAX_DELAY`) before its next request. This
+    is a one-time gate, on top of the exponential step above: it honors the
+    header for the next request without turning a short header value into a
+    long-standing delay for every later request.
+
 **Recovery** is linear: after a scope goes a full :setting:`BACKOFF_WINDOW`
 without a new trigger, its delay drops by one :setting:`BACKOFF_DELAY_FACTOR`
 step toward the configured value, and keeps dropping one step per quiet window
@@ -227,17 +229,16 @@ setting:
 
     Target number of backoff responses per :setting:`BACKOFF_WINDOW` that
     :ref:`rampup <rampup>` aims for when probing the rate limit of a scope.
-    Can be a range like ``[1, 3]``.
 
 For every :setting:`BACKOFF_WINDOW` that stays **below**
 :setting:`RAMPUP_BACKOFF_TARGET` backoff triggers, rampup increases throughput
 one step: it first lowers the delay, and once the delay reaches its minimum it
 raises the concurrency limit above the ``"min_concurrency"`` floor of the
-scope. Windows that hit the target hold the current rate, and windows that
-exceed it let normal :ref:`backoff <backoff>` reduce the rate. The result is a
-rate that converges on roughly :setting:`RAMPUP_BACKOFF_TARGET` rate-limit
-responses per window — the most throughput a scope allows without being
-penalized.
+scope. Windows that reach or exceed the target do not ramp up; the rate is
+reduced by normal :ref:`backoff <backoff>` (which grows the delay) instead.
+Rampup only ever probes upward, so the rate settles around the most throughput
+a scope allows while triggering fewer than :setting:`RAMPUP_BACKOFF_TARGET`
+rate-limit responses per window.
 
 Rampup behavior can be fine-tuned per scope by giving ``"rampup"`` a dict
 instead of ``True``:
@@ -248,7 +249,7 @@ instead of ``True``:
     THROTTLING_SCOPES = {
         "api.toscrape.com": {
             "rampup": {
-                "backoff_target": [1, 3],  # overrides RAMPUP_BACKOFF_TARGET
+                "backoff_target": 1,  # overrides RAMPUP_BACKOFF_TARGET
                 "delay_factor": 0.5,  # multiply the delay by this on each ramp-up step
                 "min_delay": 0.05,  # do not ramp the delay below this
             },
@@ -267,8 +268,9 @@ Servers may include `Retry-After
 or `RateLimit-Reset
 <https://www.ietf.org/archive/id/draft-polli-ratelimit-headers-02.html#name-ratelimit-reset>`__
 headers to indicate when you should make your next request. These headers are
-respected automatically during :ref:`backoff <backoff>`, using their values as
-minimum delays (capped at :setting:`BACKOFF_MAX_DELAY`).
+respected automatically during :ref:`backoff <backoff>`: the scope's next
+request is held back until the indicated time (capped at
+:setting:`BACKOFF_MAX_DELAY`), on top of the usual exponential backoff step.
 
 .. seealso:: :setting:`REDIRECT_MAX_DELAY`
 
@@ -304,12 +306,17 @@ You can delay a :ref:`throttling scope <throttling-scopes>` on demand through
 :meth:`crawler.throttler.delay_scope()
 <scrapy.throttling.ThrottlingManagerProtocol.delay_scope>`:
 
+.. skip: next
+
 .. code-block:: python
 
     crawler.throttler.delay_scope("example.com", 30.0)
 
-This holds back every request of the scope for at least the given number of
-seconds, counted as a :ref:`backoff <backoff>` trigger.
+This holds back the scope's next request for at least the given number of
+seconds and registers a :ref:`backoff <backoff>` trigger. Like a
+``Retry-After`` header, it is a one-time delay rather than a permanent one (the
+scope's delay also grows by one backoff step and then recovers); call it again,
+e.g. on each matching response, to keep a scope slowed down for longer.
 
 It is useful to react to situations that :ref:`automatic backoff <backoff>`
 cannot detect on its own, such as a soft block that comes back as a ``200``

scrapy/core/engine.py

@@ -283,6 +283,12 @@ class ExecutionEngine:
 
     def unpause(self) -> None:
         self.paused = False
+        # pause() cancels the coalesced throttling wakeup and the loop stops
+        # re-running itself while paused, so re-run it here: this re-arms the
+        # wakeup and keeps a crawl held only by a time-based throttling gate
+        # (nothing in flight to re-run the loop from _download) from stalling.
+        if self._slot is not None:
+            self._slot.nextcall.schedule()
 
     async def _process_start_next(self) -> None:
         """Processes the next item or request from Spider.start().
@@ -553,8 +559,13 @@ class ExecutionEngine:
             )
 
     async def _enqueue_request_async(self, request: Request) -> None:
-        assert self._slot is not None
+        # The counter is incremented in _schedule_request before this coroutine
+        # is scheduled, so it must be decremented here even on an early exit,
+        # otherwise spider_is_idle() never reports idle. The slot can be torn
+        # down (spider stopping) between the increment and this running.
         try:
+            if self._slot is None:
+                return
             stored = await self._slot.scheduler.enqueue_request_async(request)  # type: ignore[attr-defined]
             if not stored:
                 self.signals.send_catch_log(

scrapy/downloadermiddlewares/redirect.py

@@ -7,6 +7,7 @@ from urllib.parse import urljoin, urlparse
 from w3lib.url import safe_url_string
 
 from scrapy import signals
+from scrapy.downloadermiddlewares.backoff import _parse_retry_after
 from scrapy.exceptions import IgnoreRequest, NotConfigured
 from scrapy.http import HtmlResponse, Response
 from scrapy.spidermiddlewares.referer import RefererMiddleware
@@ -36,6 +37,7 @@ class BaseRedirectMiddleware:
             raise NotConfigured
 
         self.max_redirect_times: int = settings.getint("REDIRECT_MAX_TIMES")
+        self.max_delay: float = settings.getfloat("REDIRECT_MAX_DELAY")
         self.priority_adjust: int = settings.getint("REDIRECT_PRIORITY_ADJUST")
         self._referer_spider_middleware: RefererMiddleware | None = None
 
@@ -174,9 +176,30 @@ class BaseRedirectMiddleware:
                 del redirect_request.headers["Authorization"]
 
         self.handle_referer(redirect_request, response)
+        self._apply_retry_after(redirect_request, response)
 
         return redirect_request
 
+    def _apply_retry_after(self, redirect_request: Request, response: Response) -> None:
+        """Delay the redirect when *response* carries a ``Retry-After`` header.
+
+        The delay is capped at :setting:`REDIRECT_MAX_DELAY` and applied through
+        the :reqmeta:`throttling_delay` request metadata key, which holds back
+        this request only (without counting as a :ref:`backoff <backoff>`
+        trigger for its scopes). :setting:`REDIRECT_MAX_DELAY` set to ``0``
+        disables it.
+        """
+        if not self.max_delay:
+            return
+        retry_after = _parse_retry_after(response)
+        if retry_after is None:
+            return
+        redirect_request.meta["throttling_delay"] = min(retry_after, self.max_delay)
+        # This is a fresh request that may inherit an already-honored delay from
+        # its source request's meta; clear that state so the new delay applies.
+        redirect_request.meta.pop("_throttling_delayed", None)
+        redirect_request.meta.pop("_throttling_delay_deadline", None)
+
     def _redirect_request_using_get(
         self, request: Request, response: Response, redirect_url: str
     ) -> Request:

scrapy/pqueues.py

@@ -238,10 +238,14 @@ class ScrapyPriorityQueue:
         """
         if self.curprio is None:
             return None
+        # Mirror pop(): at a given priority it drains the regular queue before
+        # the start queue, so peek() must report the regular head first for the
+        # two to agree on "the next request" (a throttling-aware queue relies on
+        # this to check readiness of the exact request pop() will return).
         try:
-            queue = self._start_queues[self.curprio]
-        except KeyError:
             queue = self.queues[self.curprio]
+        except KeyError:
+            queue = self._start_queues[self.curprio]
         # Protocols can't declare optional members
         return cast("Request", queue.peek())  # type: ignore[attr-defined]
 

scrapy/settings/default_settings.py

@@ -179,6 +179,7 @@ __all__ = [
     "RANDOMIZE_DOWNLOAD_DELAY",
     "REACTOR_THREADPOOL_MAXSIZE",
     "REDIRECT_ENABLED",
+    "REDIRECT_MAX_DELAY",
     "REDIRECT_MAX_TIMES",
     "REDIRECT_PRIORITY_ADJUST",
     "REFERER_ENABLED",
@@ -531,6 +532,7 @@ RANDOMIZE_DOWNLOAD_DELAY = True
 REACTOR_THREADPOOL_MAXSIZE = 10
 
 REDIRECT_ENABLED = True
+REDIRECT_MAX_DELAY = 120.0
 REDIRECT_MAX_TIMES = 20  # uses Firefox default setting
 REDIRECT_PRIORITY_ADJUST = +2
 

scrapy/throttling.py

@@ -57,7 +57,7 @@ class RampupConfig(TypedDict, total=False):
     :setting:`RAMPUP_BACKOFF_TARGET`).
     """
 
-    backoff_target: float | list[float]
+    backoff_target: float
     delay_factor: float
     min_delay: float
 
@@ -365,24 +365,28 @@ class ThrottlingManagerProtocol(Protocol):
         *scopes* accepts the same shapes as the output of :meth:`get_scopes`
         (typically the result of :meth:`get_resolved_scopes` for a request).
 
-        When *delay* is ``None`` an exponential backoff step is applied; when
-        given, *delay* is a hard minimum delay in seconds (e.g. from a
-        :ref:`Retry-After <retry-after>` header). *cap* limits *delay* to
-        :setting:`BACKOFF_MAX_DELAY`; set it to ``False`` for trusted,
-        programmatic delays.
+        An exponential backoff step is always applied to the scope's delay.
+        When *delay* is given, the scope is *additionally* held back for at
+        least *delay* seconds before its next request: a one-time gate (e.g.
+        from a :ref:`Retry-After <retry-after>` header), not a change to the
+        steady-state delay. *cap* limits *delay* to :setting:`BACKOFF_MAX_DELAY`;
+        set it to ``False`` for trusted, programmatic delays.
         """
 
     def delay_scope(self, scope_id: str, delay: float) -> None:
-        """Hold back every request of the scope identified by *scope_id* for at
-        least *delay* seconds, counted as a :ref:`backoff <backoff>` trigger
-        for the scope.
+        """Hold back the scope identified by *scope_id* for at least *delay*
+        seconds before its next request, and register a :ref:`backoff
+        <backoff>` trigger for the scope.
+
+        Like a :ref:`Retry-After <retry-after>` response header, this is a
+        one-time delay (the scope's steady-state delay grows by one backoff
+        step and then recovers), not a permanent one; call it again to keep a
+        scope slowed down for longer.
 
         This is shorthand for :meth:`back_off(scope_id, delay=delay,
-        cap=False) <back_off>`: the programmatic equivalent of a
-        :ref:`Retry-After <retry-after>` response header. Unlike those headers,
-        *delay* is **not** capped at :setting:`BACKOFF_MAX_DELAY`: that cap
-        guards against untrusted input, whereas a ``delay_scope`` call is
-        trusted.
+        cap=False) <back_off>`. Unlike a ``Retry-After`` header, *delay* is
+        **not** capped at :setting:`BACKOFF_MAX_DELAY`: that cap guards against
+        untrusted input, whereas a ``delay_scope`` call is trusted.
         """
 
     def reconcile_quota(
@@ -866,8 +870,9 @@ class ThrottlingManager:
         manager.set_concurrency(1)
 
     def delay_scope(self, scope_id: ScopeID, delay: float) -> None:
-        # Like a Retry-After / RateLimit-Reset header, this is a hard minimum
-        # delay; unlike those, it is trusted, so it bypasses BACKOFF_MAX_DELAY.
+        # Like a Retry-After / RateLimit-Reset header, this gates the scope's
+        # next request by delay seconds (a one-time hold, not a steady-state
+        # delay); unlike those, it is trusted, so it bypasses BACKOFF_MAX_DELAY.
         self.back_off(scope_id, delay=float(delay), cap=False)
 
     def get_scope_delay(self, scope_id: ScopeID) -> float:
@@ -1076,8 +1081,10 @@ class ThrottlingScopeManager:
         :setting:`BACKOFF_EXCEPTIONS` exception) the delay grows exponentially
         by :setting:`BACKOFF_DELAY_FACTOR`, bounded by :setting:`BACKOFF_MIN_DELAY`
         and :setting:`BACKOFF_MAX_DELAY`, with :setting:`BACKOFF_JITTER` applied.
-        A ``Retry-After`` / ``RateLimit-Reset`` delay is honored as a hard
-        minimum (capped at :setting:`BACKOFF_MAX_DELAY`).
+        A ``Retry-After`` / ``RateLimit-Reset`` delay additionally holds the
+        scope back until that time before its next request (a one-time gate,
+        capped at :setting:`BACKOFF_MAX_DELAY`), without becoming the
+        steady-state delay.
 
     -   After :setting:`BACKOFF_WINDOW` seconds without a new trigger, the delay
         recovers one step at a time back towards the base delay.
@@ -1146,8 +1153,10 @@ class ThrottlingScopeManager:
         rampup = config.get("rampup")
         self._rampup_enabled: bool = bool(rampup)
         rampup_config: dict[str, Any] = rampup if isinstance(rampup, dict) else {}
-        self._rampup_target: tuple[float, float] = self._parse_target(
-            rampup_config.get("backoff_target", settings.get("RAMPUP_BACKOFF_TARGET"))
+        self._rampup_target: float = float(
+            rampup_config.get(
+                "backoff_target", settings.getfloat("RAMPUP_BACKOFF_TARGET")
+            )
         )
         self._rampup_delay_factor: float = float(rampup_config.get("delay_factor", 0.5))
         self._rampup_min_delay: float = float(rampup_config.get("min_delay", 0.0))
@@ -1191,12 +1200,6 @@ class ThrottlingScopeManager:
     def _now(now: float | None) -> float:
         return time.monotonic() if now is None else now
 
-    @staticmethod
-    def _parse_target(value: Any) -> tuple[float, float]:
-        if isinstance(value, (list, tuple)):
-            return float(value[0]), float(value[1])
-        return float(value), float(value)
-
     @staticmethod
     def _apply_jitter(value: float, jitter: float | list[float]) -> float:
         if isinstance(jitter, (list, tuple)):
@@ -1252,7 +1255,7 @@ class ThrottlingScopeManager:
         # delay or jump the concurrency limit by several steps at once).
         elapsed_windows = int((now - self._rampup_window_start) // self._window)
         self._rampup_window_start += elapsed_windows * self._window
-        if self._rampup_backoffs < self._rampup_target[0]:
+        if self._rampup_backoffs < self._rampup_target:
             self._rampup_step()
         self._rampup_backoffs = 0
 
@@ -1261,10 +1264,12 @@ class ThrottlingScopeManager:
         if self._backoff_level > 0:
             return
         if self._delay > self._rampup_min_delay:
+            # Lower only the effective delay while probing for headroom; the
+            # configured base delay is left untouched so it stays the recovery
+            # target on backoff and the value reported by get_base_delay().
             self._delay = max(
                 self._rampup_min_delay, self._delay * self._rampup_delay_factor
             )
-            self._base_delay = min(self._base_delay, self._delay)
         else:
             # Rampup is only enabled with a concurrency limit set.
             assert self._concurrency is not None
@@ -1368,20 +1373,21 @@ class ThrottlingScopeManager:
         self._backoff_level += 1
         self._rampup_backoffs += 1
         if delay is not None:
+            # A hard delay (e.g. a Retry-After header) is a one-time gate: hold
+            # the scope back for at least this long *once*, matching the HTTP
+            # semantics of "do not retry before this time". It is deliberately
+            # not turned into the steady-state inter-request delay; the delay
+            # still grows by one exponential step below (and recovers over
+            # BACKOFF_WINDOW), so a small Retry-After does not become a long
+            # standing delay for every later request.
             hard = min(float(delay), self._max_delay) if cap else float(delay)
             self._in_backoff_until = now + hard
-            self._delay = max(self._delay, hard, self._min_delay)
-            if cap:
-                self._delay = min(self._delay, self._max_delay)
-        else:
-            grown = (
-                self._delay * self._delay_factor if self._delay > 0 else self._min_delay
-            )
-            grown = max(self._min_delay, grown)
-            grown = min(grown, self._max_delay)
-            self._delay = min(
-                self._apply_jitter(grown, self._backoff_jitter), self._max_delay
-            )
+        grown = self._delay * self._delay_factor if self._delay > 0 else self._min_delay
+        grown = max(self._min_delay, grown)
+        grown = min(grown, self._max_delay)
+        self._delay = min(
+            self._apply_jitter(grown, self._backoff_jitter), self._max_delay
+        )
         self._next_allowed_time = now + self._delay
 
     def reconcile_quota(