diff --git a/docs/guides/request_throttling.mdx b/docs/guides/request_throttling.mdx index cdc420c831..81cf700104 100644 --- a/docs/guides/request_throttling.mdx +++ b/docs/guides/request_throttling.mdx @@ -34,7 +34,7 @@ To use request throttling, create a timedelta | None: value: The raw Retry-After header value. Returns: - A timedelta representing the delay, or None if the header is missing or unparsable. + A timedelta representing the delay, or None if the header is missing, unparsable, or not a positive delay. """ if not value: return None @@ -30,10 +30,10 @@ def parse_retry_after_header(value: str | None) -> timedelta | None: except ValueError: pass # Not an integer, fall through to the HTTP-date form below. else: - if seconds < 0: - # A negative delay is malformed. Reject it instead of returning a negative `timedelta`, which would - # push `throttled_until` into the past and silently disable the 429 back-off downstream. - logger.debug(f'Retry-After delay-seconds {value!r} is negative; ignoring.') + if seconds <= 0: + # A negative delay is malformed and a zero one carries no backoff, so reject both and let the caller + # apply its own backoff instead. + logger.debug(f'Retry-After delay-seconds {value!r} is not positive; ignoring.') return None return timedelta(seconds=seconds) diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index bd67eb7d61..faa998018b 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -28,7 +28,13 @@ TRequestManager = TypeVar('TRequestManager', bound=RequestManager) _NEVER_THROTTLED = datetime.min.replace(tzinfo=timezone.utc) -"""Sentinel `throttled_until` value meaning the domain has no active backoff.""" +"""Sentinel timestamp meaning one of a domain's throttle clocks has never been armed.""" + +_MAX_BACKOFF_EXPONENT = 20 +"""Highest exponent the 429 backoff doubles to. `max_delay` caps the delay far below this, while an unbounded exponent +eventually overflows the `timedelta` multiplication. Low enough that the doubling stays representable for any +`base_delay` up to a year. +""" @docs_group('Request loaders') @@ -143,14 +149,14 @@ async def purge(self) -> None: """Empty the inner manager and all sub-managers, and reset transient per-domain throttle state. The configured domain list and any robots.txt-derived `crawl_delay` are preserved. Only the dynamic backoff - state (consecutive 429 counter and `throttled_until`) is cleared. Sub-managers stay open; they're just emptied. + state (consecutive 429 counter and the throttle clocks) is cleared. Sub-managers stay open; they're just + emptied. """ await self._ensure_sub_managers() await asyncio.gather(self._inner.purge(), *(sm.purge() for sm in self._sub_managers.values())) self._in_flight_from_inner.clear() for state in self._domain_states.values(): - state.consecutive_429_count = 0 - state.throttled_until = _NEVER_THROTTLED + state.reset_throttling() @override async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None: @@ -255,7 +261,6 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | manager = self._fetch_owner(request) result = await manager.mark_request_as_handled(request) self._clear_fetch_owner(request) - self.record_success(request.url) return result @override @@ -298,33 +303,58 @@ async def is_finished(self) -> bool: 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. - Increments the consecutive 429 count and calculates the next allowed request time using exponential backoff or - the `Retry-After` value. + 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. Args: url: The URL that received a 429 response. - retry_after: Optional delay from the `Retry-After` header. If provided, it takes priority over the - calculated exponential backoff. + retry_after: Optional delay from the `Retry-After` header. If it describes a positive delay, it takes + priority over the calculated exponential backoff. Returns: - True if the URL's domain is configured for throttling and the delay was applied; 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 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. """ state = self._get_domain_state(url) if state is None: return False + now = datetime.now(timezone.utc) + + # Requests in flight when the limit was hit all come back 429. That is one rate-limit event, so only the first + # advances the exponent. Checking `crawl_delay_until` too would swallow every 429, as it is armed on every + # dispatch. + 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. + if now >= state.backoff_decays_at: + state.consecutive_429_count = 0 + state.consecutive_429_count += 1 - delay = retry_after if retry_after is not None else self._base_delay * (2 ** (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' + else: + delay = self._base_delay * 2 ** min(state.consecutive_429_count - 1, _MAX_BACKOFF_EXPONENT) + source = 'exponential backoff' + if delay > self._max_delay: - source = 'Retry-After header' if retry_after is not None else 'exponential backoff' logger.warning( f'Capping {source} delay of {delay.total_seconds():.1f}s for domain "{state.domain}" ' f'to max_delay ({self._max_delay.total_seconds():.1f}s); the domain may continue to rate-limit. ' f'Consider increasing max_delay if this recurs.' ) delay = self._max_delay - state.throttled_until = datetime.now(timezone.utc) + delay + + state.apply_backoff(now, delay) logger.info( f'Rate limit (429) detected for domain "{state.domain}" ' @@ -333,7 +363,10 @@ def record_domain_delay(self, url: str, *, retry_after: timedelta | None = None) return True def record_success(self, url: str) -> None: - """Record a successful request, resetting the backoff state for that domain. + """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. Args: url: The URL that received a successful response. @@ -457,11 +490,11 @@ def _fetchable_domains(self) -> list[str]: def _mark_domain_dispatched(self, domain: str) -> None: """Record that a request to this domain was just dispatched. - If a crawl-delay is configured, push throttled_until forward by that amount. + If a crawl-delay is configured, push `crawl_delay_until` forward by that amount. """ state = self._domain_states.get(domain) - if state is not None and state.crawl_delay is not None: - state.throttled_until = datetime.now(timezone.utc) + state.crawl_delay + if state is not None: + state.apply_crawl_delay(datetime.now(timezone.utc)) def _fetch_owner(self, request: Request) -> TRequestManager: """Return the manager the request must be given back to, leaving its in-flight record in place. @@ -501,11 +534,47 @@ class _DomainState: domain: str """The domain being tracked.""" - throttled_until: datetime = _NEVER_THROTTLED - """Earliest time the next request to this domain is allowed.""" + backoff_until: datetime = _NEVER_THROTTLED + """Earliest time the next request is allowed by the 429 backoff. Kept apart from `crawl_delay_until`, which is + armed on every dispatch and would otherwise pass for an active backoff. + """ + + crawl_delay_until: datetime = _NEVER_THROTTLED + """Earliest time the next request is allowed by the domain's crawl-delay.""" + + backoff_decays_at: datetime = _NEVER_THROTTLED + """Time after which an incoming 429 is treated as a fresh burst rather than a continuation of the current one.""" consecutive_429_count: int = 0 """Number of consecutive 429 responses (for exponential backoff).""" crawl_delay: timedelta | None = None - """Minimum interval between requests, used to push `throttled_until` on dispatch.""" + """Minimum interval between requests, used to push `crawl_delay_until` on dispatch.""" + + @property + def throttled_until(self) -> datetime: + """Earliest time the next request to this domain is allowed by either of its two independent clocks.""" + return max(self.backoff_until, self.crawl_delay_until) + + def apply_backoff(self, now: datetime, delay: timedelta) -> None: + """Block the domain for `delay`. + + If no 429 arrives for another `delay` after the domain becomes dispatchable again, the exponent resets. + """ + self.backoff_until = now + delay + # The quiet period runs from the moment the domain becomes dispatchable again, not from `backoff_until`. A + # crawl-delay longer than `delay` sets the retry cadence, and measuring from `backoff_until` would expire the + # window before the domain is even retried, making every 429 look like a fresh burst. + self.backoff_decays_at = self.throttled_until + delay + + def apply_crawl_delay(self, now: datetime) -> None: + """Block the domain for its crawl-delay, if it declared one.""" + if self.crawl_delay is not None: + self.crawl_delay_until = now + self.crawl_delay + + def reset_throttling(self) -> None: + """Clear the transient throttle state.""" + self.consecutive_429_count = 0 + self.backoff_until = _NEVER_THROTTLED + self.crawl_delay_until = _NEVER_THROTTLED + self.backoff_decays_at = _NEVER_THROTTLED diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index 68e79f3b79..fbc899b5a1 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -3,10 +3,11 @@ from __future__ import annotations import asyncio +from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pathlib import Path -from typing import Any -from unittest.mock import AsyncMock +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -18,11 +19,17 @@ from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient from crawlee.storages import RequestQueue +if TYPE_CHECKING: + from collections.abc import Iterator + THROTTLED_DOMAIN = 'throttled.com' SECOND_THROTTLED_DOMAIN = 'slow.com' NON_THROTTLED_DOMAIN = 'free.com' TEST_DOMAINS = [THROTTLED_DOMAIN] +MANAGER_MODULE = 'crawlee.request_loaders._throttling_request_manager' +CLOCK_START = datetime(2026, 1, 1, tzinfo=timezone.utc) + @pytest.fixture def memory_storage_client() -> MemoryStorageClient: @@ -78,6 +85,14 @@ def _make_request(url: str) -> Request: return Request.from_url(url) +@contextmanager +def _frozen_clock() -> Iterator[MagicMock]: + """Freeze the manager's clock at `CLOCK_START`. Move it by reassigning `clock.now.return_value`.""" + with patch(f'{MANAGER_MODULE}.datetime') as clock: + clock.now.return_value = CLOCK_START + yield clock + + async def _open_fs_manager(service_locator: ServiceLocator) -> ThrottlingRequestManager[RequestQueue]: """Open a throttling manager over the persistent storage directory, as a fresh process would.""" inner_queue = await RequestQueue.open( @@ -277,32 +292,87 @@ async def test_different_domains_independent(manager: ThrottlingRequestManager[R async def test_exponential_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """Consecutive 429s should increase delay exponentially.""" + """429s in successive backoff windows should increase the delay exponentially.""" url = f'https://{THROTTLED_DOMAIN}/page1' + state = manager._domain_states[THROTTLED_DOMAIN] - manager.record_domain_delay(url) + with _frozen_clock() as clock: + manager.record_domain_delay(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) + + assert state.consecutive_429_count == 2 + assert state.backoff_until == clock.now.return_value + manager._base_delay * 2 + + +async def test_burst_of_429s_counts_once(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Requests in flight when the limit was hit all return 429, but they are a single rate-limit event.""" + url = f'https://{THROTTLED_DOMAIN}/page1' state = manager._domain_states[THROTTLED_DOMAIN] - first_until = state.throttled_until - manager.record_domain_delay(url) - second_until = state.throttled_until + with _frozen_clock() as clock: + for _ in range(8): + assert manager.record_domain_delay(url) is True - assert second_until > first_until - assert state.consecutive_429_count == 2 + assert state.consecutive_429_count == 1 + assert state.backoff_until == clock.now.return_value + manager._base_delay -async def test_max_delay_cap(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """Backoff should cap at max_delay (60s).""" +async def test_backoff_decays_when_quiet(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A domain that stops rate-limiting for a full extra window should start the exponent over.""" 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 - for _ in range(20): + clock.now.return_value = state.backoff_decays_at manager.record_domain_delay(url) + assert state.consecutive_429_count == 1 + assert state.backoff_until == clock.now.return_value + manager._base_delay + + +@pytest.mark.parametrize( + ('base_delay', 'windows'), + [ + pytest.param(timedelta(seconds=2), 20, id='default base delay'), + pytest.param(timedelta(seconds=2), 60, id='sustained rate limiting'), + pytest.param(timedelta(hours=6), 40, id='base delay in hours'), + ], +) +async def test_max_delay_cap( + inner_queue: RequestQueue, + service_locator: ServiceLocator, + base_delay: timedelta, + windows: int, +) -> None: + """Backoff caps at `max_delay` and keeps the doubling representable however long the domain rate-limits.""" + manager = ThrottlingRequestManager( + inner_queue, + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + base_delay=base_delay, + ) + url = f'https://{THROTTLED_DOMAIN}/page1' state = manager._domain_states[THROTTLED_DOMAIN] - now = datetime.now(timezone.utc) - actual_delay = state.throttled_until - now - assert actual_delay <= manager._max_delay + timedelta(seconds=1) + with _frozen_clock() as clock: + for _ in range(windows): + armed_at = clock.now.return_value + manager.record_domain_delay(url) + # Step just past the window, staying short of its decay deadline. + clock.now.return_value = state.backoff_until + timedelta(milliseconds=1) + + assert state.consecutive_429_count == windows + assert state.backoff_until - armed_at == manager._max_delay async def test_retry_after_header_priority(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -339,16 +409,62 @@ async def test_retry_after_exceeding_max_delay_logs_warning( assert THROTTLED_DOMAIN in warnings[0].message +async def test_retry_after_zero_falls_back_to_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A zero Retry-After is no delay at all, so the exponential backoff should still engage.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + state = manager._domain_states[THROTTLED_DOMAIN] + + with _frozen_clock() as clock: + manager.record_domain_delay(url, retry_after=timedelta(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( + manager: ThrottlingRequestManager[RequestQueue], + caplog: pytest.LogCaptureFixture, +) -> None: + """A zero Retry-After falls through to the backoff, so the capping warning must not blame the header.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + state = manager._domain_states[THROTTLED_DOMAIN] + + 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)) + clock.now.return_value = state.backoff_until + timedelta(milliseconds=1) + + warnings = [r for r in caplog.records if r.levelname == 'WARNING'] + assert warnings + assert all('exponential backoff' in r.message for r in warnings) + + async def test_success_resets_backoff(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """Successful request should reset the consecutive 429 count.""" + """An explicit `record_success` should reset the consecutive 429 count.""" url = f'https://{THROTTLED_DOMAIN}/page1' + state = manager._domain_states[THROTTLED_DOMAIN] - manager.record_domain_delay(url) - manager.record_domain_delay(url) - assert manager._domain_states[THROTTLED_DOMAIN].consecutive_429_count == 2 + 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 manager._domain_states[THROTTLED_DOMAIN].consecutive_429_count == 0 + 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) + + request = await manager._sub_managers[THROTTLED_DOMAIN].fetch_next_request() + assert request is not None + await manager.mark_request_as_handled(request) + + assert manager._domain_states[THROTTLED_DOMAIN].consecutive_429_count == 1 # ── Crawl-Delay Integration Tests ───────────────────────── @@ -373,6 +489,39 @@ async def test_crawl_delay_throttles_after_dispatch(manager: ThrottlingRequestMa assert manager._is_domain_throttled(THROTTLED_DOMAIN) +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) + state = manager._domain_states[THROTTLED_DOMAIN] + + with _frozen_clock() as clock: + manager.record_domain_delay(url, retry_after=timedelta(seconds=30)) + manager._mark_domain_dispatched(THROTTLED_DOMAIN) + + assert state.crawl_delay_until == clock.now.return_value + timedelta(seconds=1) + assert state.throttled_until == clock.now.return_value + timedelta(seconds=30) + + +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) + state = manager._domain_states[THROTTLED_DOMAIN] + latency = timedelta(milliseconds=200) + + with _frozen_clock() as clock: + for _ in range(5): + # Dispatch as soon as both clocks allow it, then let the request come back 429. + 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) + + assert state.consecutive_429_count == 5 + assert state.backoff_until == clock.now.return_value + manager._base_delay * 2**4 + + # ── Fetch Scheduling Tests ──────────────────────────── @@ -463,7 +612,7 @@ async def test_expired_throttle_makes_the_domain_fetchable_again( manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) assert await manager.is_empty() is True - manager._domain_states[THROTTLED_DOMAIN].throttled_until = datetime.now(timezone.utc) - timedelta(seconds=1) + manager._domain_states[THROTTLED_DOMAIN].backoff_until = datetime.now(timezone.utc) - timedelta(seconds=1) assert await manager.is_empty() is False result = await manager.fetch_next_request() @@ -481,8 +630,8 @@ async def test_fetch_prefers_longest_overdue_domain( await two_domain_manager.add_request(overdue_url) now = datetime.now(timezone.utc) - two_domain_manager._domain_states[THROTTLED_DOMAIN].throttled_until = now - timedelta(seconds=1) - two_domain_manager._domain_states[SECOND_THROTTLED_DOMAIN].throttled_until = now - timedelta(seconds=10) + two_domain_manager._domain_states[THROTTLED_DOMAIN].backoff_until = now - timedelta(seconds=1) + two_domain_manager._domain_states[SECOND_THROTTLED_DOMAIN].backoff_until = now - timedelta(seconds=10) result = await two_domain_manager.fetch_next_request() @@ -937,8 +1086,8 @@ def test_parse_retry_after_integer_seconds() -> None: def test_parse_retry_after_zero_seconds() -> None: - """A delay of `0` ("retry immediately") is valid and must yield a zero delta, not None.""" - assert parse_retry_after_header('0') == timedelta(0) + """A delay of `0` carries no backoff, so it must be reported as a missing header.""" + assert parse_retry_after_header('0') is None def test_parse_retry_after_negative_seconds() -> None: