refactor(request-loaders)!: Add pacing signals to RequestManager

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:
 
 - <ApiLink to="class/RequestLoader">`RequestLoader`</ApiLink>: The base interface for reading requests in a crawl.
-- <ApiLink to="class/RequestManager">`RequestManager`</ApiLink>: Extends `RequestLoader` with write capabilities.
+- <ApiLink to="class/RequestManager">`RequestManager`</ApiLink>: Extends `RequestLoader` with write capabilities, and with the pacing signals a crawler reports back.
 - <ApiLink to="class/RequestManagerTandem">`RequestManagerTandem`</ApiLink>: 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 <ApiLink to="class/RequestManager">`RequestManager`</ApiLink> 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 <ApiLink to="class/RequestManager">`RequestManager`</ApiLink> 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 <ApiLink to="class/RequestManager#record_pacing_signal">`RequestManager.record_pacing_signal`</ApiLink>, with one of two signals:
+
+- <ApiLink to="class/RateLimitedSignal">`RateLimitedSignal`</ApiLink>: a site refused a request because the crawler was going too fast. It carries the wait the site asked for, if any.
+- <ApiLink to="class/MinIntervalSignal">`MinIntervalSignal`</ApiLink>: 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 <ApiLink to="class/RequestQueue">`RequestQueue`</ApiLink> 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 <ApiLink to="class/RequestManagerTandem">`RequestManagerTandem`</ApiLink> does. Otherwise a pacer nested inside it never receives the signal.
+
 ## Request manager tandem
 
 The <ApiLink to="class/RequestManagerTandem">`RequestManagerTandem`</ApiLink> class allows you to combine the read-only capabilities of a `RequestLoader` (like <ApiLink to="class/RequestList">`RequestList`</ApiLink>) with the read-write capabilities of a `RequestManager` (like <ApiLink to="class/RequestQueue">`RequestQueue`</ApiLink>). 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.

docs/guides/request_throttling.mdx

@@ -42,6 +42,8 @@ To use request throttling, create a <ApiLink to="class/ThrottlingRequestManager"
 
 5. **Cooldown handling**: While a domain is in a cooldown, its queued requests don't count as dispatchable, so the crawler's autoscaled pool idles instead of keeping a worker slot blocked. The requests still count towards completion, so the crawl waits for them and finishes only once every one has been handled. A domain that never stops rate-limiting doesn't keep the crawl running forever. For details, see [Persistent rate limiting](#persistent-rate-limiting).
 
+The crawler reports 429 responses and crawl-delays through <ApiLink to="class/RequestManager#record_pacing_signal">`RequestManager.record_pacing_signal`</ApiLink>, which <ApiLink to="class/RequestManagerTandem">`RequestManagerTandem`</ApiLink> 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.

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.

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:

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.