diff --git a/docs/guides/request_loaders.mdx b/docs/guides/request_loaders.mdx index 4ca7358901..f67d72a13e 100644 --- a/docs/guides/request_loaders.mdx +++ b/docs/guides/request_loaders.mdx @@ -25,7 +25,7 @@ The [`request_loaders`](https://github.com/apify/crawlee-python/tree/master/src/ The [`request_loaders`](https://github.com/apify/crawlee-python/tree/master/src/crawlee/request_loaders) sub-package introduces the following abstract classes: - `RequestLoader`: The base interface for reading requests in a crawl. -- `RequestManager`: Extends `RequestLoader` with write capabilities. +- `RequestManager`: Extends `RequestLoader` with write capabilities, and with the pacing signals a crawler reports back. - `RequestManagerTandem`: Combines a read-only `RequestLoader` with a writable `RequestManager`. And specific request loader implementations: @@ -71,6 +71,7 @@ class RequestManager { + add_request() + add_requests_batched() + reclaim_request() + + record_pacing_signal() + purge() + drop() } @@ -156,6 +157,21 @@ When using persistence with `SitemapRequestLoader`, make sure to use the context The `RequestManager` extends `RequestLoader` with write capabilities. In addition to reading requests, a request manager can add and reclaim them. This is essential for dynamic crawling projects where new URLs may emerge during the crawl process, or when certain requests fail and need to be retried. For more details, refer to the `RequestManager` API reference. +### Pacing signals + +A request manager decides when each request goes out, so the crawler reports back to it what a site said about the pace it wants. The crawler does so through `RequestManager.record_pacing_signal`, with one of two signals: + +- `RateLimitedSignal`: a site refused a request because the crawler was going too fast. It carries the wait the site asked for, if any. +- `MinIntervalSignal`: a site declared a minimum interval between requests, such as a robots.txt `Crawl-delay`. + +Each signal also has a `scope`, which says how much of the URL space it covers. A manager may apply a signal to a wider scope than it was given, but never to a narrower one, which would leave part of what the signal covers unpaced. A manager that can't honor a scope raises an error. + +To implement `record_pacing_signal` in your own request manager: + +- Return `False` when the manager doesn't pace its requests, as `RequestQueue` does. The crawler then logs that the signal was dropped. New signals may be added in a minor release, so also return `False` for a signal you don't handle. +- Return `True` for a `RateLimitedSignal` only if the URL won't be dispatched again until a positive delay has passed. The crawler defers such a request without spending a retry. +- If the manager wraps another one, forward every signal to it without inspecting it, as `RequestManagerTandem` does. Otherwise a pacer nested inside it never receives the signal. + ## Request manager tandem The `RequestManagerTandem` class allows you to combine the read-only capabilities of a `RequestLoader` (like `RequestList`) with the read-write capabilities of a `RequestManager` (like `RequestQueue`). This is useful for scenarios where you need to load initial requests from a static source (such as a file or database) and dynamically add or retry requests during the crawl. Additionally, it provides deduplication capabilities, ensuring that requests are not processed multiple times. diff --git a/docs/guides/request_throttling.mdx b/docs/guides/request_throttling.mdx index c02ea9cf96..76cadf37b1 100644 --- a/docs/guides/request_throttling.mdx +++ b/docs/guides/request_throttling.mdx @@ -42,6 +42,8 @@ To use request throttling, create a `RequestManager.record_pacing_signal`, which `RequestManagerTandem` forwards to the manager it wraps, so a `ThrottlingRequestManager` also paces its domains as the writable side of a tandem. For details, see [Pacing signals](./request-loaders#pacing-signals). + ## 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. diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 032cff0100..fd7e0fb76c 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -231,6 +231,32 @@ A `None` from `fetch_next_request` no longer means the loader is finished. It on The crawler now reads a stall from the status, so it also sees a `ThrottlingRequestManager` wrapped in a `RequestManagerTandem`. Such a crawl now ends with `PersistentRateLimitError` once a domain stalls, as it already did with a `ThrottlingRequestManager` passed directly. Before, the crawler didn't detect the stall. +## `record_pacing_signal` replaces the `ThrottlingRequestManager` delay methods + +`RequestManager` has a new abstract method, `record_pacing_signal`. The crawler reports HTTP 429 responses and robots.txt `Crawl-delay` directives through it, so a `ThrottlingRequestManager` now receives them wherever it sits, including inside a `RequestManagerTandem`. In v1, the crawler reached the throttler only when it was passed directly. + +`ThrottlingRequestManager.record_domain_delay` and `set_crawl_delay` are replaced by `record_pacing_signal`, which takes a `RateLimitedSignal` or a `MinIntervalSignal`. `is_throttled` and `record_success` are removed without a replacement, since the manager resets the backoff on its own once a domain stops rate-limiting. + +```python +# v1 +throttler.record_domain_delay(url, retry_after=timedelta(seconds=30)) +throttler.set_crawl_delay(url, 5) + +# v2 +from crawlee.request_loaders import MinIntervalSignal, RateLimitedSignal + +throttler.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=30))) +throttler.record_pacing_signal( + MinIntervalSignal(url, interval=timedelta(seconds=5), scope='hostname') +) +``` + +If you have a custom request manager, implement `record_pacing_signal`. Return `False` if the manager doesn't pace requests or doesn't handle the signal's `reason`, and forward every signal if it wraps another manager. For details, see [Pacing signals](../guides/request-loaders#pacing-signals). + +`ThrottlingRequestManager` now rejects a `base_delay` or `max_delay` that isn't positive, so a zero `base_delay` no longer works as a way to honor only `Retry-After`. + +A `RequestThrottledError` you raise yourself now starts the domain's backoff, and the request is deferred if the request manager paces its domain. In v1, it was an ordinary error unless you had called `record_domain_delay` first. A deferred request never fails, so raise the error only when the server has refused the request. In `AdaptivePlaywrightCrawler`, such an error from the static run now defers the request instead of falling back to the browser. + ## Logs changed If you configure loggers by name or match log messages, update them for the following changes. @@ -272,3 +298,7 @@ When a crawl processes no requests because its request manager holds only handle The ERROR message the crawler logs before raising `PersistentRateLimitError` now reads "Giving up the crawl because the request manager can't make progress." instead of "Giving up the crawl because a domain keeps rate-limiting every request." The two DEBUG messages that `RequestQueue.is_finished` logged, "Background add requests tasks are still in progress." and "The request queue is finished.", are removed together with the method. + +The warnings about an HTTP 429 that the request manager doesn't pace are reworded and logged once per domain. The warning `run()` logged at the start about `respect_robots_txt_file` without a `ThrottlingRequestManager` is replaced by a new one, logged once per domain whose robots.txt `Crawl-delay` the request manager doesn't pace. The warning about a `RequestThrottledError` handled as an ordinary error is reworded, and a new ERROR is logged when the request manager fails to record that rate limit. + +The `ThrottlingRequestManager` warning about capping a delay now calls it a "requested wait" delay instead of a "Retry-After header" delay. Its DEBUG messages about a 429 ignored during an active backoff and about resetting a domain's rate limit state are removed, and its DEBUG message about a domain's crawl-delay shows the delay with one decimal place. diff --git a/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py b/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py index 75a90dfbce..0865d7173f 100644 --- a/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py +++ b/src/crawlee/crawlers/_adaptive_playwright/_adaptive_playwright_crawler.py @@ -401,7 +401,7 @@ async def _run_request_handler(self, context: BasicCrawlingContext) -> None: if ( isinstance(static_run.exception, (ContextPipelineInitializationError, RequestHandlerError)) and isinstance(static_run.exception.wrapped_exception, RequestThrottledError) - and self._is_held_back(context.request) + and await self._report_rate_limit(context.request) ): raise static_run.exception if static_run.exception: diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index aeb8b9e158..7028163da2 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -67,7 +67,7 @@ ) from crawlee.events._types import Event, EventCrawlerStatusData from crawlee.http_clients import ImpitHttpClient -from crawlee.request_loaders import ThrottlingRequestManager +from crawlee.request_loaders import MinIntervalSignal, RateLimitedSignal from crawlee.router import Router from crawlee.sessions import SessionPool from crawlee.statistics import Statistics, StatisticsState @@ -740,14 +740,6 @@ async def run(self, requests: Sequence[str | Request] | None = None) -> FinalSta self._handled_request_this_run = False try: - if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager): - self._logger.warning( - 'The `respect_robots_txt_file` option is enabled, but the crawler is not using ' - '`ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To ' - 'enable crawl-delay support, configure the crawler to use `ThrottlingRequestManager` as the ' - 'request manager.' - ) - if self._has_finished_before: await self._statistics.reset() @@ -1606,7 +1598,9 @@ async def __run_task_function(self, task_id: str) -> None: # noqa: ARG002 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): + if isinstance(primary_error.wrapped_exception, RequestThrottledError) and await self._report_rate_limit( + request + ): await self._defer_throttled_request(request, primary_error.wrapped_exception) return @@ -1672,7 +1666,7 @@ async def __run_task_function(self, task_id: str) -> None: # noqa: ARG002 except ContextPipelineInitializationError as initialization_error: wrapped = initialization_error.wrapped_exception - if isinstance(wrapped, RequestThrottledError) and self._is_held_back(request): + if isinstance(wrapped, RequestThrottledError) and await self._report_rate_limit(request): await self._defer_throttled_request(request, wrapped) return @@ -1697,19 +1691,30 @@ async def __run_task_function(self, task_id: str) -> None: # noqa: ARG002 except Exception: 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. + async def _report_rate_limit(self, request: Request) -> bool: + """Report a rate limit for the request's URL and check whether the request manager took responsibility for it. - Logs a one-time warning when it doesn't, since the `RequestThrottledError` is then handled as an ordinary error. + Only then is a `RequestThrottledError` deferred, since otherwise the request would be dispatched again at once. + Logs a one-time warning when the manager refuses, since the error is then handled as an ordinary error. """ - manager = self._request_manager - if isinstance(manager, ThrottlingRequestManager) and manager.is_throttled(request.url): + request_manager = await self.get_request_manager() + # An error raised from here would bypass the request error handling and end the crawl. + try: + accepted = request_manager.record_pacing_signal(RateLimitedSignal(request.url)) + except Exception: + self._logger.exception( + f'The request manager failed to record a rate limit for {request.url}, so the ' + '`RequestThrottledError` is handled as an ordinary error.' + ) + return False + + if accepted: 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.', + f"`RequestThrottledError` was raised for {request.url}, but the crawler's request manager does not pace " + "its domain, so it's handled as an ordinary error. Pass a `ThrottlingRequestManager` covering the domain " + 'as `request_manager` to retry such requests once the domain stops rate-limiting.', key='throttled_error_without_backoff', level=logging.WARNING, ) @@ -1779,37 +1784,28 @@ def _record_rate_limit_status_code( 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. + True if the request manager took responsibility for the 429, so the request should be deferred. """ if status_code != HTTPStatus.TOO_MANY_REQUESTS: return False - if not isinstance(self._request_manager, ThrottlingRequestManager): - self._logger_once.log( - 'Received an HTTP 429 (Too Many Requests) response, but the crawler is not using ' - '`ThrottlingRequestManager`. Per-domain backoff and `Retry-After` headers will not be honored. ' - 'To enable per-domain rate limiting, configure the crawler to use `ThrottlingRequestManager` ' - 'as the request manager.', - key='no_throttling_manager_on_429', - level=logging.WARNING, - ) - 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): - domain = (URL(request_url).host or '').lower().removesuffix('.') - if domain: - self._logger_once.log( - f'Received an HTTP 429 (Too Many Requests) response from domain "{domain}", but it is ' - f'not in the `ThrottlingRequestManager.domains` list. Per-domain backoff will not be ' - f'applied for this domain. Add it to `domains=` to enable throttling.', - key=f'unconfigured_throttle_domain:{domain}', - level=logging.WARNING, - ) - return False + # Not `get_request_manager()`: the sub crawlers of `AdaptivePlaywrightCrawler` run this without a manager of + # their own, and opening one would leave an unused queue behind. + manager = self._request_manager + signal = RateLimitedSignal(request_url, wait=parse_retry_after_header(retry_after_header)) + if manager is not None and manager.record_pacing_signal(signal): + return True - return True + domain = URL(request_url).host or request_url + self._logger_once.log( + f'"{domain}" responded with HTTP 429 (Too Many Requests), but the crawler\'s request manager does not pace ' + 'that domain, so the response is handled like any other, with no per-domain delay. Pass a ' + f'`ThrottlingRequestManager` covering "{domain}" as `request_manager` to honor `Retry-After` and apply ' + 'exponential backoff instead.', + key=f'rate_limit_not_paced:{domain}', + level=logging.WARNING, + ) + return False 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. @@ -1854,13 +1850,30 @@ async def _is_allowed_based_on_robots_txt_file(self, url: str) -> bool: if not robots_txt_file: return True - if isinstance(self._request_manager, ThrottlingRequestManager): - crawl_delay = robots_txt_file.get_crawl_delay() - if crawl_delay is not None: - self._request_manager.set_crawl_delay(url, crawl_delay) + crawl_delay = robots_txt_file.get_crawl_delay() + if crawl_delay is not None: + await self._apply_crawl_delay(url, crawl_delay) return robots_txt_file.is_allowed(url) + async def _apply_crawl_delay(self, url: str, delay_seconds: int) -> None: + """Hand a robots.txt crawl-delay to the request manager, warning if it won't be honored.""" + # robots.txt is per origin. `hostname` is the closest pacing scope and errs wide, as HTTP and HTTPS to one host + # share a clock. + signal = MinIntervalSignal(url, interval=timedelta(seconds=delay_seconds), scope='hostname') + request_manager = await self.get_request_manager() + if request_manager.record_pacing_signal(signal): + return + + domain = URL(url).host or url + self._logger_once.log( + f'robots.txt for "{domain}" defines a crawl-delay of {delay_seconds}s, but the crawler\'s request manager ' + 'does not pace that domain, so its requests will not be paced. Pass a `ThrottlingRequestManager` covering ' + f'"{domain}" as `request_manager`.', + key=f'crawl_delay_not_paced:{domain}', + level=logging.WARNING, + ) + async def _get_robots_txt_file_for_url(self, url: str) -> RobotsTxtFile | None: """Get the RobotsTxtFile for a given URL. diff --git a/src/crawlee/errors.py b/src/crawlee/errors.py index ed1c23933b..09ef5f7ae8 100644 --- a/src/crawlee/errors.py +++ b/src/crawlee/errors.py @@ -124,8 +124,10 @@ class RequestCollisionError(Exception): 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. + The request is retried later without spending a retry or marking the session as bad. If you raise it yourself, the + crawler reports a rate limit to its request manager and defers the request only if the manager paces its domain, + and handles it as an ordinary error otherwise. Raise it only when the server has refused the request, since a + deferred request never fails. """ def __init__(self, message: str = 'Request is being retried later because its domain is rate-limiting us') -> None: diff --git a/src/crawlee/request_loaders/__init__.py b/src/crawlee/request_loaders/__init__.py index 3237ea5b72..0a6b5b314a 100644 --- a/src/crawlee/request_loaders/__init__.py +++ b/src/crawlee/request_loaders/__init__.py @@ -6,6 +6,10 @@ from ._throttling_request_manager import ThrottlingRequestManager from ._types import ( FinishedStatus, + MinIntervalSignal, + PacingScope, + PacingSignal, + RateLimitedSignal, ReadyStatus, RequestLoaderStatus, RequestSourceStatus, @@ -15,6 +19,10 @@ __all__ = [ 'FinishedStatus', + 'MinIntervalSignal', + 'PacingScope', + 'PacingSignal', + 'RateLimitedSignal', 'ReadyStatus', 'RequestList', 'RequestLoader', diff --git a/src/crawlee/request_loaders/_request_manager.py b/src/crawlee/request_loaders/_request_manager.py index 8ecb181a37..4646f09112 100644 --- a/src/crawlee/request_loaders/_request_manager.py +++ b/src/crawlee/request_loaders/_request_manager.py @@ -13,6 +13,8 @@ from crawlee._request import Request + from ._types import PacingSignal + @docs_group('Request loaders') class RequestManager(RequestLoader, ABC): @@ -79,3 +81,20 @@ async def reclaim_request(self, request: Request, *, forefront: bool = False) -> It is possible to modify the request data by supplying an updated request as a parameter. """ + + @abstractmethod + def record_pacing_signal(self, signal: PacingSignal) -> bool: + """Record a signal about the pace at which requests may go out. + + A manager that doesn't pace its requests, or doesn't handle the signal's `reason`, returns `False`. More signals + may be added in a minor release. A manager wrapping another forwards every signal to it without inspecting it, + so a pacer nested in a composition still receives it. A manager may apply a signal to a wider scope than the + one given, but never to a narrower one, and raises for a scope it can't honor. + + Args: + signal: The signal to record. + + Returns: + `True` if anything in the composition took responsibility for the signal. For a `RateLimitedSignal`, the + URL then isn't dispatched again until a positive delay has passed. + """ diff --git a/src/crawlee/request_loaders/_request_manager_tandem.py b/src/crawlee/request_loaders/_request_manager_tandem.py index 90d55f94c4..6f1f4c5c87 100644 --- a/src/crawlee/request_loaders/_request_manager_tandem.py +++ b/src/crawlee/request_loaders/_request_manager_tandem.py @@ -18,7 +18,7 @@ from crawlee.request_loaders import RequestLoader from crawlee.storage_clients.models import ProcessedRequest - from ._types import RequestSourceStatus + from ._types import PacingSignal, RequestSourceStatus logger = getLogger(__name__) @@ -109,6 +109,11 @@ async def fetch_next_request(self) -> Request | None: async def reclaim_request(self, request: Request, *, forefront: bool = False) -> None: await self._read_write_manager.reclaim_request(request, forefront=forefront) + @override + def record_pacing_signal(self, signal: PacingSignal) -> bool: + """Forward the signal to the manager. The loader side is read-only and dispatches nothing of its own.""" + return self._read_write_manager.record_pacing_signal(signal) + @override async def mark_request_as_handled(self, request: Request) -> None: await self._read_write_manager.mark_request_as_handled(request) diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index 9d202a7999..0cbf9e2aba 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -26,7 +26,7 @@ from crawlee.storage_clients import StorageClient from crawlee.storage_clients.models import ProcessedRequest - from ._types import RequestSourceStatus + from ._types import PacingScope, PacingSignal, RequestSourceStatus logger = getLogger(__name__) @@ -61,14 +61,13 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): right now, it returns `None` rather than waiting, so the caller's task slot is released. `check_readiness()` then reports the crawl as waiting until the earliest cooldown ends, so it idles instead of ending early. - Delay sources: - - HTTP 429 responses (via `record_domain_delay`) - - robots.txt crawl-delay directives (via `set_crawl_delay`) + The crawler reports HTTP 429 responses and robots.txt crawl-delay directives through `record_pacing_signal`, which + wrapping managers such as `RequestManagerTandem` forward, so the throttler paces its domains wherever it sits. The class is generic over the wrapped manager type. The first asynchronous operation opens one sub-manager per configured domain through `request_manager_opener`, so all of them share the subclass and backing store of `inner`; - the synchronous delay methods never open anything. The opener must accept `alias`, `storage_client`, and - `configuration` keyword arguments (as `RequestQueue.open` does) and return the same concrete subclass as `inner`. + `record_pacing_signal` never opens anything. The opener must accept `alias`, `storage_client`, and `configuration` + keyword arguments (as `RequestQueue.open` does) and return the same concrete subclass as `inner`. Requests a previous run left in a persistent store become visible again at open. The default `purge_on_start=True` empties them; `purge_on_start=False` resumes them. Named stores are exempt from that purge and aliased ones are not, @@ -118,16 +117,23 @@ def __init__( concrete subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`). service_locator: Service locator for creating sub-managers. If not provided, defaults to the global service 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. + base_delay: Initial delay after the first 429 response from a domain. Must be positive. + max_delay: Maximum delay between requests to a rate-limited domain. Must be positive. 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, or if - `max_domain_stall` is not positive. + `base_delay`, `max_delay` or `max_domain_stall` is not positive. """ + # A zero delay would let the crawler defer a rate-limited request and dispatch it again at once. + if base_delay <= timedelta(0): + raise ValueError(f'base_delay must be positive, got {base_delay}.') + + if max_delay <= timedelta(0): + raise ValueError(f'max_delay must be positive, got {max_delay}.') + if max_domain_stall <= timedelta(0): raise ValueError(f'max_domain_stall must be positive, got {max_domain_stall}.') @@ -382,22 +388,49 @@ async def check_readiness(self) -> RequestSourceStatus: return probed - 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. Every call, including a manual one or a 429 inside a - burst, also starts or extends the domain's stall clock. + @override + def record_pacing_signal(self, signal: PacingSignal) -> bool: + """Record a pacing signal for the domain of its URL. - Args: - url: The URL that received a 429 response. - retry_after: Optional delay from the `Retry-After` header. If it describes a positive delay, it takes - priority over the calculated exponential backoff. + A `RateLimitedSignal` backs the domain off for the signal's `wait` if it's positive, and exponentially from + `base_delay` otherwise. Only the first refusal of a burst advances the backoff. A `MinIntervalSignal` becomes + the domain's crawl-delay. The first one wins, so a robots.txt re-fetch can't change the dispatch cadence + mid-crawl. Returns: - 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. + `True` if the URL's domain is in `domains`, so the manager paces it by the signal. `False` if the domain + isn't in `domains` or the signal's `reason` isn't one this manager handles, in which case the call is a + no-op. + + Raises: + ValueError: If the signal's `scope` is wider than a hostname, or one this manager doesn't know. + """ + if signal.reason == 'rate_limited': + self._check_scope(signal.scope) + return self._record_rate_limit(signal.url, signal.wait) + + if signal.reason == 'min_interval': + self._check_scope(signal.scope) + return self._record_declared_interval(signal.url, signal.interval) + + return False + + @staticmethod + def _check_scope(scope: PacingScope | None) -> None: + """Raise unless a signal scoped to `scope` can be honored as declared or wider.""" + # No scope means the reporter can't tell how far the signal reaches, so the hostname grouping is as good as any. + if scope is None or scope == 'hostname': + return + + raise ValueError( + f'Cannot honor a pacing signal scoped to "{scope}": this manager paces by hostname, so it would leave part ' + f'of what the signal covers unpaced. Pace by "{scope}" with a request manager that groups requests by it.' + ) + + def _record_rate_limit(self, url: str, wait: timedelta | None) -> bool: + """Put the URL's domain into a backoff after a refusal, doubling from `base_delay` while it keeps refusing. + + Every call, including one inside a burst, also starts or extends the domain's stall clock. """ state = self._get_domain_state(url) if state is None: @@ -413,10 +446,6 @@ def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None) # 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. @@ -425,10 +454,10 @@ def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None) 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' + # A non-positive wait is no delay at all, so fall back to the backoff and let it engage. + if wait is not None and wait > timedelta(0): + delay = wait + source = 'requested wait' else: delay = self._base_delay * 2 ** min(state.consecutive_429_count - 1, _MAX_BACKOFF_EXPONENT) source = 'exponential backoff' @@ -449,44 +478,17 @@ def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None) ) return True - 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. It doesn't affect the stall clock. - - Args: - url: The URL that received a successful response. - """ + def _record_declared_interval(self, url: str, interval: timedelta) -> bool: + """Record a declared minimum interval as the crawl-delay of the URL's domain, unless it already has one.""" state = self._get_domain_state(url) - if state is not None and state.consecutive_429_count > 0: - 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. - - The delay is locked once set so robots.txt re-fetches (e.g. after LRU eviction) can't change the in-flight - dispatch cadence and cause oscillation mid-crawl. Subsequent calls for the same domain are no-ops. + if state is None: + return False - Args: - url: A URL from the domain to throttle. - delay_seconds: The crawl-delay value in seconds. - """ - state = self._get_domain_state(url) - if state is None or state.crawl_delay is not None: - return + if state.crawl_delay is None: + state.crawl_delay = interval + logger.debug(f'Set crawl-delay for domain "{state.domain}" to {interval.total_seconds():.1f}s') - state.crawl_delay = timedelta(seconds=delay_seconds) - logger.debug(f'Set crawl-delay for domain "{state.domain}" to {delay_seconds}s') + return True @staticmethod def _normalize_domain(hostname: str) -> str: @@ -602,13 +604,6 @@ async def _migrate_to_sub_manager(self, request: Request) -> None: 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) - if state is None: - return False - return datetime.now(UTC) < state.throttled_until - def _fetchable_domains(self) -> list[str]: """Return the configured domains that are not in a cooldown right now, longest-overdue first.""" now = datetime.now(UTC) diff --git a/src/crawlee/request_loaders/_types.py b/src/crawlee/request_loaders/_types.py index c03b6cc2ec..543a5a9c0d 100644 --- a/src/crawlee/request_loaders/_types.py +++ b/src/crawlee/request_loaders/_types.py @@ -1,12 +1,12 @@ from __future__ import annotations -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import TYPE_CHECKING, ClassVar, Literal from crawlee._utils.docs import docs_group if TYPE_CHECKING: - from datetime import datetime + from datetime import datetime, timedelta @docs_group('Request loaders') @@ -64,3 +64,54 @@ class FinishedStatus: RequestSourceStatus = ReadyStatus | WaitingStatus | StalledStatus | FinishedStatus """Readiness of a request source, as reported by `RequestLoader.check_readiness`.""" + + +PacingScope = str +"""How much of the URL space a pacing signal covers. + +The vocabulary is open, so a manager that paces by something else can be reported to in its own terms. Crawlee itself +reports `'hostname'`, and `'registrable_domain'` names a whole site with its subdomains. `ThrottlingRequestManager` +accepts only `'hostname'`. +""" + + +@docs_group('Request loaders') +@dataclass(frozen=True, kw_only=True) +class RateLimitedSignal: + """The source turned a request away because the crawler was going too fast, such as with an HTTP 429.""" + + reason: ClassVar[Literal['rate_limited']] = 'rate_limited' + """The signal name, so a check doesn't need to import the class.""" + + url: str = field(kw_only=False) + """The URL that was turned away.""" + + wait: timedelta | None = None + """How long the source asked to wait before trying again, if it said.""" + + scope: PacingScope | None = None + """How far the refusal reaches, if the reporter can tell. `None` lets the manager apply it however it groups + requests. + """ + + +@docs_group('Request loaders') +@dataclass(frozen=True, kw_only=True) +class MinIntervalSignal: + """The source declared a minimum interval between requests, such as a robots.txt `Crawl-delay`.""" + + reason: ClassVar[Literal['min_interval']] = 'min_interval' + """The signal name, so a check doesn't need to import the class.""" + + url: str = field(kw_only=False) + """A URL of the source that declared the interval.""" + + interval: timedelta + """The declared minimum interval between two requests to the source.""" + + scope: PacingScope + """How much of the URL space the interval applies to.""" + + +PacingSignal = RateLimitedSignal | MinIntervalSignal +"""A signal about the pace at which requests may go out, as reported by `RequestManager.record_pacing_signal`.""" diff --git a/src/crawlee/storages/_request_queue.py b/src/crawlee/storages/_request_queue.py index cb13f79a5e..ba15cce560 100644 --- a/src/crawlee/storages/_request_queue.py +++ b/src/crawlee/storages/_request_queue.py @@ -21,7 +21,7 @@ from crawlee import Request from crawlee.configuration import Configuration - from crawlee.request_loaders import RequestLoaderStatus + from crawlee.request_loaders import PacingSignal, RequestLoaderStatus from crawlee.storage_clients import StorageClient from crawlee.storage_clients._base import RequestQueueClient from crawlee.storage_clients.models import ProcessedRequest, RequestQueueMetadata @@ -293,6 +293,14 @@ async def reclaim_request( """ return await self._client.reclaim_request(request, forefront=forefront) + @override + def record_pacing_signal(self, signal: PacingSignal) -> bool: + """Refuse the signal, since a queue hands requests out as fast as they're asked for. + + Pacing is the job of a manager wrapped around the queue, such as `ThrottlingRequestManager`. + """ + return False + @override async def check_readiness(self) -> RequestLoaderStatus: """Report whether the queue has a request to hand over, is waiting for one, or is done. 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 6e2b212123..77321d3f80 100644 --- a/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py +++ b/tests/unit/crawlers/_adaptive_playwright/test_adaptive_playwright_crawler.py @@ -1012,6 +1012,7 @@ async def handler(context: AdaptivePlaywrightCrawlingContext) -> None: assert http_client.crawl.await_count == 2 assert len(handled) == 1 assert handled[0].request.retry_count == 0 + assert throttler._domain_states['throttled.placeholder.com'].consecutive_429_count == 1 async def test_static_handler_raised_throttled_error_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None: @@ -1040,16 +1041,16 @@ async def track_browser_navigation(context: AdaptivePlaywrightPreNavCrawlingCont browser_navigations() @crawler.router.default_handler - async def handler(context: AdaptivePlaywrightCrawlingContext) -> None: + async def handler(_context: AdaptivePlaywrightCrawlingContext) -> None: handler_calls() if handler_calls.call_count == 1: - throttler.record_domain_delay(context.request.url) raise RequestThrottledError await crawler.run([url]) browser_navigations.assert_not_called() assert handler_calls.call_count == 2 + assert throttler._domain_states['throttled.placeholder.com'].consecutive_429_count == 1 async def test_static_throttled_error_without_backoff_falls_back_to_browser(test_urls: list[str]) -> None: diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 3b3e2d5fa6..da1f39b9a5 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -17,7 +17,7 @@ from itertools import product from multiprocessing import get_context from typing import TYPE_CHECKING, Any, Literal, cast -from unittest.mock import ANY, AsyncMock, Mock, call, patch +from unittest.mock import ANY, AsyncMock, Mock, call, create_autospec, patch import pytest @@ -40,7 +40,10 @@ from crawlee.events import Event, EventCrawlerStatusData, LocalEventManager from crawlee.http_clients import HttpClient from crawlee.request_loaders import ( + MinIntervalSignal, + RateLimitedSignal, RequestList, + RequestManager, RequestManagerTandem, RequestSourceStatus, StalledStatus, @@ -2905,44 +2908,64 @@ async def handler(context: BasicCrawlingContext) -> None: assert global_event_manager.active is False -async def test_warn_no_throttling_manager_once_on_429(caplog: pytest.LogCaptureFixture) -> None: - """A 429 from a crawler without ThrottlingRequestManager logs a recommendation, only once per instance.""" - crawler = BasicCrawler(configure_logging=False) - with caplog.at_level(logging.WARNING, logger='crawlee'): - crawler._record_rate_limit_status_code(429, request_url='https://a.test/') - crawler._record_rate_limit_status_code(429, request_url='https://b.test/') - - matching = [ - r for r in caplog.records if 'ThrottlingRequestManager' in r.getMessage() and 'HTTP 429' in r.getMessage() - ] - assert len(matching) == 1 - - -async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest.LogCaptureFixture) -> None: - """A 429 from a domain not in `ThrottlingRequestManager.domains` logs once per domain; repeats are suppressed.""" - inner = await RequestQueue.open(name='test-inner-warn', storage_client=MemoryStorageClient()) - throttler = ThrottlingRequestManager( - inner, - domains=['only-this.test'], - request_manager_opener=RequestQueue.open, - ) - crawler = BasicCrawler(configure_logging=False, request_manager=throttler) +@pytest.mark.parametrize( + 'use_throttler', + [ + pytest.param(False, id='no-manager'), + pytest.param(True, id='unconfigured-domain'), + ], +) +async def test_unpaced_429_warns_once(caplog: pytest.LogCaptureFixture, *, use_throttler: bool) -> None: + """A 429 the request manager doesn't pace logs a warning once per domain.""" + request_manager: ThrottlingRequestManager[RequestQueue] | None = None + if use_throttler: + inner = await RequestQueue.open(name='test-inner-warn', storage_client=MemoryStorageClient()) + request_manager = ThrottlingRequestManager( + inner, domains=['only-this.test'], request_manager_opener=RequestQueue.open + ) + crawler = BasicCrawler(configure_logging=False, request_manager=request_manager) with caplog.at_level(logging.WARNING, logger='crawlee'): crawler._record_rate_limit_status_code(429, request_url='https://A.example.com/page1') crawler._record_rate_limit_status_code(429, request_url='https://a.example.com/page2') crawler._record_rate_limit_status_code(429, request_url='https://other.example.com/page1') - matching = [ - r - for r in caplog.records - if 'not in the' in r.getMessage() and 'ThrottlingRequestManager.domains' in r.getMessage() - ] + matching = [r for r in caplog.records if 'does not pace that domain' in r.getMessage()] assert len(matching) == 2 assert any('a.example.com' in r.getMessage() for r in matching) assert any('other.example.com' in r.getMessage() for r in matching) +async def test_429_reports_requested_wait() -> None: + """A 429 is reported to the request manager with the wait from its `Retry-After` header.""" + request_manager = create_autospec(RequestManager, instance=True, spec_set=True) + request_manager.record_pacing_signal.return_value = True + crawler = BasicCrawler(configure_logging=False, request_manager=request_manager) + url = 'https://a.example.com/page1' + + assert crawler._record_rate_limit_status_code(429, request_url=url, retry_after_header='30') is True + request_manager.record_pacing_signal.assert_called_once_with(RateLimitedSignal(url, wait=timedelta(seconds=30))) + + +async def test_unpaced_crawl_delay_warns_once( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A robots.txt crawl-delay the request manager doesn't pace logs a warning once per domain.""" + crawler = BasicCrawler(configure_logging=False, respect_robots_txt_file=True) + robots_txt_file = Mock(spec=RobotsTxtFile) + robots_txt_file.get_crawl_delay.return_value = 5 + robots_txt_file.is_allowed.return_value = True + monkeypatch.setattr(crawler, '_get_robots_txt_file_for_url', AsyncMock(return_value=robots_txt_file)) + + with caplog.at_level(logging.WARNING, logger='crawlee'): + for url in ['https://a.example.com/page1', 'https://a.example.com/page2', 'https://b.example.com/page1']: + await crawler._is_allowed_based_on_robots_txt_file(url) + + matching = [r for r in caplog.records if 'defines a crawl-delay' in r.getMessage()] + assert len(matching) == 2 + + async def test_crawl_waits_out_backoff() -> None: """Test that a domain in a cooldown reads as waiting, yet its queued requests are still crawled.""" storage_client = MemoryStorageClient() @@ -2968,7 +2991,7 @@ async def test_crawl_waits_out_backoff() -> None: async def handler(context: BasicCrawlingContext) -> None: dispatched_at.append(time.monotonic()) if len(dispatched_at) == 1: - throttler.record_domain_delay(context.request.url, retry_after=timedelta(milliseconds=500)) + throttler.record_pacing_signal(RateLimitedSignal(context.request.url, wait=timedelta(milliseconds=500))) statuses_during_cooldown.append(await throttler.check_readiness()) await crawler.run(['https://throttled.placeholder.com/a', 'https://throttled.placeholder.com/b']) @@ -3008,7 +3031,7 @@ async def open_throttler() -> ThrottlingRequestManager[RequestQueue]: async def test_handler_raised_throttled_error_is_deferred(monkeypatch: pytest.MonkeyPatch) -> None: - """A `RequestThrottledError` from the handler costs neither a retry nor session reputation.""" + """A `RequestThrottledError` from the handler starts a backoff without costing a retry or session reputation.""" throttler = await open_throttler_held_until_reclaim(monkeypatch) failed_request_handler = AsyncMock() crawler = BasicCrawler( @@ -3017,25 +3040,20 @@ async def test_handler_raised_throttled_error_is_deferred(monkeypatch: pytest.Mo 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 + handler = AsyncMock(side_effect=[RequestThrottledError, None]) + crawler.router.default_handler(handler) stats = await crawler.run([THROTTLED_URL]) - assert len(contexts) == 2 + assert handler.await_count == 2 failed_request_handler.assert_not_called() - assert contexts[1].request.retry_count == 0 + context = handler.call_args.args[0] + assert context.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 + assert context.session is not None + assert context.session.error_score == 0 + assert throttler._domain_states['throttled.placeholder.com'].consecutive_429_count == 1 async def test_deferred_request_is_reclaimed_to_the_front(monkeypatch: pytest.MonkeyPatch) -> None: @@ -3050,7 +3068,6 @@ async def test_deferred_request_is_reclaimed_to_the_front(monkeypatch: pytest.Mo async def handler(context: BasicCrawlingContext) -> None: handled.append(context.request) if len(handled) == 1: - throttler.record_domain_delay(context.request.url) raise RequestThrottledError await crawler.run([THROTTLED_URL]) @@ -3062,7 +3079,7 @@ async def handler(context: BasicCrawlingContext) -> None: @pytest.mark.parametrize( 'use_throttler', [ - pytest.param(True, id='no-recorded-delay'), + pytest.param(True, id='unconfigured-domain'), pytest.param(False, id='no-throttler'), ], ) @@ -3071,8 +3088,13 @@ async def test_throttled_error_without_backoff_is_ordinary_error( *, 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 + """A `RequestThrottledError` whose domain nothing paces is retried and failed like any other error.""" + request_manager: ThrottlingRequestManager[RequestQueue] | None = None + if use_throttler: + inner = await RequestQueue.open() + request_manager = ThrottlingRequestManager( + inner, domains=['other.placeholder.com'], request_manager_opener=RequestQueue.open + ) failed_request_handler = AsyncMock() crawler = BasicCrawler(request_manager=request_manager, max_request_retries=1) crawler.failed_request_handler(failed_request_handler) @@ -3085,7 +3107,52 @@ async def test_throttled_error_without_backoff_is_ordinary_error( 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) + assert any('does not pace its domain' in record.message for record in caplog.records) + + +async def test_pacing_failure_is_ordinary_error( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A request manager that fails to record a rate limit doesn't end the crawl, and the error is retried as usual.""" + queue = await RequestQueue.open() + monkeypatch.setattr(queue, 'record_pacing_signal', Mock(side_effect=RuntimeError('pacing failed'))) + failed_request_handler = AsyncMock() + crawler = BasicCrawler(request_manager=queue, max_request_retries=1) + crawler.failed_request_handler(failed_request_handler) + handler = AsyncMock(side_effect=RequestThrottledError) + crawler.router.default_handler(handler) + + stats = await crawler.run([THROTTLED_URL]) + + assert handler.await_count == 2 + failed_request_handler.assert_called_once() + assert stats.requests_failed == 1 + assert any( + record.levelno == logging.ERROR and 'failed to record a rate limit' in record.message + for record in caplog.records + ) + + +@pytest.mark.parametrize( + 'wrap_in_tandem', + [ + pytest.param(False, id='direct'), + pytest.param(True, id='tandem'), + ], +) +async def test_robots_crawl_delay_reaches_throttler(monkeypatch: pytest.MonkeyPatch, *, wrap_in_tandem: bool) -> None: + """A robots.txt crawl-delay reaches the throttler, also when it's wrapped in a `RequestManagerTandem`.""" + throttler = await open_throttler() + request_manager = RequestManagerTandem(RequestList(), throttler) if wrap_in_tandem else throttler + crawler = BasicCrawler(request_manager=request_manager, respect_robots_txt_file=True) + robots_txt_file = Mock(spec=RobotsTxtFile) + robots_txt_file.get_crawl_delay.return_value = 5 + robots_txt_file.is_allowed.return_value = True + monkeypatch.setattr(crawler, '_get_robots_txt_file_for_url', AsyncMock(return_value=robots_txt_file)) + + assert await crawler._is_allowed_based_on_robots_txt_file(THROTTLED_URL) + assert throttler._domain_states['throttled.placeholder.com'].crawl_delay == timedelta(seconds=5) async def test_keep_alive_ignores_stall() -> None: @@ -3096,7 +3163,6 @@ async def test_keep_alive_ignores_stall() -> 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: @@ -3133,7 +3199,6 @@ async def handler(context: BasicCrawlingContext) -> None: if len(visited) > MAX_HANDLER_CALLS: crawler.stop() if context.request.url == THROTTLED_URL: - throttler.record_domain_delay(context.request.url) raise RequestThrottledError # The slow request belongs to the stalled domain, since a request in progress elsewhere would keep the stall # from being reported. @@ -3176,7 +3241,6 @@ async def handler(context: BasicCrawlingContext) -> None: if calls > MAX_HANDLER_CALLS: crawler.stop() if context.request.url == THROTTLED_URL: - throttler.record_domain_delay(context.request.url) raise RequestThrottledError # This request keeps the stall from being reported while in flight, so watch the stall clock. It's of another # domain, so handling it doesn't reset that clock and the stall is live once the limit is reached. @@ -3203,7 +3267,7 @@ async def test_stall_waits_for_paced_domain() -> None: max_delay=timedelta(milliseconds=100), max_domain_stall=MAX_DOMAIN_STALL, ) - throttler.set_crawl_delay(paced_urls[0], 1) + throttler.record_pacing_signal(MinIntervalSignal(paced_urls[0], interval=timedelta(seconds=1), scope='hostname')) crawler = BasicCrawler(request_manager=throttler) visited = list[str]() @@ -3214,7 +3278,6 @@ async def handler(context: BasicCrawlingContext) -> None: 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): @@ -3248,7 +3311,6 @@ async def handler(context: BasicCrawlingContext) -> None: calls += 1 if calls > MAX_HANDLER_CALLS: crawler.stop() - throttler.record_domain_delay(context.request.url) raise RequestThrottledError with pytest.raises(PersistentRateLimitError): @@ -3278,7 +3340,6 @@ async def handler(context: BasicCrawlingContext) -> None: 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]) diff --git a/tests/unit/crawlers/_http/test_http_crawler.py b/tests/unit/crawlers/_http/test_http_crawler.py index 0422b7456a..682345d048 100644 --- a/tests/unit/crawlers/_http/test_http_crawler.py +++ b/tests/unit/crawlers/_http/test_http_crawler.py @@ -13,7 +13,12 @@ from crawlee.crawlers import HttpCrawler from crawlee.errors import PersistentRateLimitError from crawlee.http_clients import HttpClient, HttpCrawlingResult -from crawlee.request_loaders import ThrottlingRequestManager, _throttling_request_manager +from crawlee.request_loaders import ( + RequestList, + RequestManagerTandem, + ThrottlingRequestManager, + _throttling_request_manager, +) from crawlee.sessions import SessionPool from crawlee.statistics import Statistics from crawlee.storages import RequestQueue @@ -740,6 +745,32 @@ async def handler(context: HttpCrawlingContext) -> None: assert session.is_usable +async def test_tandem_429_deferred(monkeypatch: pytest.MonkeyPatch) -> None: + """A 429 reaches a throttler wrapped in a `RequestManagerTandem`, so the request is deferred without a retry.""" + throttler = await open_throttler_held_until_reclaim(monkeypatch) + 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=RequestManagerTandem(RequestList(['https://throttled.placeholder.com/page']), throttler), + max_request_retries=0, + ) + crawler.failed_request_handler(failed_request_handler) + handler = AsyncMock() + crawler.router.default_handler(handler) + + await crawler.run() + + handler.assert_awaited_once() + context = handler.call_args.args[0] + assert context.http_response.status_code == 200 + failed_request_handler.assert_not_called() + assert context.request.retry_count == 0 + assert (context.request.session_rotation_count or 0) == 0 + assert throttler._domain_states['throttled.placeholder.com'].consecutive_429_count == 1 + + @pytest.mark.parametrize( 'retry_on_blocked', [ diff --git a/tests/unit/request_loaders/test_pacing_signal.py b/tests/unit/request_loaders/test_pacing_signal.py new file mode 100644 index 0000000000..af9610383d --- /dev/null +++ b/tests/unit/request_loaders/test_pacing_signal.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +from datetime import timedelta + +import pytest + +from crawlee.request_loaders import MinIntervalSignal, PacingSignal, RateLimitedSignal + + +@pytest.mark.parametrize( + ('signal', 'reason'), + [ + pytest.param(RateLimitedSignal('https://example.com'), 'rate_limited', id='rate-limited'), + pytest.param( + MinIntervalSignal('https://example.com', interval=timedelta(seconds=1), scope='hostname'), + 'min_interval', + id='min-interval', + ), + ], +) +def test_signal_reason(signal: PacingSignal, reason: str) -> None: + """Test that each signal carries its own reason.""" + assert signal.reason == reason diff --git a/tests/unit/storages/test_request_manager_tandem.py b/tests/unit/storages/test_request_manager_tandem.py index 1e7c9bc081..00e2292a0f 100644 --- a/tests/unit/storages/test_request_manager_tandem.py +++ b/tests/unit/storages/test_request_manager_tandem.py @@ -9,6 +9,7 @@ from crawlee import Request from crawlee.request_loaders import ( FinishedStatus, + RateLimitedSignal, ReadyStatus, RequestLoader, RequestManager, @@ -157,3 +158,21 @@ async def test_transfers_only_when_loader_ready() -> None: assert request is not None assert request.url == 'https://a.placeholder.com' loader.fetch_next_request.assert_not_awaited() + + +@pytest.mark.parametrize( + 'accepted', + [ + pytest.param(True, id='accepted'), + pytest.param(False, id='refused'), + ], +) +def test_pacing_signal_forwarded_to_manager(*, accepted: bool) -> None: + """Test that the tandem forwards a pacing signal to its manager and returns the manager's answer.""" + loader = create_autospec(RequestLoader, instance=True, spec_set=True) + manager = create_autospec(RequestManager, instance=True, spec_set=True) + manager.record_pacing_signal.return_value = accepted + signal = RateLimitedSignal('https://a.placeholder.com') + + assert RequestManagerTandem(loader, manager).record_pacing_signal(signal) is accepted + manager.record_pacing_signal.assert_called_once_with(signal) diff --git a/tests/unit/storages/test_request_queue.py b/tests/unit/storages/test_request_queue.py index f3f692006c..9cc9058802 100644 --- a/tests/unit/storages/test_request_queue.py +++ b/tests/unit/storages/test_request_queue.py @@ -9,6 +9,7 @@ from crawlee import Request, service_locator from crawlee.configuration import Configuration +from crawlee.request_loaders import RateLimitedSignal from crawlee.storage_clients import MemoryStorageClient, StorageClient from crawlee.storage_clients.models import AddRequestsResponse, ProcessedRequest, UnprocessedRequest from crawlee.storages import RequestQueue @@ -545,6 +546,11 @@ async def test_reclaim_request_with_forefront(rq: RequestQueue) -> None: assert next_request.url == 'https://example.com/first' +async def test_record_pacing_signal_is_refused(rq: RequestQueue) -> None: + """Test that a plain queue refuses a pacing signal, since it doesn't pace its dispatch.""" + assert rq.record_pacing_signal(RateLimitedSignal('https://example.com')) is False + + async def test_check_readiness(rq: RequestQueue) -> None: """Test that the readiness follows a request from adding to handling.""" assert (await rq.check_readiness()).status == 'finished' diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index a7fc38abd6..bf7a3c670d 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -7,7 +7,7 @@ from datetime import UTC, datetime, timedelta from pathlib import Path from typing import TYPE_CHECKING, Any -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, Mock, patch import pytest @@ -15,7 +15,7 @@ from crawlee._service_locator import ServiceLocator from crawlee._utils.http import parse_retry_after_header from crawlee.configuration import Configuration -from crawlee.request_loaders import WaitingStatus +from crawlee.request_loaders import MinIntervalSignal, PacingSignal, RateLimitedSignal, WaitingStatus from crawlee.request_loaders._throttling_request_manager import ThrottlingRequestManager from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient from crawlee.storages import RequestQueue @@ -176,25 +176,8 @@ async def test_domain_matching_is_case_insensitive( assert await manager._sub_managers['example.com'].get_total_count() == 1 # 429 bookkeeping must also flow through the lowercased state. - manager.record_domain_delay('https://Example.Com/page2') - 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 + manager.record_pacing_signal(RateLimitedSignal('https://Example.Com/page2')) + assert manager._domain_states['example.com'].throttled_until > datetime.now(UTC) @pytest.mark.parametrize( @@ -225,7 +208,7 @@ async def test_domain_matching_normalizes_spelling( service_locator=service_locator, ) - assert manager.record_domain_delay(url) is True + assert manager.record_pacing_signal(RateLimitedSignal(url)) is True @pytest.mark.parametrize( @@ -289,24 +272,25 @@ async def test_add_requests_routes_mixed_domains( async def test_429_triggers_domain_delay(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """After record_domain_delay(), the domain should be throttled and the call should report success.""" - assert manager.record_domain_delay(f'https://{THROTTLED_DOMAIN}/page1') is True - assert manager._is_domain_throttled(THROTTLED_DOMAIN) + """After a rate-limited signal, the domain should be throttled and the call should report success.""" + assert manager.record_pacing_signal(RateLimitedSignal(f'https://{THROTTLED_DOMAIN}/page1')) is True + assert manager._domain_states[THROTTLED_DOMAIN].throttled_until > datetime.now(UTC) -async def test_record_domain_delay_returns_false_for_unconfigured_domain( +async def test_rate_limit_refused_unconfigured( manager: ThrottlingRequestManager[RequestQueue], ) -> None: - """record_domain_delay should return False (no-op) when the URL's domain is not in the configured list.""" - assert manager.record_domain_delay(f'https://{NON_THROTTLED_DOMAIN}/page1') is False - assert not manager._is_domain_throttled(NON_THROTTLED_DOMAIN) + """A rate-limited signal should return False (no-op) when the URL's domain is not in the configured list.""" + assert manager.record_pacing_signal(RateLimitedSignal(f'https://{NON_THROTTLED_DOMAIN}/page1')) is False + assert manager._domain_states[THROTTLED_DOMAIN].consecutive_429_count == 0 -async def test_different_domains_independent(manager: ThrottlingRequestManager[RequestQueue]) -> None: +async def test_different_domains_independent(two_domain_manager: ThrottlingRequestManager[RequestQueue]) -> None: """Throttling one domain should NOT affect other domains.""" - manager.record_domain_delay(f'https://{THROTTLED_DOMAIN}/page1') - assert manager._is_domain_throttled(THROTTLED_DOMAIN) - assert not manager._is_domain_throttled(NON_THROTTLED_DOMAIN) + two_domain_manager.record_pacing_signal(RateLimitedSignal(f'https://{THROTTLED_DOMAIN}/page1')) + now = datetime.now(UTC) + assert two_domain_manager._domain_states[THROTTLED_DOMAIN].throttled_until > now + assert two_domain_manager._domain_states[SECOND_THROTTLED_DOMAIN].throttled_until <= now async def test_exponential_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -315,12 +299,12 @@ async def test_exponential_backoff(manager: ThrottlingRequestManager[RequestQueu state = manager._domain_states[THROTTLED_DOMAIN] with _frozen_clock() as clock: - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert state.backoff_until == clock.now.return_value + manager._base_delay # Past the first window, but well before it decays. clock.now.return_value += manager._base_delay + timedelta(seconds=1) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert state.consecutive_429_count == 2 assert state.backoff_until == clock.now.return_value + manager._base_delay * 2 @@ -333,7 +317,7 @@ async def test_burst_of_429s_counts_once(manager: ThrottlingRequestManager[Reque with _frozen_clock() as clock: for _ in range(8): - assert manager.record_domain_delay(url) is True + assert manager.record_pacing_signal(RateLimitedSignal(url)) is True assert state.consecutive_429_count == 1 assert state.backoff_until == clock.now.return_value + manager._base_delay @@ -345,13 +329,13 @@ async def test_backoff_decays_when_quiet(manager: ThrottlingRequestManager[Reque state = manager._domain_states[THROTTLED_DOMAIN] with _frozen_clock() as clock: - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) clock.now.return_value += manager._base_delay + timedelta(seconds=1) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert state.consecutive_429_count == 2 clock.now.return_value = state.backoff_decays_at - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert state.consecutive_429_count == 1 assert state.backoff_until == clock.now.return_value + manager._base_delay @@ -385,7 +369,7 @@ async def test_max_delay_cap( with _frozen_clock() as clock: for _ in range(windows): armed_at = clock.now.return_value - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) # Step just past the window, staying short of its decay deadline. clock.now.return_value = state.backoff_until + timedelta(milliseconds=1) @@ -397,7 +381,7 @@ async def test_retry_after_header_priority(manager: ThrottlingRequestManager[Req """Explicit Retry-After should override exponential backoff.""" url = f'https://{THROTTLED_DOMAIN}/page1' - manager.record_domain_delay(url, retry_after=timedelta(seconds=30)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=30))) state = manager._domain_states[THROTTLED_DOMAIN] now = datetime.now(UTC) @@ -415,7 +399,7 @@ async def test_retry_after_exceeding_max_delay_logs_warning( url = f'https://{THROTTLED_DOMAIN}/page1' with caplog.at_level('WARNING', logger='crawlee.request_loaders._throttling_request_manager'): - manager.record_domain_delay(url, retry_after=manager._max_delay + timedelta(seconds=240)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=manager._max_delay + timedelta(seconds=240))) state = manager._domain_states[THROTTLED_DOMAIN] actual_delay = state.throttled_until - datetime.now(UTC) @@ -423,7 +407,7 @@ async def test_retry_after_exceeding_max_delay_logs_warning( warnings = [r for r in caplog.records if r.levelname == 'WARNING'] assert len(warnings) == 1 - assert 'Retry-After header' in warnings[0].message + assert 'requested wait' in warnings[0].message assert THROTTLED_DOMAIN in warnings[0].message @@ -433,10 +417,9 @@ async def test_retry_after_zero_falls_back_to_backoff(manager: ThrottlingRequest state = manager._domain_states[THROTTLED_DOMAIN] with _frozen_clock() as clock: - manager.record_domain_delay(url, retry_after=timedelta(0)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(0))) assert state.backoff_until == clock.now.return_value + manager._base_delay - assert manager._is_domain_throttled(THROTTLED_DOMAIN) async def test_capping_warning_names_the_backoff_on_zero_retry_after( @@ -449,7 +432,7 @@ async def test_capping_warning_names_the_backoff_on_zero_retry_after( with caplog.at_level('WARNING', logger=MANAGER_MODULE), _frozen_clock() as clock: for _ in range(20): - manager.record_domain_delay(url, retry_after=timedelta(0)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(0))) clock.now.return_value = state.backoff_until + timedelta(milliseconds=1) warnings = [r for r in caplog.records if r.levelname == 'WARNING'] @@ -457,26 +440,11 @@ async def test_capping_warning_names_the_backoff_on_zero_retry_after( assert all('exponential backoff' in r.message for r in warnings) -async def test_success_resets_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """An explicit `record_success` should reset the consecutive 429 count.""" - url = f'https://{THROTTLED_DOMAIN}/page1' - state = manager._domain_states[THROTTLED_DOMAIN] - - with _frozen_clock() as clock: - manager.record_domain_delay(url) - clock.now.return_value += manager._base_delay + timedelta(seconds=1) - manager.record_domain_delay(url) - assert state.consecutive_429_count == 2 - - manager.record_success(url) - assert state.consecutive_429_count == 0 - - async def test_handled_request_keeps_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: """Marking a request as handled must not reset the backoff, since failed requests are marked handled too.""" url = f'https://{THROTTLED_DOMAIN}/page1' await manager.add_request(url) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) request = await manager._sub_managers[THROTTLED_DOMAIN].fetch_next_request() assert request is not None @@ -489,32 +457,90 @@ async def test_handled_request_keeps_backoff(manager: ThrottlingRequestManager[R async def test_crawl_delay_integration(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """set_crawl_delay() should record the delay for the domain.""" + """A min-interval signal should record the delay for the domain.""" url = f'https://{THROTTLED_DOMAIN}/page1' - manager.set_crawl_delay(url, 5) + manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=5), scope='hostname')) state = manager._domain_states[THROTTLED_DOMAIN] assert state.crawl_delay == timedelta(seconds=5) +async def test_min_interval_keeps_first_value(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A later min-interval signal is accepted but doesn't change the crawl-delay.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=5), scope='hostname')) + + accepted = manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=9), scope='hostname')) + + assert accepted is True + assert manager._domain_states[THROTTLED_DOMAIN].crawl_delay == timedelta(seconds=5) + + +async def test_min_interval_refused_unconfigured(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A min-interval signal for a domain not in the configured list is a no-op.""" + signal = MinIntervalSignal(f'https://{NON_THROTTLED_DOMAIN}/page1', interval=timedelta(seconds=5), scope='hostname') + + assert manager.record_pacing_signal(signal) is False + assert manager._domain_states[THROTTLED_DOMAIN].crawl_delay is None + + +@pytest.mark.parametrize( + 'signal', + [ + pytest.param(RateLimitedSignal(f'https://{THROTTLED_DOMAIN}/page1', scope='site'), id='rate-limited'), + pytest.param( + MinIntervalSignal(f'https://{THROTTLED_DOMAIN}/page1', interval=timedelta(seconds=5), scope='site'), + id='min-interval', + ), + pytest.param(RateLimitedSignal(f'https://{NON_THROTTLED_DOMAIN}/page1', scope='site'), id='unconfigured'), + pytest.param( + RateLimitedSignal(f'https://{THROTTLED_DOMAIN}/page1', scope='registrable_domain'), id='registrable-domain' + ), + ], +) +async def test_unknown_scope_raises(manager: ThrottlingRequestManager[RequestQueue], signal: PacingSignal) -> None: + """A signal scoped wider than a hostname, or in a vocabulary the manager doesn't know, is rejected.""" + with pytest.raises(ValueError, match='Cannot honor a pacing signal scoped to'): + manager.record_pacing_signal(signal) + + +@pytest.mark.parametrize( + 'scope', + [ + pytest.param(None, id='no-scope'), + pytest.param('hostname', id='hostname'), + ], +) +async def test_hostname_scope_is_honored(manager: ThrottlingRequestManager[RequestQueue], scope: str | None) -> None: + """A signal with no scope or scoped to a hostname is applied to the URL's domain.""" + assert manager.record_pacing_signal(RateLimitedSignal(f'https://{THROTTLED_DOMAIN}/page1', scope=scope)) is True + + +async def test_unknown_reason_is_refused(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A signal with a reason the manager doesn't handle is refused, even with a scope it can't honor.""" + signal = Mock(reason='future', scope='site', url=f'https://{THROTTLED_DOMAIN}/page1') + + assert manager.record_pacing_signal(signal) is False + + async def test_crawl_delay_throttles_after_dispatch(manager: ThrottlingRequestManager[RequestQueue]) -> None: """After dispatching a request, crawl-delay should throttle the next one.""" url = f'https://{THROTTLED_DOMAIN}/page1' - manager.set_crawl_delay(url, 5) + manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=5), scope='hostname')) manager._mark_domain_dispatched(THROTTLED_DOMAIN) - assert manager._is_domain_throttled(THROTTLED_DOMAIN) + assert manager._domain_states[THROTTLED_DOMAIN].throttled_until > datetime.now(UTC) async def test_dispatch_does_not_shorten_active_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: """A short crawl-delay armed on dispatch must not cut an active 429 backoff short.""" url = f'https://{THROTTLED_DOMAIN}/page1' - manager.set_crawl_delay(url, 1) + manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=1), scope='hostname')) state = manager._domain_states[THROTTLED_DOMAIN] with _frozen_clock() as clock: - manager.record_domain_delay(url, retry_after=timedelta(seconds=30)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=30))) manager._mark_domain_dispatched(THROTTLED_DOMAIN) assert state.crawl_delay_until == clock.now.return_value + timedelta(seconds=1) @@ -524,7 +550,7 @@ async def test_dispatch_does_not_shorten_active_backoff(manager: ThrottlingReque async def test_backoff_escalates_under_a_long_crawl_delay(manager: ThrottlingRequestManager[RequestQueue]) -> None: """A crawl-delay longer than the backoff sets the retry cadence, and the exponent must still escalate.""" url = f'https://{THROTTLED_DOMAIN}/page1' - manager.set_crawl_delay(url, 10) + manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=10), scope='hostname')) state = manager._domain_states[THROTTLED_DOMAIN] latency = timedelta(milliseconds=200) @@ -534,7 +560,7 @@ async def test_backoff_escalates_under_a_long_crawl_delay(manager: ThrottlingReq clock.now.return_value = max(clock.now.return_value, state.throttled_until) manager._mark_domain_dispatched(THROTTLED_DOMAIN) clock.now.return_value += latency - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert state.consecutive_429_count == 5 assert state.backoff_until == clock.now.return_value + manager._base_delay * 2**4 @@ -580,7 +606,7 @@ async def test_fetch_skips_throttled_sub_manager( await manager.add_request(throttled_url) await manager.add_request(free_url) - manager.record_domain_delay(throttled_url) + manager.record_pacing_signal(RateLimitedSignal(throttled_url)) result = await manager.fetch_next_request() @@ -592,7 +618,7 @@ async def test_fetch_returns_none_when_all_throttled(manager: ThrottlingRequestM """A throttled domain must release the caller's concurrency slot instead of waiting out its cooldown.""" url = f'https://{THROTTLED_DOMAIN}/page1' await manager.add_request(url) - manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=60))) assert await asyncio.wait_for(manager.fetch_next_request(), timeout=1.0) is None @@ -601,7 +627,7 @@ async def test_cooldown_reads_as_waiting(manager: ThrottlingRequestManager[Reque """Test that requests waiting out a cooldown make the manager wait until the cooldown ends.""" url = f'https://{THROTTLED_DOMAIN}/page1' await manager.add_request(url) - manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=60))) ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) @@ -609,7 +635,9 @@ async def test_cooldown_reads_as_waiting(manager: ThrottlingRequestManager[Reque async def test_crawl_delay_reads_as_waiting(manager: ThrottlingRequestManager[RequestQueue]) -> None: """A crawl-delay armed on dispatch makes the manager wait before the domain's next request.""" - manager.set_crawl_delay(f'https://{THROTTLED_DOMAIN}/', 30) + manager.record_pacing_signal( + MinIntervalSignal(f'https://{THROTTLED_DOMAIN}/', interval=timedelta(seconds=30), scope='hostname') + ) await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1') await manager.add_request(f'https://{THROTTLED_DOMAIN}/page2') @@ -627,7 +655,7 @@ async def test_expired_throttle_makes_the_domain_fetchable_again( await manager.add_request(url) with _frozen_clock() as clock: - manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=60))) assert (await manager.check_readiness()).status == 'waiting' clock.now.return_value += timedelta(seconds=61) @@ -646,8 +674,8 @@ async def test_waits_for_earliest_cooldown(two_domain_manager: ThrottlingRequest await two_domain_manager.add_request(early_url) with _frozen_clock(): - two_domain_manager.record_domain_delay(late_url, retry_after=timedelta(seconds=30)) - two_domain_manager.record_domain_delay(early_url, retry_after=timedelta(seconds=10)) + two_domain_manager.record_pacing_signal(RateLimitedSignal(late_url, wait=timedelta(seconds=30))) + two_domain_manager.record_pacing_signal(RateLimitedSignal(early_url, wait=timedelta(seconds=10))) status = await two_domain_manager.check_readiness() assert status == WaitingStatus(ready_at=CLOCK_START + timedelta(seconds=10)) @@ -656,14 +684,14 @@ async def test_waits_for_earliest_cooldown(two_domain_manager: ThrottlingRequest async def test_empty_cooldown_not_waited(manager: ThrottlingRequestManager[RequestQueue]) -> None: """Test that a domain in a cooldown with nothing left to do doesn't keep the manager from finishing.""" url = f'https://{THROTTLED_DOMAIN}/page1' - manager.set_crawl_delay(url, 30) + manager.record_pacing_signal(MinIntervalSignal(url, interval=timedelta(seconds=30), scope='hostname')) await manager.add_request(url) request = await manager.fetch_next_request() assert request is not None await manager.mark_request_as_handled(request) - assert manager.is_throttled(url) + assert manager._domain_states[THROTTLED_DOMAIN].throttled_until > datetime.now(UTC) assert (await manager.check_readiness()).status == 'finished' @@ -925,11 +953,11 @@ async def test_purge_clears_requests_and_resets_throttle_state( """purge should empty inner and sub-managers, reset transient backoff, and preserve domain configuration.""" url = f'https://{THROTTLED_DOMAIN}/page1' await manager.add_request(url) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) state = manager._domain_states[THROTTLED_DOMAIN] assert state.consecutive_429_count == 1 - assert manager._is_domain_throttled(THROTTLED_DOMAIN) + assert state.throttled_until > datetime.now(UTC) await manager.purge() @@ -939,7 +967,7 @@ async def test_purge_clears_requests_and_resets_throttle_state( assert (await manager.check_readiness()).status == 'finished' # Transient backoff state is reset. assert state.consecutive_429_count == 0 - assert not manager._is_domain_throttled(THROTTLED_DOMAIN) + assert state.throttled_until <= datetime.now(UTC) async def test_sub_managers_opened_for_every_configured_domain( @@ -1109,9 +1137,9 @@ async def test_reclaim_routes_to_sub_manager_after_restart(fs_service_locator: S def make_stalled(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) + manager.record_pacing_signal(RateLimitedSignal(url)) clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) async def test_stall_reported_after_window(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1120,12 +1148,12 @@ async def test_stall_reported_after_window(manager: ThrottlingRequestManager[Req await manager.add_request(url) with _frozen_clock() as clock: - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) readiness = await manager.check_readiness() assert readiness.status == 'stalled' @@ -1139,9 +1167,9 @@ async def test_suppressed_429_extends_stall(manager: ThrottlingRequestManager[Re state = manager._domain_states[THROTTLED_DOMAIN] with _frozen_clock() as clock: - manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=60))) clock.now.return_value += timedelta(seconds=30) - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert state.consecutive_429_count == 1 assert state.rate_limited_since == CLOCK_START @@ -1155,7 +1183,7 @@ async def test_first_429_after_idle_not_stalled(manager: ThrottlingRequestManage with _frozen_clock() as clock: clock.now.return_value += manager._max_domain_stall * 2 - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) @@ -1166,7 +1194,7 @@ async def test_old_rate_limit_not_stalled(manager: ThrottlingRequestManager[Requ await manager.add_request(url) with _frozen_clock() as clock: - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) assert (await manager.check_readiness()).status == 'ready' @@ -1217,17 +1245,6 @@ async def test_reclaim_keeps_stall(manager: ThrottlingRequestManager[RequestQueu assert (await manager.check_readiness()).status == 'stalled' -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: - make_stalled(manager, clock, url) - manager.record_success(url) - assert (await manager.check_readiness()).status == 'stalled' - - @pytest.mark.parametrize( 'ready_url', [ @@ -1257,7 +1274,7 @@ async def test_waiting_domain_masks_stall(two_domain_manager: ThrottlingRequestM with _frozen_clock() as clock: make_stalled(two_domain_manager, clock, url) - two_domain_manager.record_domain_delay(waiting_url, retry_after=timedelta(seconds=60)) + two_domain_manager.record_pacing_signal(RateLimitedSignal(waiting_url, wait=timedelta(seconds=60))) ready_at = two_domain_manager._domain_states[SECOND_THROTTLED_DOMAIN].throttled_until assert await two_domain_manager.check_readiness() == WaitingStatus(ready_at=ready_at) @@ -1271,7 +1288,7 @@ async def test_empty_waiting_domain_does_not_mask_stall( with _frozen_clock() as clock: make_stalled(two_domain_manager, clock, url) - two_domain_manager.record_domain_delay(f'https://{SECOND_THROTTLED_DOMAIN}/page1') + two_domain_manager.record_pacing_signal(RateLimitedSignal(f'https://{SECOND_THROTTLED_DOMAIN}/page1')) readiness = await two_domain_manager.check_readiness() assert readiness.status == 'stalled' @@ -1382,6 +1399,37 @@ async def test_max_domain_stall_must_be_positive( ) +@pytest.mark.parametrize( + 'option', + [ + pytest.param('base_delay', id='base-delay'), + pytest.param('max_delay', id='max-delay'), + ], +) +@pytest.mark.parametrize( + 'delay', + [ + pytest.param(timedelta(0), id='zero'), + pytest.param(timedelta(seconds=-1), id='negative'), + ], +) +async def test_delays_must_be_positive( + inner_queue: RequestQueue, + service_locator: ServiceLocator, + option: str, + delay: timedelta, +) -> None: + """A non-positive `base_delay` or `max_delay` is rejected.""" + with pytest.raises(ValueError, match=option): + ThrottlingRequestManager( + inner_queue, + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + **{option: delay}, + ) + + async def test_inner_request_migrates_while_throttled( manager: ThrottlingRequestManager[RequestQueue], inner_queue: RequestQueue, @@ -1389,7 +1437,7 @@ async def test_inner_request_migrates_while_throttled( """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)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=60))) assert await manager.fetch_next_request() is None @@ -1424,7 +1472,7 @@ async def test_migrated_copy_is_unhandled( await inner_queue.add_request(url) with _frozen_clock() as clock: - manager.record_domain_delay(url) + manager.record_pacing_signal(RateLimitedSignal(url)) assert await manager.fetch_next_request() is None clock.now.return_value = manager._domain_states[THROTTLED_DOMAIN].throttled_until @@ -1441,7 +1489,7 @@ async def test_migration_count_is_capped( ) -> 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)) + manager.record_pacing_signal(RateLimitedSignal(f'https://{THROTTLED_DOMAIN}/', wait=timedelta(seconds=60))) with patch(f'{MANAGER_MODULE}._MAX_INNER_MIGRATIONS', 2): assert await manager.fetch_next_request() is None @@ -1467,7 +1515,7 @@ async def test_migration_failure_reclaims_request( """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)) + manager.record_pacing_signal(RateLimitedSignal(url, wait=timedelta(seconds=60))) await manager._ensure_sub_managers() sub_manager = manager._sub_managers[THROTTLED_DOMAIN] diff --git a/tests/unit/utils.py b/tests/unit/utils.py index 71ce08763c..55873193de 100644 --- a/tests/unit/utils.py +++ b/tests/unit/utils.py @@ -71,8 +71,8 @@ async def open_throttler_held_until_reclaim( ) -> ThrottlingRequestManager[RequestQueue]: """Open a throttler for `throttled.placeholder.com` whose backoff lasts until the crawler gives a request back. - The throttler runs on a stopped clock, so a backoff can't run out before the crawler checks whether the domain is - held back. Each reclaimed request moves the clock past the backoff, so a deferred request is dispatched right away. + The throttler runs on a stopped clock, so a backoff can't run out before the crawler reports the rate limit again. + Each reclaimed request moves the clock past the backoff, so a deferred request is dispatched right away. """ backoff = timedelta(minutes=1) clock = Mock()