fix(throttling-manager): Defer throttled 429s without spending retries and give up on persistently rate-limited domains

docs/guides/request_throttling.mdx

@@ -32,15 +32,29 @@ To use request throttling, create a <ApiLink to="class/ThrottlingRequestManager"
 
 ## How it works
 
-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.
+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. A request that reached the inner manager before its domain was configured moves into the domain's sub-manager if it's fetched during the domain's cooldown. Otherwise it's crawled from the inner manager.
 
 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.
 
 4. **Fair scheduling**: `fetch_next_request` sorts available sub-managers by how long each domain has been waiting, ensuring no domain is starved.
 
-5. **Cooldown handling**: While a domain is in a cooldown, its queued requests don't count as dispatchable, so the crawler's autoscaled pool idles instead of keeping a worker slot blocked. The requests still count towards completion, so the crawl waits for them and finishes only once every one has been handled.
+5. **Cooldown handling**: While a domain is in a cooldown, its queued requests don't count as dispatchable, so the crawler's autoscaled pool idles instead of keeping a worker slot blocked. The requests still count towards completion, so the crawl waits for them and finishes only once every one has been handled. A domain that never stops rate-limiting doesn't keep the crawl running forever. For details, see [Persistent rate limiting](#persistent-rate-limiting).
+
+## Persistent rate limiting
+
+A 429 from a throttled domain doesn't fail the request. The request goes back to its queue and waits out the backoff, while the crawler keeps crawling other domains. The retry doesn't count towards `max_request_retries` or `max_session_rotations`, and the session isn't marked as bad. The 429 doesn't reach `error_handler` or `failed_request_handler`. Setting `ignore_http_error_status_codes={429}` doesn't pass the response to your request handler either, because the request is retried once the backoff ends.
+
+Some domains keep rate-limiting every request no matter how long the crawler waits. A domain that has done so for longer than the manager's `max_domain_stall` (15 minutes by default) counts as stalled. The crawler still finishes every other domain first, including domains waiting out a backoff or a crawl-delay. Once only stalled domains have requests left, `crawler.run()` raises <ApiLink to="class/PersistentRateLimitError">`PersistentRateLimitError`</ApiLink> instead of waiting forever. The error message names the stalled domains. To fix the stall, lower the crawler's concurrency or drop those domains.
+
+The requests of a stalled domain stay queued. To resume them, call `crawler.run()` again once `max_domain_stall` has passed since the domain's last 429, or start a new process with `purge_on_start` disabled.
+
+Note that:
+
+- A crawler running with `keep_alive=True` never raises the error. Instead, a stalled domain's requests don't make the crawler start new tasks until `max_domain_stall` has passed since the domain's last 429. A task started for other requests can still pick them up.
+- The stall clock survives a failed run. Rerunning in the same process within `max_domain_stall` of the last 429 raises the error again, before any request is sent if nothing else is queued. To reset the clock at once, create a new crawler with a new `ThrottlingRequestManager`.
+- A backoff or `Retry-After` delay longer than `max_domain_stall` delays detection until the domain's next 429.
 
 ## Sub-manager storage
 

src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py

@@ -15,7 +15,7 @@
 from crawlee._utils.time import SharedTimeout
 from crawlee._utils.urls import to_absolute_url_iterator
 from crawlee.crawlers._basic import BasicCrawler, BasicCrawlerOptions, ContextPipeline
-from crawlee.errors import SessionError
+from crawlee.errors import RequestThrottledError, SessionError
 from crawlee.statistics import StatisticsState
 
 from ._http_crawling_context import HttpCrawlingContext, ParsedHttpCrawlingContext, TParseResult, TSelectResult
@@ -295,6 +295,7 @@ async def _handle_status_code_response(
             context: The current crawling context containing the HTTP response.
 
         Raises:
+            RequestThrottledError: If the response is a 429 from a domain throttled by a `ThrottlingRequestManager`.
             SessionError: If the status code indicates the session is blocked.
             HttpStatusCodeError: If the status code represents a server error or is explicitly configured as an error.
             HttpClientStatusCodeError: If the status code represents a client error.
@@ -303,11 +304,12 @@ async def _handle_status_code_response(
             The original crawling context if no errors are detected.
         """
         status_code = context.http_response.status_code
-        self._record_rate_limit_status_code(
+        if self._record_rate_limit_status_code(
             status_code,
             request_url=context.request.url,
             retry_after_header=context.http_response.headers.get('retry-after'),
-        )
+        ):
+            raise RequestThrottledError(f'{context.request.url} responded with 429.')
         if self._retry_on_blocked:
             self._raise_for_session_blocked_status_code(context.session, status_code)
         self._raise_for_error_status_code(status_code)

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -57,8 +57,10 @@
     ContextPipelineInterruptedError,
     HttpClientStatusCodeError,
     HttpStatusCodeError,
+    PersistentRateLimitError,
     RequestCollisionError,
     RequestHandlerError,
+    RequestThrottledError,
     SessionError,
     UserDefinedErrorHandlerError,
     UserHandlerTimeoutError,
@@ -342,7 +344,8 @@ def __init__(
             additional_http_error_status_codes: Additional HTTP status codes to treat as errors,
                 triggering automatic retries when encountered.
             ignore_http_error_status_codes: HTTP status codes that are typically considered errors but should be treated
-                as successful responses.
+                as successful responses. Doesn't apply to a 429 from a domain throttled by a `ThrottlingRequestManager`,
+                which is retried later.
             concurrency_settings: Settings to fine-tune concurrency levels.
             request_handler_timeout: Maximum duration allowed for a single request handler to run.
             statistics: A custom `Statistics` instance, allowing the use of non-default configuration.
@@ -698,6 +701,10 @@ async def run(
                 queue will be purged. A run that ended with an exception does not count as a previous run, so a
                 retry keeps the requests that were still pending. Named request queues are considered persistent
                 and are never purged implicitly.
+
+        Raises:
+            PersistentRateLimitError: If a domain throttled by a `ThrottlingRequestManager` has rate-limited every
+                request for longer than the manager's `max_domain_stall` and no other requests are left.
         """
         if self._running:
             raise RuntimeError(
@@ -1420,6 +1427,19 @@ async def __is_finished_function(self) -> bool:
         if self._keep_alive:
             return False
 
+        # Check for a stall only when the pool is idle. An in-flight request can still enqueue new work or end the
+        # stall.
+        if (
+            isinstance(self._request_manager, ThrottlingRequestManager)
+            and self._autoscaled_pool.current_concurrency == 0
+            and (reason := await self._request_manager.get_stall_reason()) is not None
+        ):
+            error = PersistentRateLimitError(f'Giving up: {reason}')
+            self._logger.error(
+                'Giving up the crawl because a domain keeps rate-limiting every request.', exc_info=error
+            )
+            raise error
+
         request_manager = await self.get_request_manager()
         return await request_manager.is_finished()
 
@@ -1507,6 +1527,10 @@ async def __run_task_function(self) -> None:
             await self._handle_request_error(context, request_error)
 
         except RequestHandlerError as primary_error:
+            if isinstance(primary_error.wrapped_exception, RequestThrottledError) and self._is_held_back(request):
+                await self._defer_throttled_request(request, primary_error.wrapped_exception)
+                return
+
             primary_error = cast(
                 'RequestHandlerError[TCrawlingContext]', primary_error
             )  # valid thanks to ContextPipeline
@@ -1568,6 +1592,11 @@ async def __run_task_function(self) -> None:
             await self._mark_request_as_handled(request)
 
         except ContextPipelineInitializationError as initialization_error:
+            wrapped = initialization_error.wrapped_exception
+            if isinstance(wrapped, RequestThrottledError) and self._is_held_back(request):
+                await self._defer_throttled_request(request, wrapped)
+                return
+
             self._logger.debug(
                 'An exception occurred during the initialization of crawling context',
                 exc_info=initialization_error,
@@ -1589,6 +1618,32 @@ async def __run_task_function(self) -> None:
                 except Exception:  # noqa: PERF203
                     self._logger.exception('Error in deferred cleanup')
 
+    def _is_held_back(self, request: Request) -> bool:
+        """Check whether a `ThrottlingRequestManager` holds the request's domain back, so a deferral can't spin."""
+        manager = self._request_manager
+        if isinstance(manager, ThrottlingRequestManager) and manager.is_throttled(request.url):
+            return True
+
+        self._logger_once.log(
+            f'`RequestThrottledError` was raised for {request.url}, but no `ThrottlingRequestManager` holds its domain '
+            "back, so it's handled as an ordinary error. Raise it only for a configured domain, after "
+            '`ThrottlingRequestManager.record_domain_delay` has put the domain into a backoff.',
+            key='throttled_error_without_backoff',
+            level=logging.WARNING,
+        )
+        return False
+
+    async def _defer_throttled_request(self, request: Request, error: RequestThrottledError) -> None:
+        """Give a rate-limited request back to the request manager without counting a failure or a retry."""
+        request.state = RequestState.ERROR_HANDLER
+        self._logger.debug(
+            f'Deferring request because its domain is rate-limiting us. {error}',
+            extra={'url': request.url, 'unique_key': request.unique_key},
+        )
+        request_manager = await self.get_request_manager()
+        await request_manager.reclaim_request(request, forefront=request.forefront)
+        self._statistics.record_request_processing_deferral(request.unique_key)
+
     async def _run_request_handler(self, context: BasicCrawlingContext) -> None:
         context.request.state = RequestState.BEFORE_NAV
         await self._context_pipeline(
@@ -1630,7 +1685,7 @@ def _record_rate_limit_status_code(
         *,
         request_url: str,
         retry_after_header: str | None = None,
-    ) -> None:
+    ) -> bool:
         """Record a 429 Too Many Requests response so the request's domain gets a backoff.
 
         Rate limiting is independent of session blocking, so this runs for every response regardless of
@@ -1640,9 +1695,13 @@ def _record_rate_limit_status_code(
             status_code: The HTTP status code to check.
             request_url: The request URL, used for per-domain rate limit tracking.
             retry_after_header: The value of the `Retry-After` response header, if present.
+
+        Returns:
+            True if a `ThrottlingRequestManager` recorded the 429 for the request's domain, so the request should be
+                deferred.
         """
         if status_code != HTTPStatus.TOO_MANY_REQUESTS:
-            return
+            return False
 
         if not isinstance(self._request_manager, ThrottlingRequestManager):
             self._logger_once.log(
@@ -1653,7 +1712,7 @@ def _record_rate_limit_status_code(
                 key='no_throttling_manager_on_429',
                 level=logging.WARNING,
             )
-            return
+            return False
 
         retry_after = parse_retry_after_header(retry_after_header)
         if not self._request_manager.record_domain_delay(request_url, retry_after=retry_after):
@@ -1666,6 +1725,9 @@ def _record_rate_limit_status_code(
                     key=f'unconfigured_throttle_domain:{domain}',
                     level=logging.WARNING,
                 )
+            return False
+
+        return True
 
     def _raise_for_session_blocked_status_code(self, session: Session | None, status_code: int) -> None:
         """Raise an exception if the given status code indicates the session is blocked.

src/crawlee/crawlers/_playwright/_playwright_crawler.py

@@ -21,7 +21,7 @@
 from crawlee._utils.urls import to_absolute_url_iterator
 from crawlee.browsers import BrowserPool
 from crawlee.crawlers._basic import BasicCrawler, BasicCrawlerOptions, ContextPipeline
-from crawlee.errors import SessionError
+from crawlee.errors import RequestThrottledError, SessionError
 from crawlee.fingerprint_suite import DefaultFingerprintGenerator, FingerprintGenerator, HeaderGeneratorOptions
 from crawlee.fingerprint_suite._header_generator import fingerprint_browser_type_from_playwright_browser_type
 from crawlee.http_clients import ImpitHttpClient
@@ -500,6 +500,7 @@ async def _handle_status_code_response(self, context: TPostNavContext) -> AsyncG
             context: The current crawling context containing the response.
 
         Raises:
+            RequestThrottledError: If the response is a 429 from a domain throttled by a `ThrottlingRequestManager`.
             SessionError: If the status code indicates the session is blocked.
             HttpStatusCodeError: If the status code represents a server error or is explicitly configured as an error.
             HttpClientStatusCodeError: If the status code represents a client error.
@@ -508,11 +509,12 @@ async def _handle_status_code_response(self, context: TPostNavContext) -> AsyncG
             The original crawling context if no errors are detected.
         """
         status_code = context.response.status
-        self._record_rate_limit_status_code(
+        if self._record_rate_limit_status_code(
             status_code,
             request_url=context.request.url,
             retry_after_header=context.response.headers.get('retry-after'),
-        )
+        ):
+            raise RequestThrottledError(f'{context.request.url} responded with 429.')
         if self._retry_on_blocked:
             self._raise_for_session_blocked_status_code(context.session, status_code)
         self._raise_for_error_status_code(status_code)

src/crawlee/errors.py

@@ -13,9 +13,11 @@
     'ContextPipelineInterruptedError',
     'HttpClientStatusCodeError',
     'HttpStatusCodeError',
+    'PersistentRateLimitError',
     'ProxyError',
     'RequestCollisionError',
     'RequestHandlerError',
+    'RequestThrottledError',
     'ServiceConflictError',
     'SessionError',
     'UserDefinedErrorHandlerError',
@@ -116,3 +118,23 @@ class ContextPipelineInterruptedError(Exception):
 @docs_group('Errors')
 class RequestCollisionError(Exception):
     """Raised when a request cannot be processed due to a conflict with required resources."""
+
+
+@docs_group('Errors')
+class RequestThrottledError(Exception):
+    """Raised when a domain throttled by a `ThrottlingRequestManager` responds with 429.
+
+    The request is retried later without spending a retry or marking the session as bad. If you raise it yourself, call
+    `ThrottlingRequestManager.record_domain_delay` first. Otherwise it's handled as an ordinary error.
+    """
+
+    def __init__(self, message: str = 'Request is being retried later because its domain is rate-limiting us') -> None:
+        super().__init__(message)
+
+
+@docs_group('Errors')
+class PersistentRateLimitError(Exception):
+    """Raised when a domain has rate-limited every request for longer than `max_domain_stall`.
+
+    Raised only once no other requests are left. The domain's requests stay queued.
+    """