WIP

docs/topics/throttling.rst

@@ -231,8 +231,8 @@ setting:
 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 reach or exceed the target do not ramp up; the rate is
+raises the concurrency limit of the 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`
@@ -460,10 +460,6 @@ following keys:
     for domain scopes, :setting:`CONCURRENT_REQUESTS_PER_IP` for IP scopes, and
     :setting:`THROTTLING_SCOPE_CONCURRENCY` for any other scope.
 
-``min_concurrency`` (:class:`int`)
-    Concurrency floor that :ref:`backoff <backoff>` and :ref:`rampup <rampup>`
-    never drop below. Defaults to ``1``.
-
 ``delay`` (:class:`float`)
     Minimum seconds between requests for the scope. Defaults to
     :setting:`DOWNLOAD_DELAY`.

scrapy/throttling.py

@@ -71,9 +71,6 @@ class ThrottlingScopeConfig(TypedDict, total=False):
 
     concurrency: int
 
-    min_concurrency: int
-    """Floor. Never drop below this during backoff/rampup."""
-
     delay: float
 
     jitter: float | list[float]
@@ -875,14 +872,6 @@ class ThrottlingManager:
             conflicts.append(f"delay={config['delay']!r} < Crawl-delay {capped}")
         if config.get("concurrency") is not None and int(config["concurrency"]) > 1:
             conflicts.append(f"concurrency={config['concurrency']!r} > 1")
-        # A min_concurrency floor above 1 keeps the scope above the single-slot
-        # concurrency that a Crawl-delay implies, since set_concurrency(1) is
-        # clamped back up to it.
-        if (
-            config.get("min_concurrency") is not None
-            and int(config["min_concurrency"]) > 1
-        ):
-            conflicts.append(f"min_concurrency={config['min_concurrency']!r} > 1")
         if conflicts:
             logger.warning(
                 f"Throttling scope {scope_id!r} is configured with {' and '.join(conflicts)}, "
@@ -1120,7 +1109,7 @@ class ThrottlingScopeManager:
 
     -   When the scope is configured with a ``"concurrency"`` limit (or with
         ``"rampup"``), no more than that many requests are allowed in flight at
-        once, never dropping below the ``"min_concurrency"`` floor.
+        once.
 
     -   When the scope sets ``"rampup": True``, throughput is increased every
         :setting:`BACKOFF_WINDOW` that stays under :setting:`RAMPUP_BACKOFF_TARGET`
@@ -1176,7 +1165,6 @@ class ThrottlingScopeManager:
             for exc in backoff.get("exceptions", settings.getlist("BACKOFF_EXCEPTIONS"))
         )
         self._window: float = settings.getfloat("BACKOFF_WINDOW")
-        self._min_concurrency: int = int(config.get("min_concurrency", 1))
 
         # Rampup.
         rampup = config.get("rampup")
@@ -1197,7 +1185,8 @@ class ThrottlingScopeManager:
         if configured_concurrency is not None:
             self._concurrency: int | None = int(configured_concurrency)
         elif self._rampup_enabled:
-            self._concurrency = self._min_concurrency
+            # Rampup starts conservative at a single slot and probes upward.
+            self._concurrency = 1
         else:
             self._concurrency = _default_scope_concurrency(settings, self._id) or None
         # Used as the load denominator when the scope enforces no explicit
@@ -1239,9 +1228,16 @@ class ThrottlingScopeManager:
         return value * random.uniform(1 - jitter, 1 + jitter)  # noqa: S311
 
     def _effective_delay(self) -> float:
-        if self._backoff_level == 0 and self._delay > 0:
-            return self._apply_jitter(self._delay, self._jitter)
-        return self._delay
+        # ``self._delay`` is the deterministic delay (the base delay, or the
+        # bounded exponential value while backing off); jitter is applied here,
+        # per use, so successive delays spread out without compounding and
+        # without piling probability mass on the min/max bounds (which clipping
+        # a jittered value would do). While backing off, the backoff jitter
+        # applies; otherwise the plain delay jitter does.
+        if self._delay <= 0:
+            return self._delay
+        jitter = self._backoff_jitter if self._backoff_level > 0 else self._jitter
+        return self._apply_jitter(self._delay, jitter)
 
     def _recover(self, now: float) -> None:
         if self._backoff_level == 0 or self._last_backoff_time is None:
@@ -1321,9 +1317,13 @@ class ThrottlingScopeManager:
             self._consumed = 0.0
 
     def can_send(self, now: float | None = None, amount: float | None = None) -> float:
+        # can_send() only refreshes passive, time-based state (backoff recovery
+        # and the quota window) to reflect the current time; it performs no
+        # active throughput probing. That way a readiness check (is_ready() /
+        # get_time_until_ready()) has no side effect on the send rate: rampup
+        # only advances on an actual send, from record_sent().
         now = self._now(now)
         self._recover(now)
-        self._maybe_rampup(now)
         self._maybe_reset_quota(now)
         waits = [0.0]
         if self._in_backoff_until is not None:
@@ -1348,6 +1348,9 @@ class ThrottlingScopeManager:
         self._last_seen = now
         if self._in_backoff_until is not None and now >= self._in_backoff_until:
             self._in_backoff_until = None
+        # An actual send is the cue to probe for more throughput (rampup),
+        # rather than a mere readiness check; see can_send().
+        self._maybe_rampup(now)
         self._next_allowed_time = now + self._effective_delay()
         self._active += 1
         if self._quota is not None and amount is not None:
@@ -1412,12 +1415,11 @@ class ThrottlingScopeManager:
             hard = min(float(delay), self._max_delay) if cap else float(delay)
             self._in_backoff_until = now + hard
         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
+        # Store the deterministic, bounded delay; jitter is applied per use in
+        # _effective_delay(), so it neither compounds across successive backoff
+        # steps nor concentrates on BACKOFF_MIN_DELAY / BACKOFF_MAX_DELAY.
+        self._delay = min(max(self._min_delay, grown), self._max_delay)
+        self._next_allowed_time = now + self._effective_delay()
 
     def reconcile_quota(
         self,
@@ -1461,7 +1463,7 @@ class ThrottlingScopeManager:
             self._delay = delay
 
     def set_concurrency(self, concurrency: int) -> None:
-        self._concurrency = max(self._min_concurrency, int(concurrency))
+        self._concurrency = max(1, int(concurrency))
         self._fire_slot_waiters()
 
     def is_idle(self, now: float, max_idle: float) -> bool:

scrapy/utils/asyncio.py

@@ -8,7 +8,7 @@ import time
 from collections.abc import AsyncIterator, Callable, Coroutine, Iterable, Sequence
 from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar
 
-from twisted.internet.defer import Deferred, DeferredList
+from twisted.internet.defer import Deferred
 from twisted.internet.task import LoopingCall, deferLater
 from twisted.internet.threads import deferToThread
 
@@ -347,6 +347,10 @@ async def wait_for_first(
     Unfired deferreds in the ``pending`` set are neither cancelled nor
     otherwise modified; the caller is responsible for any cleanup.
 
+    A deferred that fails counts as done and its failure is **not** re-raised
+    here (it stays on the deferred for the caller to inspect or handle), exactly
+    as a failed awaitable lands in the ``done`` set of :func:`asyncio.wait`.
+
     Returns ``(set(), set())`` immediately when *deferreds* is empty.
 
     Works transparently in asyncio-reactor, non-asyncio-reactor, and
@@ -377,17 +381,31 @@ async def wait_for_first(
     # circular import
     from scrapy.utils.defer import maybe_deferred_to_future  # noqa: PLC0415
 
-    timeout_deferred = (
-        deferLater(reactor, timeout, lambda: None) if timeout is not None else None
-    )
-    waiter = DeferredList(
-        [*deferreds, *([] if timeout_deferred is None else [timeout_deferred])],
-        fireOnOneCallback=True,
-        fireOnOneErrback=True,
-        consumeErrors=True,
-    )
+    # Fire a single signal Deferred as soon as any input Deferred (or the
+    # timeout) completes, whether it succeeds or fails. Unlike a DeferredList
+    # with fireOnOneErrback (which would raise the first failure into this
+    # coroutine), _fire passes each result through untouched, so a failure just
+    # marks its Deferred as done and stays there for the caller — matching the
+    # asyncio branch and asyncio.wait(return_when=FIRST_COMPLETED).
+    signal: Deferred[None] = Deferred()
+
+    def _fire(result: Any) -> Any:
+        if not signal.called:
+            signal.callback(None)
+        return result
+
+    for d in deferreds:
+        d.addBoth(_fire)
+
+    timeout_deferred: Deferred[Any] | None = None
+    if timeout is not None:
+        timeout_deferred = deferLater(reactor, timeout, lambda: None)
+        timeout_deferred.addBoth(_fire)
+        # Swallow the CancelledError from the finally-block cancel() below.
+        timeout_deferred.addErrback(lambda _: None)
+
     try:
-        await maybe_deferred_to_future(waiter)
+        await maybe_deferred_to_future(signal)
     finally:
         if timeout_deferred is not None and not timeout_deferred.called:
             timeout_deferred.cancel()