diff --git a/docs/guides/request_throttling.mdx b/docs/guides/request_throttling.mdx index 81cf700104..c02ea9cf96 100644 --- a/docs/guides/request_throttling.mdx +++ b/docs/guides/request_throttling.mdx @@ -32,7 +32,7 @@ To use request throttling, create a `PersistentRateLimitError` 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 diff --git a/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py b/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py index e3f87b4bfc..ed4d79b143 100644 --- a/src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py +++ b/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) diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index 39e928e2f9..7addbb5c4c 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/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. diff --git a/src/crawlee/crawlers/_playwright/_playwright_crawler.py b/src/crawlee/crawlers/_playwright/_playwright_crawler.py index 39c31afda1..bd7cdbedc4 100644 --- a/src/crawlee/crawlers/_playwright/_playwright_crawler.py +++ b/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) diff --git a/src/crawlee/errors.py b/src/crawlee/errors.py index 539bcf7711..ed1c23933b 100644 --- a/src/crawlee/errors.py +++ b/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. + """ diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index 49465c7ae9..c87124ac9a 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -39,20 +39,23 @@ `base_delay` up to a year. """ +_MAX_INNER_MIGRATIONS = 1000 +"""How many requests one fetch may move out of `inner` before it returns `None`.""" + @docs_group('Request loaders') class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): """A request manager that wraps another and enforces per-domain delays. Requests for explicitly configured domains are routed into dedicated sub-managers, so each request lives in exactly - one store and is deduplicated there. A request that reached `inner` before its domain was configured stays and is - completed there, without the domain's delay. + one store and is deduplicated there. A configured-domain request that's already in `inner` moves into the domain's + sub-manager if it's fetched during the domain's cooldown. Otherwise it's dispatched and completed in `inner`. `fetch_next_request()` takes from the sub-manager whose domain has been waiting the longest, skipping domains in a cooldown, and falls back to the inner manager when no sub-manager yields a request. If nothing can be dispatched - right now, it returns `None` rather than waiting, so the caller's task slot is released. `is_empty()` reports the - same view and reads as empty while every remaining request sits in a cooldown, whereas `is_finished()` counts those - requests, so the crawl idles until they are dispatchable instead of ending early. + right now, it returns `None` rather than waiting, so the caller's task slot is released. `is_empty()` reads as + empty while every remaining request sits in a cooldown or belongs to a stalled domain, whereas `is_finished()` + counts those requests, so the crawl idles until they are dispatchable instead of ending early. Delay sources: - HTTP 429 responses (via `record_domain_delay`) @@ -93,6 +96,7 @@ def __init__( service_locator: ServiceLocator | None = None, base_delay: timedelta = timedelta(seconds=2), max_delay: timedelta = timedelta(seconds=60), + max_domain_stall: timedelta = timedelta(seconds=900), ) -> None: """Initialize the throttling manager. @@ -112,14 +116,22 @@ def __init__( locator, ensuring consistency with the crawler's storage backend. base_delay: Initial delay after the first 429 response from a domain. max_delay: Maximum delay between requests to a rate-limited domain. + max_domain_stall: How long a domain may rate-limit every request before the crawler raises + `PersistentRateLimitError`. A crawler running with `keep_alive` never raises it. The crawler checks + for a stall about twice a second, so a window under a second may not be detected. Raises: - ValueError: If a non-blank entry of `domains` does not yield a hostname a crawled URL could match. + ValueError: If a non-blank entry of `domains` does not yield a hostname a crawled URL could match, or if + `max_domain_stall` is not positive. """ + if max_domain_stall <= timedelta(0): + raise ValueError(f'max_domain_stall must be positive, got {max_domain_stall}.') + self._inner: TRequestManager = inner self._service_locator = service_locator if service_locator is not None else global_service_locator self._base_delay = base_delay self._max_delay = max_delay + self._max_domain_stall = max_domain_stall self._request_manager_opener = request_manager_opener # Padding on an entry would otherwise survive parsing into a key no crawled hostname can match. domain_keys = [self._parse_configured_domain(entry) for d in domains if (entry := d.strip())] @@ -128,11 +140,14 @@ def __init__( self._sub_managers_ready = False self._sub_managers_lock = asyncio.Lock() self._in_flight_from_inner: set[tuple[str, str]] = set() - """`(unique_key, url)` pairs of configured-domain requests that `fetch_next_request` took from `inner`, where - they live if they were added before their domain was listed, and where they must be completed. The URL is part - of the key because an explicit `unique_key` is only unique per store. Identical pairs held by `inner` and by a - sub-manager are indistinguishable, so their completions can cross; both stores hold the key, so the cost is a - duplicate crawl and a retry without the domain's delay.""" + """`(unique_key, url)` pairs of configured-domain requests that `fetch_next_request` took from `inner` and + dispatched, which must be completed there. A request fetched while its domain is in a cooldown moves into the + domain's sub-manager instead. The URL is part of the key because an explicit `unique_key` is only unique per + store. Identical pairs held by `inner` and by a sub-manager are indistinguishable, so their completions can + cross; both stores hold the key, so the cost is a duplicate crawl and a retry without the domain's delay.""" + self._migrated_from_inner = 0 + """Number of requests moved from `inner` into a sub-manager. Both stores count such a request, so the handled + and total counts subtract it.""" @property def inner(self) -> TRequestManager: @@ -152,12 +167,13 @@ 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 the throttle clocks) is cleared. Sub-managers stay open; they're just - emptied. + state (consecutive 429 counter, the throttle clocks and the stall clock) 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() + self._migrated_from_inner = 0 for state in self._domain_states.values(): state.reset_throttling() @@ -245,7 +261,7 @@ async def fetch_next_request(self) -> Request | None: self._mark_domain_dispatched(domain) return request - request = await self._inner.fetch_next_request() + request = await self._fetch_from_inner() if request is not None and self._extract_domain(request.url) in self._domain_states: self._in_flight_from_inner.add((request.unique_key, request.url)) return request @@ -261,6 +277,11 @@ async def reclaim_request(self, request: Request, *, forefront: bool = False) -> @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: await self._ensure_sub_managers() + # Reached on success, when retries run out, or when the request is skipped. None of these ends in a 429, so the + # domain isn't turning every request away. + state = self._get_domain_state(request.url) + if state is not None: + state.rate_limited_since = None manager = self._fetch_owner(request) result = await manager.mark_request_as_handled(request) self._clear_fetch_owner(request) @@ -272,7 +293,7 @@ async def get_handled_count(self) -> int: counts = await asyncio.gather( self._inner.get_handled_count(), *(sm.get_handled_count() for sm in self._sub_managers.values()) ) - return sum(counts) + return sum(counts) - self._migrated_from_inner @override async def get_total_count(self) -> int: @@ -280,18 +301,25 @@ async def get_total_count(self) -> int: counts = await asyncio.gather( self._inner.get_total_count(), *(sm.get_total_count() for sm in self._sub_managers.values()) ) - return sum(counts) + return sum(counts) - self._migrated_from_inner @override async def is_empty(self) -> bool: """Report whether anything can be dispatched right now. - Requests queued for a domain in a cooldown do not count. They still count towards `is_finished`, so the crawl - waits for them. + Requests queued for a domain in a cooldown don't count. They still count towards `is_finished`, so the crawl + waits for them. Requests of a domain that has rate-limited every request for longer than `max_domain_stall` + don't count either. """ await self._ensure_sub_managers() + now = datetime.now(timezone.utc) results = await asyncio.gather( - self._inner.is_empty(), *(self._sub_managers[d].is_empty() for d in self._fetchable_domains()) + self._inner.is_empty(), + *( + self._sub_managers[d].is_empty() + for d in self._fetchable_domains() + if not self._domain_states[d].is_stall_candidate(now, self._max_domain_stall) + ), ) return all(results) @@ -303,12 +331,64 @@ async def is_finished(self) -> bool: ) return all(results) + async def get_stall_reason(self) -> str | None: + """Explain why the crawl can't make progress, or return `None` if it can. + + A domain is stalled once it has rate-limited every request for longer than `max_domain_stall`. It's reported + only when nothing else can be dispatched and no domain in a cooldown has requests left. + """ + await self._ensure_sub_managers() + now = datetime.now(timezone.utc) + dispatchable: list[RequestManager] = [self._inner] + cooling: list[RequestManager] = [] + candidates: list[str] = [] + + for domain, state in self._domain_states.items(): + # A domain that turned away every request for the whole window isn't dispatchable just because its backoff + # lapsed between two 429s. + if state.is_stall_candidate(now, self._max_domain_stall): + candidates.append(domain) + elif now >= state.throttled_until: + dispatchable.append(self._sub_managers[domain]) + else: + cooling.append(self._sub_managers[domain]) + + if not candidates: + return None + + all_dispatchable_empty = all(await asyncio.gather(*(manager.is_empty() for manager in dispatchable))) + if not all_dispatchable_empty: + return None + + # A domain waiting out a cooldown with work left will still make progress, so the crawl isn't stuck yet. + all_cooling_finished = all(await asyncio.gather(*(manager.is_finished() for manager in cooling))) + if not all_cooling_finished: + return None + + candidates_empty = await asyncio.gather(*(self._sub_managers[domain].is_empty() for domain in candidates)) + stalled = [domain for domain, is_empty in zip(candidates, candidates_empty, strict=True) if not is_empty] + if not stalled: + return None + + summary = ', '.join( + f'"{domain}" ({(now - since).total_seconds():.0f}s)' + for domain in stalled + if (since := self._domain_states[domain].rate_limited_since) is not None + ) + window = self._max_domain_stall.total_seconds() + return ( + f'{summary} rate-limited every request for longer than `max_domain_stall` ({window:.0f}s). Waiting ' + "longer will not help - lower the crawler's concurrency, or drop these domains. Their requests are " + 'still queued, so re-running with purge_on_start disabled will resume them if the rate limit lifts.' + ) + 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. 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. + domain pushes back, not how many requests were in flight. Every call, including a manual one or a 429 inside a + burst, also starts or extends the domain's stall clock. Args: url: The URL that received a 429 response. @@ -325,6 +405,10 @@ def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None) now = datetime.now(timezone.utc) + state.last_rate_limited_at = now + if state.rate_limited_since is None: + state.rate_limited_since = now + # 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. @@ -369,7 +453,7 @@ def record_success(self, url: str) -> None: """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. + the domain has stopped rate-limiting for a full extra window. It doesn't affect the stall clock. Args: url: The URL that received a successful response. @@ -379,6 +463,14 @@ def record_success(self, url: str) -> None: logger.debug(f'Resetting rate limit state for domain "{state.domain}" after successful request') state.consecutive_429_count = 0 + def is_throttled(self, url: str) -> bool: + """Check whether the URL's domain is in a cooldown, so requests to it are held back. + + Args: + url: A URL from the domain to check. + """ + return self._is_domain_throttled(self._extract_domain(url)) + def set_crawl_delay(self, url: str, delay_seconds: int) -> None: """Set the robots.txt crawl-delay for a domain. @@ -476,6 +568,40 @@ async def _ensure_sub_managers(self) -> None: self._sub_managers_ready = True + async def _fetch_from_inner(self) -> Request | None: + """Fetch the next request from `inner`, moving each one whose domain is in a cooldown into its sub-manager.""" + for _ in range(_MAX_INNER_MIGRATIONS): + request = await self._inner.fetch_next_request() + if request is None: + return None + + state = self._get_domain_state(request.url) + if state is None or datetime.now(timezone.utc) >= state.throttled_until: + return request + + await self._migrate_to_sub_manager(request) + + return None + + async def _migrate_to_sub_manager(self, request: Request) -> None: + """Move a request fetched from `inner` into its domain's sub-manager.""" + sub_manager = self._sub_managers[self._extract_domain(request.url)] + try: + # A copy, because some storage clients keep the added object and `inner` marks this one handled below. + processed = await sub_manager.add_request(request.model_copy(deep=True)) + # Includes cancellation, which would otherwise leave the request in progress in `inner`. + except BaseException: + await self._inner.reclaim_request(request, forefront=True) + raise + + # A storage that refuses the request returns `None` instead of raising. + if processed is None: + await self._inner.reclaim_request(request, forefront=True) + raise RuntimeError(f'The storage refused to move request {request.url} into its domain queue.') + + await self._inner.mark_request_as_handled(request) + self._migrated_from_inner += 1 + def _is_domain_throttled(self, domain: str) -> bool: """Check if a domain is currently throttled.""" state = self._domain_states.get(domain) @@ -554,6 +680,12 @@ class _DomainState: crawl_delay: timedelta | None = None """Minimum interval between requests, used to push `crawl_delay_until` on dispatch.""" + rate_limited_since: datetime | None = None + """Time of the first 429 since the domain last completed a request.""" + + last_rate_limited_at: datetime | None = None + """Time of the most recent 429 from the domain.""" + @property def throttled_until(self) -> datetime: """Earliest time the next request to this domain is allowed by either of its two independent clocks.""" @@ -570,6 +702,12 @@ def apply_backoff(self, now: datetime, delay: timedelta) -> None: # window before the domain is even retried, making every 429 look like a fresh burst. self.backoff_decays_at = self.throttled_until + delay + def is_stall_candidate(self, now: datetime, window: timedelta) -> bool: + """Check whether the domain still rate-limits us and has rate-limited every request for longer than `window`.""" + since = self.rate_limited_since + last = self.last_rate_limited_at + return since is not None and last is not None and now - last <= window and now - since > window + 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: @@ -581,3 +719,5 @@ def reset_throttling(self) -> None: self.backoff_until = _NEVER_THROTTLED self.crawl_delay_until = _NEVER_THROTTLED self.backoff_decays_at = _NEVER_THROTTLED + self.rate_limited_since = None + self.last_rate_limited_at = None diff --git a/src/crawlee/statistics/_statistics.py b/src/crawlee/statistics/_statistics.py index 3d568fdda2..030269abf2 100644 --- a/src/crawlee/statistics/_statistics.py +++ b/src/crawlee/statistics/_statistics.py @@ -42,6 +42,10 @@ def run(self) -> int: self._runs += 1 return self._runs + def undo_run(self) -> None: + """Stop counting the most recent run towards `retry_count`.""" + self._runs -= 1 + def finish(self) -> timedelta: """Mark the job as finished.""" if self._last_run_at_ns is None: @@ -240,6 +244,13 @@ def record_request_processing_finish(self, request_id_or_key: str) -> None: del self._requests_in_progress[request_id_or_key] + @ensure_context + def record_request_processing_deferral(self, request_id_or_key: str) -> None: + """Mark a request as deferred, so this attempt doesn't count as a retry.""" + record = self._requests_in_progress.get(request_id_or_key) + if record is not None: + record.undo_run() + @ensure_context def record_request_processing_failure(self, request_id_or_key: str) -> None: """Mark a request as failed.""" diff --git a/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py b/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py index 638a285a7b..35a6e998aa 100644 --- a/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py +++ b/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py @@ -7,7 +7,7 @@ from datetime import timedelta from itertools import cycle from typing import TYPE_CHECKING, cast -from unittest.mock import Mock, call, patch +from unittest.mock import AsyncMock, Mock, call, patch import pytest from bs4 import Tag @@ -29,6 +29,7 @@ AdaptivePlaywrightCrawlerStatisticState, ) from crawlee.crawlers._adaptive_playwright._adaptive_playwright_crawling_context import AdaptiveContextError +from crawlee.request_loaders import ThrottlingRequestManager from crawlee.sessions import SessionPool from crawlee.statistics import Statistics from crawlee.storage_clients import SqlStorageClient @@ -937,6 +938,45 @@ async def request_handler(_context: AdaptivePlaywrightCrawlingContext) -> None: mocked_handler.assert_called() +async def test_throttled_429_is_deferred() -> None: + """A throttled 429 from the browser sub crawler is retried later without spending a retry.""" + url = 'https://throttled.placeholder.com/page' + throttler = ThrottlingRequestManager( + await RequestQueue.open(), + domains=['throttled.placeholder.com'], + request_manager_opener=RequestQueue.open, + base_delay=timedelta(milliseconds=50), + max_delay=timedelta(milliseconds=100), + ) + failed_request_handler = AsyncMock() + crawler = AdaptivePlaywrightCrawler.with_beautifulsoup_static_parser( + rendering_type_predictor=_SimpleRenderingTypePredictor( + rendering_types=cycle(['client only']), detection_probability_recommendation=cycle([0]) + ), + request_manager=throttler, + max_request_retries=0, + max_session_rotations=0, + ) + crawler.failed_request_handler(failed_request_handler) + statuses = [429, 200] + handled: list[AdaptivePlaywrightCrawlingContext] = [] + + @crawler.pre_navigation_hook + async def fulfill_with_next_status(context: AdaptivePlaywrightPreNavCrawlingContext) -> None: + status = statuses.pop(0) + await context.page.route(url, lambda route: route.fulfill(status=status, body='')) + + @crawler.router.default_handler + async def handler(context: AdaptivePlaywrightCrawlingContext) -> None: + handled.append(context) + + await crawler.run([url]) + + assert [context.response.status for context in handled] == [200] + failed_request_handler.assert_not_called() + assert handled[0].request.retry_count == 0 + + @pytest.mark.parametrize( 'optional_module_name', [ diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index ce2c2783fb..441a1aa8c6 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -26,7 +26,13 @@ from crawlee._utils.robots import RobotsTxtFile from crawlee.configuration import Configuration from crawlee.crawlers import BasicCrawler -from crawlee.errors import RequestCollisionError, SessionError, UserDefinedErrorHandlerError +from crawlee.errors import ( + PersistentRateLimitError, + RequestCollisionError, + RequestThrottledError, + SessionError, + UserDefinedErrorHandlerError, +) from crawlee.events import Event, EventCrawlerStatusData, LocalEventManager from crawlee.http_clients import HttpClient from crawlee.request_loaders import RequestList, RequestManagerTandem, ThrottlingRequestManager @@ -2620,3 +2626,260 @@ async def handler(context: BasicCrawlingContext) -> None: assert empty_during_cooldown == [True] assert len(dispatched_at) == 2 assert dispatched_at[1] - dispatched_at[0] >= 0.5 + + +THROTTLED_URL = 'https://throttled.placeholder.com/page' + +# Well above the autoscaled pool's idle tick of 0.5 s, so a stall is detected even on a slow runner and lasts for +# more than one tick. +MAX_DOMAIN_STALL = timedelta(milliseconds=1200) + +# Guards the tests against an endless crawl if deferral or stall detection breaks. +MAX_HANDLER_CALLS = 50 + + +async def _open_throttler() -> ThrottlingRequestManager[RequestQueue]: + """Open a throttler for `THROTTLED_URL` with short delays and a short stall window.""" + return ThrottlingRequestManager( + await RequestQueue.open(), + domains=['throttled.placeholder.com'], + request_manager_opener=RequestQueue.open, + base_delay=timedelta(milliseconds=50), + max_delay=timedelta(milliseconds=100), + max_domain_stall=MAX_DOMAIN_STALL, + ) + + +async def test_handler_raised_throttled_error_is_deferred() -> None: + """A `RequestThrottledError` from the handler costs neither a retry nor session reputation.""" + throttler = await _open_throttler() + failed_request_handler = AsyncMock() + crawler = BasicCrawler( + request_manager=throttler, + session_pool=SessionPool(max_pool_size=1), + max_request_retries=0, + ) + crawler.failed_request_handler(failed_request_handler) + contexts = list[BasicCrawlingContext]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + contexts.append(context) + if len(contexts) == 1: + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + + stats = await crawler.run([THROTTLED_URL]) + + assert len(contexts) == 2 + failed_request_handler.assert_not_called() + assert contexts[1].request.retry_count == 0 + assert stats.requests_finished == 1 + assert stats.requests_failed == 0 + session = contexts[1].session + assert session is not None + assert session.error_score == 0 + + +@pytest.mark.parametrize( + 'use_throttler', + [ + pytest.param(True, id='no-recorded-delay'), + pytest.param(False, id='no-throttler'), + ], +) +async def test_throttled_error_without_backoff_is_ordinary_error( + caplog: pytest.LogCaptureFixture, + *, + use_throttler: bool, +) -> None: + """A `RequestThrottledError` whose domain nothing holds back is retried and failed like any other error.""" + request_manager = await _open_throttler() if use_throttler else None + failed_request_handler = AsyncMock() + crawler = BasicCrawler(request_manager=request_manager, max_request_retries=1) + crawler.failed_request_handler(failed_request_handler) + handler = AsyncMock(side_effect=RequestThrottledError) + crawler.router.default_handler(handler) + + with caplog.at_level(logging.WARNING): + stats = await crawler.run([THROTTLED_URL]) + + assert handler.await_count == 2 + failed_request_handler.assert_called_once() + assert stats.requests_failed == 1 + assert any('no `ThrottlingRequestManager` holds its domain back' in record.message for record in caplog.records) + + +async def test_keep_alive_ignores_stall() -> None: + """A crawler running with `keep_alive` keeps waiting on a stalled domain instead of raising.""" + throttler = await _open_throttler() + crawler = BasicCrawler(request_manager=throttler, keep_alive=True) + stall_reasons = list[str | None]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + + async def stop_after_stall() -> None: + stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, timeout=10)) + # Past the pool's next check, where a crawler without `keep_alive` raises. + await asyncio.sleep(0.6) + crawler.stop() + + stopper = asyncio.create_task(stop_after_stall()) + try: + await crawler.run([THROTTLED_URL]) + await stopper + finally: + stopper.cancel() + + assert stall_reasons[0] is not None + + +async def test_stall_waits_for_in_flight_work() -> None: + """A stall isn't raised while a request is in flight, so the work it enqueues is still crawled.""" + throttler = await _open_throttler() + # An overloaded system still runs `min_concurrency` tasks, so the throttled request keeps being dispatched while the + # slow one is in flight. + crawler = BasicCrawler( + request_manager=throttler, + concurrency_settings=ConcurrencySettings(min_concurrency=2, desired_concurrency=2), + ) + visited = list[str]() + stall_reasons = list[str | None]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + visited.append(context.request.url) + if len(visited) > MAX_HANDLER_CALLS: + crawler.stop() + if context.request.url == THROTTLED_URL: + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + if context.request.url == 'https://free.placeholder.com/slow': + stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, timeout=10)) + # Two more pool checks, so at least one ran entirely while the stall was live and this request in flight. + checks = is_finished.await_count + await poll_until_condition(lambda: is_finished.await_count >= checks + 2, timeout=10) + await context.add_requests(['https://free.placeholder.com/next']) + + with ( + patch.object(throttler, 'is_finished', wraps=throttler.is_finished) as is_finished, + pytest.raises(PersistentRateLimitError), + ): + await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) + + assert stall_reasons[0] is not None + assert 'https://free.placeholder.com/next' in visited + + +async def test_max_requests_per_crawl_precedes_stall() -> None: + """Reaching `max_requests_per_crawl` ends the crawl normally even when a domain has stalled.""" + throttler = await _open_throttler() + crawler = BasicCrawler( + request_manager=throttler, + max_requests_per_crawl=1, + # An overloaded system still runs `min_concurrency` tasks, so the throttled request keeps being dispatched + # while the slow one is in flight. + concurrency_settings=ConcurrencySettings(min_concurrency=2, desired_concurrency=2), + ) + calls = 0 + stall_reasons = list[str | None]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + nonlocal calls + calls += 1 + if calls > MAX_HANDLER_CALLS: + crawler.stop() + if context.request.url == THROTTLED_URL: + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, timeout=10)) + + stats = await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) + + assert stall_reasons[0] is not None + assert stats.requests_finished == 1 + + +async def test_stall_waits_for_paced_domain() -> None: + """A stalled domain doesn't end the crawl while a domain paced by its crawl-delay still has work.""" + paced_urls = [f'https://paced.placeholder.com/{i}' for i in range(3)] + throttler = ThrottlingRequestManager( + await RequestQueue.open(), + domains=['throttled.placeholder.com', 'paced.placeholder.com'], + request_manager_opener=RequestQueue.open, + base_delay=timedelta(milliseconds=50), + max_delay=timedelta(milliseconds=100), + max_domain_stall=MAX_DOMAIN_STALL, + ) + throttler.set_crawl_delay(paced_urls[0], 1) + crawler = BasicCrawler(request_manager=throttler) + visited = list[str]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + visited.append(context.request.url) + # A deferral every ~100 ms over a few seconds stays well below this. + if len(visited) > 200: + crawler.stop() + if context.request.url == THROTTLED_URL: + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + + with pytest.raises(PersistentRateLimitError): + await crawler.run([THROTTLED_URL, *paced_urls]) + + assert set(paced_urls) <= set(visited) + + +async def test_inner_held_429_is_deferred_and_stalls() -> None: + """A rate-limited request held by `inner` moves into its domain's sub-manager and can stall the crawl.""" + throttler = await _open_throttler() + await throttler.inner.add_request(THROTTLED_URL) + crawler = BasicCrawler(request_manager=throttler) + calls = 0 + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + nonlocal calls + calls += 1 + if calls > MAX_HANDLER_CALLS: + crawler.stop() + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + + with pytest.raises(PersistentRateLimitError): + await crawler.run() + + assert await throttler.inner.is_finished() is True + assert await throttler.is_finished() is False + + +@pytest.mark.parametrize( + ('outcomes', 'expected_histogram'), + [ + pytest.param(['429', 'ok'], [1], id='deferral-only'), + pytest.param(['error', 'error', '429', 'ok'], [0, 0, 1], id='after-retries'), + ], +) +async def test_deferral_not_counted_as_retry(outcomes: list[str], expected_histogram: list[int]) -> None: + """A deferral doesn't show up in the retry histogram, while earlier retries still do.""" + throttler = await _open_throttler() + crawler = BasicCrawler(request_manager=throttler, max_request_retries=3) + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + outcome = outcomes.pop(0) + if outcome == 'error': + raise RuntimeError('Handler failed') + if outcome == '429': + throttler.record_domain_delay(context.request.url) + raise RequestThrottledError + + stats = await crawler.run([THROTTLED_URL]) + + assert outcomes == [] + assert stats.retry_histogram == expected_histogram diff --git a/tests/unit/crawlers/_http/test_http_crawler.py b/tests/unit/crawlers/_http/test_http_crawler.py index 6a68710d98..817e34a305 100644 --- a/tests/unit/crawlers/_http/test_http_crawler.py +++ b/tests/unit/crawlers/_http/test_http_crawler.py @@ -1,8 +1,9 @@ from __future__ import annotations import json +import logging from datetime import timedelta -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from unittest.mock import AsyncMock, Mock from urllib.parse import parse_qs, urlencode @@ -10,6 +11,8 @@ from crawlee import ConcurrencySettings, Request, RequestState from crawlee.crawlers import HttpCrawler +from crawlee.errors import PersistentRateLimitError +from crawlee.http_clients import HttpClient, HttpCrawlingResult from crawlee.request_loaders import ThrottlingRequestManager from crawlee.sessions import SessionPool from crawlee.statistics import Statistics @@ -23,7 +26,6 @@ from crawlee._types import BasicCrawlingContext from crawlee.crawlers import HttpCrawlingContext - from crawlee.http_clients._base import HttpClient # Payload, e.g. data for a form submission. PAYLOAD = { @@ -690,6 +692,59 @@ async def failed_request_handler(context: BasicCrawlingContext, _error: Exceptio await queue.drop() +def _crawl_result(status_code: int) -> HttpCrawlingResult: + """Build an HTTP client result with the given status code and an empty body.""" + return HttpCrawlingResult(http_response=Mock(status_code=status_code, headers={}, read=AsyncMock(return_value=b''))) + + +@pytest.mark.parametrize( + 'crawler_kwargs', + [ + pytest.param({}, id='default'), + pytest.param({'retry_on_blocked': False}, id='no_retry_on_blocked'), + pytest.param({'ignore_http_error_status_codes': {429}}, id='ignored_429'), + ], +) +async def test_throttled_429_is_deferred(crawler_kwargs: dict[str, Any]) -> None: + """A throttled 429 is retried later without spending a retry, a session rotation or session reputation.""" + throttler = ThrottlingRequestManager( + await RequestQueue.open(), + domains=['throttled.placeholder.com'], + request_manager_opener=RequestQueue.open, + base_delay=timedelta(milliseconds=50), + max_delay=timedelta(milliseconds=100), + ) + http_client = AsyncMock(spec=HttpClient) + http_client.crawl.side_effect = [_crawl_result(429), _crawl_result(200)] + failed_request_handler = AsyncMock() + crawler = HttpCrawler( + http_client=http_client, + request_manager=throttler, + session_pool=SessionPool(max_pool_size=1), + max_request_retries=0, + max_session_rotations=0, + **crawler_kwargs, + ) + crawler.failed_request_handler(failed_request_handler) + handled: list[HttpCrawlingContext] = [] + + @crawler.router.default_handler + async def handler(context: HttpCrawlingContext) -> None: + handled.append(context) + + await crawler.run(['https://throttled.placeholder.com/page']) + + assert [context.http_response.status_code for context in handled] == [200] + failed_request_handler.assert_not_called() + request = handled[0].request + assert request.retry_count == 0 + assert (request.session_rotation_count or 0) == 0 + session = handled[0].session + assert session is not None + assert session.error_score == 0 + assert session.is_usable + + @pytest.mark.parametrize( 'retry_on_blocked', [ @@ -697,31 +752,38 @@ async def failed_request_handler(context: BasicCrawlingContext, _error: Exceptio pytest.param(False, id='no_retry_on_blocked'), ], ) -async def test_records_429_regardless_of_retry_on_blocked( - mock_request_handler: AsyncMock, +async def test_persistent_429_raises_after_stall( server_url: URL, + caplog: pytest.LogCaptureFixture, *, retry_on_blocked: bool, ) -> None: - """Rate limiting is a separate concern from session blocking, so a 429 must be recorded either way.""" - domain = server_url.host or '' - inner = await RequestQueue.open(alias='throttle-429-inner') + """A domain that rate-limits every request for longer than `max_domain_stall` ends the run with an error.""" throttler = ThrottlingRequestManager( - inner, - domains=[domain], + await RequestQueue.open(alias='throttle-429-inner'), + domains=[server_url.host or ''], request_manager_opener=RequestQueue.open, - # Long enough that the assertion below cannot race the backoff expiring. - base_delay=timedelta(seconds=30), + base_delay=timedelta(milliseconds=50), + max_delay=timedelta(milliseconds=100), + max_domain_stall=timedelta(milliseconds=1200), ) + failed_request_handler = AsyncMock() crawler = HttpCrawler( - request_handler=mock_request_handler, + request_handler=AsyncMock(), request_manager=throttler, retry_on_blocked=retry_on_blocked, max_request_retries=0, - # Without this, a 429 retires the session and the rotation retries walk the backoff up to `max_delay`. max_session_rotations=0, ) + crawler.failed_request_handler(failed_request_handler) + + with caplog.at_level(logging.ERROR), pytest.raises(PersistentRateLimitError, match='Giving up: '): + await crawler.run([str(server_url / 'status/429')]) - await crawler.run([str(server_url / 'status/429')]) + assert any(record.levelno == logging.ERROR and record.exc_info for record in caplog.records) + assert await throttler.is_finished() is False + failed_request_handler.assert_not_called() - assert throttler._is_domain_throttled(domain) + # A failed run keeps the queue, and the stall clock still runs, so a re-run gives up again. + with pytest.raises(PersistentRateLimitError): + await crawler.run() diff --git a/tests/unit/crawlers/_playwright/test_playwright_crawler.py b/tests/unit/crawlers/_playwright/test_playwright_crawler.py index cea2e56d4b..7323811ccc 100644 --- a/tests/unit/crawlers/_playwright/test_playwright_crawler.py +++ b/tests/unit/crawlers/_playwright/test_playwright_crawler.py @@ -38,6 +38,7 @@ from crawlee.fingerprint_suite._header_generator import fingerprint_browser_type_from_playwright_browser_type from crawlee.http_clients import ImpitHttpClient from crawlee.proxy_configuration import ProxyConfiguration +from crawlee.request_loaders import ThrottlingRequestManager from crawlee.sessions import Session, SessionPool from crawlee.statistics import Statistics from crawlee.statistics._error_snapshotter import ErrorSnapshotter @@ -1320,6 +1321,47 @@ async def failed_handler(context: BasicCrawlingContext | PlaywrightCrawlingConte assert set(failed_handler_calls) <= {HELLO_WORLD.decode()} +async def test_throttled_429_is_deferred() -> None: + """A throttled 429 is retried later without spending a retry, a session rotation or session reputation.""" + url = 'https://throttled.placeholder.com/page' + throttler = ThrottlingRequestManager( + await RequestQueue.open(), + domains=['throttled.placeholder.com'], + request_manager_opener=RequestQueue.open, + base_delay=timedelta(milliseconds=50), + max_delay=timedelta(milliseconds=100), + ) + failed_request_handler = AsyncMock() + crawler = PlaywrightCrawler( + request_manager=throttler, + session_pool=SessionPool(max_pool_size=1), + max_request_retries=0, + max_session_rotations=0, + ) + crawler.failed_request_handler(failed_request_handler) + statuses = [429, 200] + handled: list[PlaywrightCrawlingContext] = [] + + @crawler.pre_navigation_hook + async def fulfill_with_next_status(context: PlaywrightPreNavCrawlingContext) -> None: + status = statuses.pop(0) + await context.page.route(url, lambda route: route.fulfill(status=status, body='')) + + @crawler.router.default_handler + async def handler(context: PlaywrightCrawlingContext) -> None: + handled.append(context) + + await crawler.run([url]) + + assert [context.response.status for context in handled] == [200] + failed_request_handler.assert_not_called() + assert handled[0].request.retry_count == 0 + assert (handled[0].request.session_rotation_count or 0) == 0 + session = handled[0].session + assert session is not None + assert session.error_score == 0 + + def test_import_error_handled() -> None: blocked = { mod_name: None for mod_name in sys.modules if mod_name == 'playwright' or mod_name.startswith('playwright.') diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index fbc899b5a1..7e461fa269 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -179,6 +179,23 @@ async def test_domain_matching_is_case_insensitive( assert manager._is_domain_throttled('example.com') +@pytest.mark.parametrize( + ('url', 'expected'), + [ + pytest.param(f'https://{THROTTLED_DOMAIN}/page2', True, id='rate-limited'), + pytest.param(f'https://{NON_THROTTLED_DOMAIN}/page1', False, id='unconfigured'), + ], +) +async def test_is_throttled(manager: ThrottlingRequestManager[RequestQueue], url: str, *, expected: bool) -> None: + """Only a configured domain in a cooldown is reported as throttled.""" + assert manager.is_throttled(f'https://{THROTTLED_DOMAIN}/page1') is False + + manager.record_domain_delay(f'https://{THROTTLED_DOMAIN}/page1') + manager.record_domain_delay(f'https://{NON_THROTTLED_DOMAIN}/page1') + + assert manager.is_throttled(url) is expected + + @pytest.mark.parametrize( ('configured', 'url'), [ @@ -1069,6 +1086,374 @@ async def test_reclaim_routes_to_sub_manager_after_restart(fs_service_locator: S assert await restarted.inner.is_empty() +# ── Stall Detection Tests ───────────────────────────── + + +def _stall(manager: ThrottlingRequestManager[Any], clock: MagicMock, url: str) -> None: + """Rate-limit `url` twice, a full stall window apart, so its domain counts as stalled.""" + manager.record_domain_delay(url) + clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) + manager.record_domain_delay(url) + + +async def test_stall_reported_after_window(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A domain rate-limiting every request for longer than the window is reported as stalled.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + manager.record_domain_delay(url) + assert await manager.get_stall_reason() is None + + clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) + manager.record_domain_delay(url) + reason = await manager.get_stall_reason() + + assert reason is not None + assert f'"{THROTTLED_DOMAIN}" (901s)' in reason + assert '`max_domain_stall` (900s)' in reason + + +async def test_suppressed_429_extends_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A 429 inside an active backoff still counts as the domain turning us away.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + state = manager._domain_states[THROTTLED_DOMAIN] + + with _frozen_clock() as clock: + manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + clock.now.return_value += timedelta(seconds=30) + manager.record_domain_delay(url) + + assert state.consecutive_429_count == 1 + assert state.rate_limited_since == CLOCK_START + assert state.last_rate_limited_at == CLOCK_START + timedelta(seconds=30) + + +async def test_first_429_after_idle_not_stalled(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Idle time before the first 429 doesn't count towards the stall window.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + clock.now.return_value += manager._max_domain_stall * 2 + manager.record_domain_delay(url) + assert await manager.get_stall_reason() is None + + +async def test_old_rate_limit_not_stalled(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A domain that stopped rate-limiting a window ago is waited out, not stalled.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + manager.record_domain_delay(url) + clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) + assert await manager.get_stall_reason() is None + + +@pytest.mark.parametrize( + 'from_inner', + [ + pytest.param(False, id='sub-manager'), + pytest.param(True, id='inner'), + ], +) +async def test_handled_request_clears_stall( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, + *, + from_inner: bool, +) -> None: + """Marking a request of the domain as handled resets its stall clock, whoever holds the request.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + + with _frozen_clock() as clock: + await (inner_queue if from_inner else manager).add_request(url) + request = await manager.fetch_next_request() + assert request is not None + await manager.add_request(f'https://{THROTTLED_DOMAIN}/page2') + + _stall(manager, clock, url) + assert await manager.get_stall_reason() is not None + + await manager.mark_request_as_handled(request) + assert await manager.get_stall_reason() is None + + +async def test_reclaim_keeps_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Reclaiming a request doesn't reset the stall clock.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + request = await manager.fetch_next_request() + assert request is not None + _stall(manager, clock, url) + + await manager.reclaim_request(request) + assert await manager.get_stall_reason() is not None + + +async def test_record_success_keeps_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """`record_success` doesn't reset the stall clock.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + _stall(manager, clock, url) + manager.record_success(url) + assert await manager.get_stall_reason() is not None + + +@pytest.mark.parametrize( + 'ready_url', + [ + pytest.param(f'https://{NON_THROTTLED_DOMAIN}/page1', id='inner'), + pytest.param(f'https://{SECOND_THROTTLED_DOMAIN}/page1', id='other-domain'), + ], +) +async def test_ready_work_masks_stall( + two_domain_manager: ThrottlingRequestManager[RequestQueue], ready_url: str +) -> None: + """Dispatchable work anywhere else keeps a stalled domain from ending the crawl.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await two_domain_manager.add_request(url) + await two_domain_manager.add_request(ready_url) + + with _frozen_clock() as clock: + _stall(two_domain_manager, clock, url) + assert await two_domain_manager.get_stall_reason() is None + + +async def test_waiting_domain_masks_stall(two_domain_manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A domain waiting out a cooldown with work left keeps a stalled domain from ending the crawl.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + waiting_url = f'https://{SECOND_THROTTLED_DOMAIN}/page1' + await two_domain_manager.add_request(url) + await two_domain_manager.add_request(waiting_url) + + with _frozen_clock() as clock: + _stall(two_domain_manager, clock, url) + two_domain_manager.record_domain_delay(waiting_url, retry_after=timedelta(seconds=60)) + assert await two_domain_manager.get_stall_reason() is None + + +async def test_empty_waiting_domain_does_not_mask_stall( + two_domain_manager: ThrottlingRequestManager[RequestQueue], +) -> None: + """A domain in a cooldown with no work left doesn't keep a stalled domain from being reported.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await two_domain_manager.add_request(url) + + with _frozen_clock() as clock: + _stall(two_domain_manager, clock, url) + two_domain_manager.record_domain_delay(f'https://{SECOND_THROTTLED_DOMAIN}/page1') + reason = await two_domain_manager.get_stall_reason() + + assert reason is not None + assert THROTTLED_DOMAIN in reason + assert SECOND_THROTTLED_DOMAIN not in reason + + +async def test_lapsed_backoff_does_not_mask_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A stalled domain between two 429s isn't progress just because its backoff has run out.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + _stall(manager, clock, url) + clock.now.return_value = manager._domain_states[THROTTLED_DOMAIN].throttled_until + timedelta(seconds=1) + assert await manager.get_stall_reason() is not None + + +async def test_empty_domain_not_stalled(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A domain with no requests left is finished, not stalled.""" + with _frozen_clock() as clock: + _stall(manager, clock, f'https://{THROTTLED_DOMAIN}/page1') + assert await manager.get_stall_reason() is None + + +async def test_is_empty_skips_stall_candidate(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A stalled domain doesn't count for `is_empty`, but its request can still be fetched.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + + with _frozen_clock() as clock: + _stall(manager, clock, url) + clock.now.return_value = manager._domain_states[THROTTLED_DOMAIN].throttled_until + timedelta(seconds=1) + + assert await manager.is_empty() is True + request = await manager.fetch_next_request() + + assert request is not None + assert request.url == url + + +async def test_purge_resets_stall_and_migrations( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """A purge clears the stall clock and the migration count.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + state = manager._domain_states[THROTTLED_DOMAIN] + + with _frozen_clock() as clock: + _stall(manager, clock, url) + assert await manager.fetch_next_request() is None + + await manager.purge() + + assert state.rate_limited_since is None + assert state.last_rate_limited_at is None + assert await manager.get_total_count() == 0 + assert await manager.get_handled_count() == 0 + + +@pytest.mark.parametrize( + 'max_domain_stall', + [ + pytest.param(timedelta(0), id='zero'), + pytest.param(timedelta(seconds=-1), id='negative'), + ], +) +async def test_max_domain_stall_must_be_positive( + inner_queue: RequestQueue, + service_locator: ServiceLocator, + max_domain_stall: timedelta, +) -> None: + """A non-positive `max_domain_stall` is rejected.""" + with pytest.raises(ValueError, match='max_domain_stall'): + ThrottlingRequestManager( + inner_queue, + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + max_domain_stall=max_domain_stall, + ) + + +# ── Inner Migration Tests ───────────────────────────── + + +async def test_inner_request_migrates_while_throttled( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """An inner request whose domain is in a cooldown moves into the domain's sub-manager instead of being fetched.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + + assert await manager.fetch_next_request() is None + + assert await inner_queue.is_finished() is True + assert await manager._sub_managers[THROTTLED_DOMAIN].get_total_count() == 1 + assert await manager.get_total_count() == 1 + assert await manager.get_handled_count() == 0 + assert await manager.is_finished() is False + + +async def test_inner_request_dispatched_when_not_throttled( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """An inner request whose domain isn't in a cooldown is fetched from inner.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + + request = await manager.fetch_next_request() + + assert request is not None + assert request.url == url + assert await manager._sub_managers[THROTTLED_DOMAIN].get_total_count() == 0 + + +async def test_migrated_copy_is_unhandled( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """The request that lands in the sub-manager isn't marked handled along with the inner one.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + + with _frozen_clock() as clock: + manager.record_domain_delay(url) + assert await manager.fetch_next_request() is None + + clock.now.return_value = manager._domain_states[THROTTLED_DOMAIN].throttled_until + request = await manager.fetch_next_request() + + assert request is not None + assert request.url == url + assert request.handled_at is None + + +async def test_migration_count_is_capped( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """One fetch moves at most `_MAX_INNER_MIGRATIONS` requests out of inner.""" + await inner_queue.add_requests([f'https://{THROTTLED_DOMAIN}/page{i}' for i in range(3)]) + manager.record_domain_delay(f'https://{THROTTLED_DOMAIN}/', retry_after=timedelta(seconds=60)) + + with patch(f'{MANAGER_MODULE}._MAX_INNER_MIGRATIONS', 2): + assert await manager.fetch_next_request() is None + + assert await inner_queue.get_handled_count() == 2 + assert await manager._sub_managers[THROTTLED_DOMAIN].get_total_count() == 2 + + +@pytest.mark.parametrize( + ('add_request_mock', 'expected_error'), + [ + pytest.param(AsyncMock(side_effect=RuntimeError('storage failure')), RuntimeError, id='error'), + pytest.param(AsyncMock(side_effect=asyncio.CancelledError), asyncio.CancelledError, id='cancelled'), + pytest.param(AsyncMock(return_value=None), RuntimeError, id='refused'), + ], +) +async def test_migration_failure_reclaims_request( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, + add_request_mock: AsyncMock, + expected_error: type[BaseException], +) -> None: + """A request whose migration fails goes back to inner before the error is raised.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + await manager._ensure_sub_managers() + + sub_manager = manager._sub_managers[THROTTLED_DOMAIN] + with patch.object(sub_manager, 'add_request', add_request_mock), pytest.raises(expected_error): + await manager.fetch_next_request() + + assert await inner_queue.get_handled_count() == 0 + assert await manager.get_total_count() == 1 + request = await inner_queue.fetch_next_request() + assert request is not None + assert request.url == url + + +async def test_stall_sees_migrated_inner_work( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """Moving a stalled domain's request out of inner lets the stall be reported.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + + with _frozen_clock() as clock: + request = await manager.fetch_next_request() + assert request is not None + _stall(manager, clock, url) + await manager.reclaim_request(request) + assert await manager.get_stall_reason() is None + + assert await manager.fetch_next_request() is None + assert await manager.get_stall_reason() is not None + + # ── Utility Tests ──────────────────────────────────────