From 4b63cfe3017a877b71a4e60338beddb90cdccc65 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Thu, 13 Aug 2026 20:57:21 +0000 Subject: [PATCH] correct the backoff for concurrent 429s and zero Retry-After --- src/crawlee/_utils/http.py | 10 +- .../_throttling_request_manager.py | 92 ++++++++--- tests/unit/test_throttling_request_manager.py | 144 +++++++++++++++--- 3 files changed, 195 insertions(+), 51 deletions(-) diff --git a/src/crawlee/_utils/http.py b/src/crawlee/_utils/http.py index 4713f66dea..ce92ba3bb0 100644 --- a/src/crawlee/_utils/http.py +++ b/src/crawlee/_utils/http.py @@ -19,7 +19,7 @@ def parse_retry_after_header(value: str | None) -> 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, a zero one carries no back-off. Reject both, so the caller falls back to + # its own back-off instead of silently losing it. + 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 d48c49ec92..31ae02be47 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -28,7 +28,7 @@ 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 a dispatch clock has never been armed.""" @docs_group('Request loaders') @@ -120,13 +120,12 @@ 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 are kept around so they don't - need to be re-opened on the next request — they're just emptied. + state (consecutive 429 counter and the two dispatch clocks) is cleared. Sub-managers are kept around so they + don't need to be re-opened on the next request — they're just emptied. """ await asyncio.gather(self._inner.purge(), *(sm.purge() for sm in self._sub_managers.values())) 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: @@ -258,9 +257,7 @@ async def reclaim_request(self, request: Request, *, forefront: bool = False) -> @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: manager = self._select_manager(request.url) - result = await manager.mark_request_as_handled(request) - self.record_success(request.url) - return result + return await manager.mark_request_as_handled(request) @override async def get_handled_count(self) -> int: @@ -291,33 +288,54 @@ 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 + True if the URL's domain is configured for throttling and the 429 was recorded; 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: + 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 ** (state.consecutive_429_count - 1)) + 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}" ' @@ -398,11 +416,11 @@ def _get_earliest_available_time(self, now: datetime) -> datetime: 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 _signal_new_work(self) -> None: """Wake `fetch_next_request` if it is sleeping inside a throttle wait.""" @@ -450,11 +468,41 @@ 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 that, the exponent resets.""" + self.backoff_until = now + delay + self.backoff_decays_at = self.backoff_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 0451297fff..c32ca32949 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -3,9 +3,10 @@ from __future__ import annotations import asyncio +from contextlib import contextmanager from datetime import datetime, timedelta, timezone -from typing import Any -from unittest.mock import AsyncMock, patch +from typing import TYPE_CHECKING, Any +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -16,10 +17,16 @@ from crawlee.storage_clients import MemoryStorageClient from crawlee.storages import RequestQueue +if TYPE_CHECKING: + from collections.abc import Iterator + THROTTLED_DOMAIN = 'throttled.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: @@ -55,6 +62,14 @@ def _make_request(url: str) -> Request: return Request.from_url(url) +@contextmanager +def _frozen_clock(start: datetime = _CLOCK_START) -> Iterator[MagicMock]: + """Freeze the manager's clock at `start`. Move it by reassigning `clock.now.return_value`.""" + with patch(f'{_MANAGER_MODULE}.datetime') as clock: + clock.now.return_value = start + yield clock + + # ── Request Routing Tests ───────────────────────────────── @@ -163,32 +178,67 @@ 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 + + +async def test_max_delay_cap(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Backoff should cap at max_delay (60s).""" + 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(20): + 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 == 20 + assert state.backoff_until - armed_at == manager._max_delay async def test_retry_after_header_priority(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -225,16 +275,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 ───────────────────────── @@ -320,9 +416,9 @@ async def test_sleep_when_all_throttled(manager: ThrottlingRequestManager[Reques with patch(target, new_callable=AsyncMock) as mock_wait: async def wait_side_effect(*_args: Any, **_kwargs: Any) -> None: - # Set throttled_until firmly in the past so the next iteration reliably unblocks the domain regardless of + # Set the backoff firmly in the past so the next iteration reliably unblocks the domain regardless of # clock resolution or scheduling jitter on slow CI runners. - 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) mock_wait.side_effect = wait_side_effect @@ -544,8 +640,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 back-off, 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: