fix(throttling-manager): apply throttling behind a tandem and with retry_on_blocked disabled

src/crawlee/crawlers/_abstract_http/_abstract_http_crawler.py

@@ -303,13 +303,13 @@ async def _handle_status_code_response(
             The original crawling context if no errors are detected.
         """
         status_code = context.http_response.status_code
+        self._record_rate_limit_status_code(
+            status_code,
+            request_url=context.request.url,
+            retry_after_header=context.http_response.headers.get('retry-after'),
+        )
         if self._retry_on_blocked:
-            self._raise_for_session_blocked_status_code(
-                context.session,
-                status_code,
-                request_url=context.request.url,
-                retry_after_header=context.http_response.headers.get('retry-after'),
-            )
+            self._raise_for_session_blocked_status_code(context.session, status_code)
         self._raise_for_error_status_code(status_code)
         yield context
 

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -65,7 +65,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 RequestManagerTandem, ThrottlingRequestManager
 from crawlee.router import Router
 from crawlee.sessions import SessionPool
 from crawlee.statistics import Statistics, StatisticsState
@@ -705,7 +705,7 @@ async def run(
 
         self._running = True
 
-        if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager):
+        if self._respect_robots_txt_file and self._get_throttling_manager() is None:
             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 '
@@ -1619,51 +1619,69 @@ def _raise_for_error_status_code(self, status_code: int) -> None:
         if is_status_code_server_error(status_code) and not is_ignored_status:
             raise HttpStatusCodeError('Error status code returned', status_code)
 
-    def _raise_for_session_blocked_status_code(
+    def _get_throttling_manager(self) -> ThrottlingRequestManager | None:
+        """Return the crawler's `ThrottlingRequestManager`, unwrapping any `RequestManagerTandem` around it."""
+        manager = self._request_manager
+        while isinstance(manager, RequestManagerTandem):
+            manager = manager.request_manager
+
+        return manager if isinstance(manager, ThrottlingRequestManager) else None
+
+    def _record_rate_limit_status_code(
         self,
-        session: Session | None,
         status_code: int,
         *,
         request_url: str,
         retry_after_header: str | None = None,
     ) -> None:
-        """Raise an exception if the given status code indicates the session is blocked.
+        """Record a 429 Too Many Requests response so the domain gets a per-domain backoff.
 
-        If the status code is 429 (Too Many Requests), the domain is recorded as rate-limited in the
-        `ThrottlingRequestManager` for per-domain backoff.
+        Other status codes are ignored. Rate limiting is independent of `retry_on_blocked`, so this runs for every
+        response, not only for those checked against the session's blocking rules.
 
         Args:
-            session: The session used for the request. If `None`, no check is performed.
             status_code: The HTTP status code to check.
             request_url: The request URL, used for per-domain rate limit tracking.
             retry_after_header: The value of the `Retry-After` response header, if present.
-
-        Raises:
-            SessionError: If the status code indicates the session is blocked.
         """
-        if status_code == HTTPStatus.TOO_MANY_REQUESTS:
-            if isinstance(self._request_manager, ThrottlingRequestManager):
-                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()
-                    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,
-                        )
-            else:
+        if status_code != HTTPStatus.TOO_MANY_REQUESTS:
+            return
+
+        throttling_manager = self._get_throttling_manager()
+
+        if throttling_manager is None:
+            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
+
+        retry_after = parse_retry_after_header(retry_after_header)
+        if not throttling_manager.record_domain_delay(request_url, retry_after=retry_after):
+            domain = (URL(request_url).host or '').lower()
+            if domain:
                 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',
+                    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,
                 )
 
+    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.
+
+        Args:
+            session: The session used for the request. If `None`, no check is performed.
+            status_code: The HTTP status code to check.
+
+        Raises:
+            SessionError: If the status code indicates the session is blocked.
+        """
         if session is not None and session.is_blocked_status_code(
             status_code=status_code,
             ignore_http_error_status_codes=self._ignore_http_error_status_codes,
@@ -1697,10 +1715,11 @@ 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):
+        throttling_manager = self._get_throttling_manager()
+        if throttling_manager is not None:
             crawl_delay = robots_txt_file.get_crawl_delay()
             if crawl_delay is not None:
-                self._request_manager.set_crawl_delay(url, crawl_delay)
+                throttling_manager.set_crawl_delay(url, crawl_delay)
 
         return robots_txt_file.is_allowed(url)
 

src/crawlee/crawlers/_playwright/_playwright_crawler.py

@@ -534,13 +534,13 @@ async def _handle_status_code_response(self, context: TPostNavContext) -> AsyncG
             The original crawling context if no errors are detected.
         """
         status_code = context.response.status
+        self._record_rate_limit_status_code(
+            status_code,
+            request_url=context.request.url,
+            retry_after_header=context.response.headers.get('retry-after'),
+        )
         if self._retry_on_blocked:
-            self._raise_for_session_blocked_status_code(
-                context.session,
-                status_code,
-                request_url=context.request.url,
-                retry_after_header=context.response.headers.get('retry-after'),
-            )
+            self._raise_for_session_blocked_status_code(context.session, status_code)
         self._raise_for_error_status_code(status_code)
         yield context
 

src/crawlee/request_loaders/_request_manager_tandem.py

@@ -32,6 +32,11 @@ def __init__(self, request_loader: RequestLoader, request_manager: RequestManage
         self._read_only_loader = request_loader
         self._read_write_manager = request_manager
 
+    @property
+    def request_manager(self) -> RequestManager:
+        """The wrapped manager that stores the requests, both its own and those handed over by the loader."""
+        return self._read_write_manager
+
     @override
     async def get_handled_count(self) -> int:
         return await self._read_write_manager.get_handled_count()

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -28,7 +28,7 @@
 from crawlee.crawlers import BasicCrawler
 from crawlee.errors import RequestCollisionError, SessionError, UserDefinedErrorHandlerError
 from crawlee.events import Event, EventCrawlerStatusData, LocalEventManager
-from crawlee.request_loaders import RequestList, RequestManagerTandem, ThrottlingRequestManager
+from crawlee.request_loaders import RequestList, RequestManager, RequestManagerTandem, ThrottlingRequestManager
 from crawlee.sessions import Session, SessionPool
 from crawlee.statistics import FinalStatistics, StatisticsState
 from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient
@@ -2464,8 +2464,8 @@ async def test_warn_no_throttling_manager_once_on_429(caplog: pytest.LogCaptureF
     """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._raise_for_session_blocked_status_code(session=None, status_code=429, request_url='https://a.test/')
-        crawler._raise_for_session_blocked_status_code(session=None, status_code=429, request_url='https://b.test/')
+        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()
@@ -2484,15 +2484,9 @@ async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest.
     crawler = BasicCrawler(configure_logging=False, request_manager=throttler)
 
     with caplog.at_level(logging.WARNING, logger='crawlee'):
-        crawler._raise_for_session_blocked_status_code(
-            session=None, status_code=429, request_url='https://A.example.com/page1'
-        )
-        crawler._raise_for_session_blocked_status_code(
-            session=None, status_code=429, request_url='https://a.example.com/page2'
-        )
-        crawler._raise_for_session_blocked_status_code(
-            session=None, status_code=429, request_url='https://other.example.com/page1'
-        )
+        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
@@ -2502,3 +2496,49 @@ async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest.
     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)
