fix(throttling-manager): Correct the backoff for concurrent 429s and zero Retry-After

docs/guides/request_throttling.mdx

@@ -34,7 +34,7 @@ To use request throttling, create a <ApiLink to="class/ThrottlingRequestManager"
 
 1. **Insertion-time routing**: When you add requests via `add_request` or `add_requests`, each request is checked against the configured domain list. Matching requests go directly into a per-domain sub-manager; all others go to the inner manager. Each request added this way lives in exactly one store, so it is deduplicated there.
 
-2. **429 backoff**: When the crawler detects an HTTP 429 response, the `ThrottlingRequestManager` records an exponential backoff delay for that domain (starting at 2s, doubling up to 60s). If the response includes a `Retry-After` header, that value takes priority.
+2. **429 backoff**: When the crawler detects an HTTP 429 response, the `ThrottlingRequestManager` records an exponential backoff delay for that domain (starting at 2s, doubling up to 60s). Requests already in flight when the limit was hit are treated as a single rate-limit event, so the delay doubles once per backoff window rather than once per 429. Once the domain goes a full extra window without rate-limiting, the next 429 starts the backoff over at the initial delay. If the response includes a `Retry-After` header with a positive delay, that value takes priority.
 
 3. **Crawl-delay**: If `robots.txt` specifies a `crawl-delay`, the manager enforces a minimum interval between requests to that domain.
 

src/crawlee/_utils/http.py

@@ -19,7 +19,7 @@ def parse_retry_after_header(value: str | None) -> timedelta | None:
         value: The raw Retry-After header value.
 
     Returns:
-        A timedelta representing the delay, or None if the header is missing or unparsable.
+        A timedelta representing the delay, or None if the header is missing, unparsable, or not a positive delay.
     """
     if not value:
         return None
@@ -30,10 +30,10 @@ def parse_retry_after_header(value: str | None) -> timedelta | None:
     except ValueError:
         pass  # Not an integer, fall through to the HTTP-date form below.
     else:
-        if seconds < 0:
-            # A negative delay is malformed. Reject it instead of returning a negative `timedelta`, which would
-            # push `throttled_until` into the past and silently disable the 429 back-off downstream.
-            logger.debug(f'Retry-After delay-seconds {value!r} is negative; ignoring.')
+        if seconds <= 0:
+            # A negative delay is malformed and a zero one carries no backoff, so reject both and let the caller
+            # apply its own backoff instead.
+            logger.debug(f'Retry-After delay-seconds {value!r} is not positive; ignoring.')
             return None
         return timedelta(seconds=seconds)
 

src/crawlee/request_loaders/_throttling_request_manager.py

@@ -28,7 +28,13 @@
 TRequestManager = TypeVar('TRequestManager', bound=RequestManager)
 
 _NEVER_THROTTLED = datetime.min.replace(tzinfo=timezone.utc)
-"""Sentinel `throttled_until` value meaning the domain has no active backoff."""
+"""Sentinel timestamp meaning one of a domain's throttle clocks has never been armed."""
+
+_MAX_BACKOFF_EXPONENT = 20
+"""Highest exponent the 429 backoff doubles to. `max_delay` caps the delay far below this, while an unbounded exponent
+eventually overflows the `timedelta` multiplication. Low enough that the doubling stays representable for any
+`base_delay` up to a year.
+"""
 
 
 @docs_group('Request loaders')
@@ -143,14 +149,14 @@ async def purge(self) -> None:
         """Empty the inner manager and all sub-managers, and reset transient per-domain throttle state.
 
         The configured domain list and any robots.txt-derived `crawl_delay` are preserved. Only the dynamic backoff
-        state (consecutive 429 counter and `throttled_until`) is cleared. Sub-managers stay open; they're just emptied.
+        state (consecutive 429 counter and the throttle clocks) is cleared. Sub-managers stay open; they're just
+        emptied.
         """
         await self._ensure_sub_managers()
         await asyncio.gather(self._inner.purge(), *(sm.purge() for sm in self._sub_managers.values()))
         self._in_flight_from_inner.clear()
         for state in self._domain_states.values():
-            state.consecutive_429_count = 0
-            state.throttled_until = _NEVER_THROTTLED
+            state.reset_throttling()
 
     @override
     async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None:
@@ -255,7 +261,6 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
         manager = self._fetch_owner(request)
         result = await manager.mark_request_as_handled(request)
         self._clear_fetch_owner(request)
-        self.record_success(request.url)
         return result
 
     @override
@@ -298,33 +303,58 @@ async def is_finished(self) -> bool:
     def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None) -> bool:
         """Record a 429 Too Many Requests response for the domain of the given URL.
 
-        Increments the consecutive 429 count and calculates the next allowed request time using exponential backoff or
-        the `Retry-After` value.
+        Advances the consecutive 429 count and calculates the next allowed request time using exponential backoff or
+        the `Retry-After` value. Only the first 429 of a burst advances the count, so the delay tracks how hard the
+        domain pushes back, not how many requests were in flight.
 
         Args:
             url: The URL that received a 429 response.
-            retry_after: Optional delay from the `Retry-After` header. If provided, it takes priority over the
-                calculated exponential backoff.
+            retry_after: Optional delay from the `Retry-After` header. If it describes a positive delay, it takes
+                priority over the calculated exponential backoff.
 
         Returns:
-            True if the URL's domain is configured for throttling and the delay was applied; False if the domain is not
-            in the configured `domains` list, in which case the call is a no-op.
+            True if the URL's domain is configured for throttling, whether or not this 429 advanced the backoff; False
+            if the domain is not in the configured `domains` list, in which case the call is a no-op.
         """
         state = self._get_domain_state(url)
         if state is None:
             return False
 