+
+
+@pytest.mark.parametrize(
+    'tandem_depth',
+    [
+        pytest.param(1, id='single_tandem'),
+        pytest.param(2, id='nested_tandems'),
+    ],
+)
+async def test_records_429_through_tandem(tandem_depth: int, caplog: pytest.LogCaptureFixture) -> None:
+    """A throttler reached through a tandem still gets the 429, and the crawler stops claiming there is no throttler."""
+    inner = await RequestQueue.open(alias=f'tandem-429-{tandem_depth}', storage_client=MemoryStorageClient())
+    throttler = ThrottlingRequestManager(
+        inner,
+        domains=['throttled.test'],
+        request_manager_opener=RequestQueue.open,
+    )
+
+    manager: RequestManager = throttler
+    for depth in range(tandem_depth):
+        manager = await RequestList([f'https://loader{depth}.test/']).to_tandem(manager)
+
+    crawler = BasicCrawler(configure_logging=False, request_manager=manager)
+
+    with caplog.at_level(logging.WARNING, logger='crawlee'):
+        crawler._record_rate_limit_status_code(429, request_url='https://throttled.test/page1')
+
+    assert throttler._is_domain_throttled('throttled.test')
+    assert not [r for r in caplog.records if 'not using' in r.getMessage()]
+
+
+async def test_no_crawl_delay_warning_for_tandem_wrapped_throttler(caplog: pytest.LogCaptureFixture) -> None:
+    """The robots.txt crawl-delay warning must not fire when the throttler is wrapped in a tandem."""
+    inner = await RequestQueue.open(alias='tandem-crawl-delay', storage_client=MemoryStorageClient())
+    throttler = ThrottlingRequestManager(
+        inner,
+        domains=['throttled.test'],
+        request_manager_opener=RequestQueue.open,
+    )
+    tandem = await RequestList([]).to_tandem(throttler)
+    crawler = BasicCrawler(configure_logging=False, request_manager=tandem, respect_robots_txt_file=True)
+
+    with caplog.at_level(logging.WARNING, logger='crawlee'):
+        await crawler.run()
+
+    assert not [r for r in caplog.records if 'Crawl-delay directives' in r.getMessage()]

tests/unit/crawlers/_http/test_http_crawler.py

@@ -9,6 +9,7 @@
 
 from crawlee import ConcurrencySettings, Request, RequestState
 from crawlee.crawlers import HttpCrawler
+from crawlee.request_loaders import ThrottlingRequestManager
 from crawlee.sessions import SessionPool
 from crawlee.statistics import Statistics
 from crawlee.storages import RequestQueue
@@ -686,3 +687,38 @@ async def failed_request_handler(context: BasicCrawlingContext, _error: Exceptio
     }
 
     await queue.drop()
+
+
+@pytest.mark.parametrize(
+    'retry_on_blocked',
+    [
+        pytest.param(True, id='retry_on_blocked'),
+        pytest.param(False, id='no_retry_on_blocked'),
+    ],
+)
+async def test_records_429_regardless_of_retry_on_blocked(
+    mock_request_handler: AsyncMock,
+    server_url: URL,
+    *,
+    retry_on_blocked: bool,
+) -> None:
+    """Rate limiting is a separate concern from session blocking, so a 429 must be recorded either way."""
+    domain = server_url.host or ''
+    inner = await RequestQueue.open(alias=f'throttle-429-{retry_on_blocked}')
+    throttler = ThrottlingRequestManager(
+        inner,
+        domains=[domain],
+        request_manager_opener=RequestQueue.open,
+    )
+    crawler = HttpCrawler(
+        request_handler=mock_request_handler,
+        request_manager=throttler,
+        retry_on_blocked=retry_on_blocked,
+        max_request_retries=0,
+        # Without this, a 429 retires the session and the retries walk the backoff up to `max_delay`.
+        use_session_pool=False,
+    )
+
+    await crawler.run([str(server_url / 'status/429')])
+
+    assert throttler._is_domain_throttled(domain)