+        now = datetime.now(timezone.utc)
+
+        # Requests in flight when the limit was hit all come back 429. That is one rate-limit event, so only the first
+        # advances the exponent. Checking `crawl_delay_until` too would swallow every 429, as it is armed on every
+        # dispatch.
+        if now < state.backoff_until:
+            logger.debug(
+                f'Ignoring an HTTP 429 from domain "{state.domain}" received during an active backoff '
+                f'(consecutive: {state.consecutive_429_count}).'
+            )
+            return True
+
+        # The domain has been quiet for a full extra window, so this 429 opens a new run instead of continuing the old.
+        if now >= state.backoff_decays_at:
+            state.consecutive_429_count = 0
+
         state.consecutive_429_count += 1
-        delay = retry_after if retry_after is not None else self._base_delay * (2 ** (state.consecutive_429_count - 1))
+
+        # A non-positive `Retry-After` is no delay at all, so fall back to the backoff and let it engage.
+        if retry_after is not None and retry_after > timedelta(0):
+            delay = retry_after
+            source = 'Retry-After header'
+        else:
+            delay = self._base_delay * 2 ** min(state.consecutive_429_count - 1, _MAX_BACKOFF_EXPONENT)
+            source = 'exponential backoff'
+
         if delay > self._max_delay:
-            source = 'Retry-After header' if retry_after is not None else 'exponential backoff'
             logger.warning(
                 f'Capping {source} delay of {delay.total_seconds():.1f}s for domain "{state.domain}" '
                 f'to max_delay ({self._max_delay.total_seconds():.1f}s); the domain may continue to rate-limit. '
                 f'Consider increasing max_delay if this recurs.'
             )
             delay = self._max_delay
-        state.throttled_until = datetime.now(timezone.utc) + delay
+
+        state.apply_backoff(now, delay)
 
         logger.info(
             f'Rate limit (429) detected for domain "{state.domain}" '
@@ -333,7 +363,10 @@ def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None)
         return True
 
     def record_success(self, url: str) -> None:
-        """Record a successful request, resetting the backoff state for that domain.
+        """Reset a domain's consecutive 429 count, so the next 429 starts the backoff over at `base_delay`.
+
+        An active backoff window is not lifted. The manager does not call this itself; the count decays on its own once
+        the domain has stopped rate-limiting for a full extra window.
 
         Args:
             url: The URL that received a successful response.
@@ -457,11 +490,11 @@ def _fetchable_domains(self) -> list[str]:
     def _mark_domain_dispatched(self, domain: str) -> None:
         """Record that a request to this domain was just dispatched.
 
-        If a crawl-delay is configured, push throttled_until forward by that amount.
+        If a crawl-delay is configured, push `crawl_delay_until` forward by that amount.
         """
         state = self._domain_states.get(domain)
-        if state is not None and state.crawl_delay is not None:
-            state.throttled_until = datetime.now(timezone.utc) + state.crawl_delay
+        if state is not None:
+            state.apply_crawl_delay(datetime.now(timezone.utc))
 
     def _fetch_owner(self, request: Request) -> TRequestManager:
         """Return the manager the request must be given back to, leaving its in-flight record in place.
@@ -501,11 +534,47 @@ class _DomainState:
     domain: str
     """The domain being tracked."""
 
-    throttled_until: datetime = _NEVER_THROTTLED
-    """Earliest time the next request to this domain is allowed."""
+    backoff_until: datetime = _NEVER_THROTTLED
+    """Earliest time the next request is allowed by the 429 backoff. Kept apart from `crawl_delay_until`, which is
+    armed on every dispatch and would otherwise pass for an active backoff.
+    """
+
+    crawl_delay_until: datetime = _NEVER_THROTTLED
+    """Earliest time the next request is allowed by the domain's crawl-delay."""
+
+    backoff_decays_at: datetime = _NEVER_THROTTLED
+    """Time after which an incoming 429 is treated as a fresh burst rather than a continuation of the current one."""
 
     consecutive_429_count: int = 0
     """Number of consecutive 429 responses (for exponential backoff)."""
 
     crawl_delay: timedelta | None = None
-    """Minimum interval between requests, used to push `throttled_until` on dispatch."""
+    """Minimum interval between requests, used to push `crawl_delay_until` on dispatch."""
+
+    @property
+    def throttled_until(self) -> datetime:
+        """Earliest time the next request to this domain is allowed by either of its two independent clocks."""
+        return max(self.backoff_until, self.crawl_delay_until)
+
+    def apply_backoff(self, now: datetime, delay: timedelta) -> None:
+        """Block the domain for `delay`.
+
+        If no 429 arrives for another `delay` after the domain becomes dispatchable again, the exponent resets.
+        """
+        self.backoff_until = now + delay
+        # The quiet period runs from the moment the domain becomes dispatchable again, not from `backoff_until`. A
+        # crawl-delay longer than `delay` sets the retry cadence, and measuring from `backoff_until` would expire the
+        # window before the domain is even retried, making every 429 look like a fresh burst.
+        self.backoff_decays_at = self.throttled_until + delay
+
+    def apply_crawl_delay(self, now: datetime) -> None:
+        """Block the domain for its crawl-delay, if it declared one."""
+        if self.crawl_delay is not None:
+            self.crawl_delay_until = now + self.crawl_delay
+
+    def reset_throttling(self) -> None:
+        """Clear the transient throttle state."""
+        self.consecutive_429_count = 0
+        self.backoff_until = _NEVER_THROTTLED
+        self.crawl_delay_until = _NEVER_THROTTLED
+        self.backoff_decays_at = _NEVER_THROTTLED