From 62439854496083460d5b2e9ac3d21a278165e85c Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Wed, 19 Aug 2026 17:52:55 +0000 Subject: [PATCH 1/4] stop holding a concurrency slot while a domain is throttled --- .../_throttling_request_manager.py | 119 +++++------------- tests/unit/test_throttling_request_manager.py | 104 ++++++++------- 2 files changed, 94 insertions(+), 129 deletions(-) diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index d48c49ec92..2775b4d563 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -1,7 +1,6 @@ from __future__ import annotations import asyncio -import contextlib from dataclasses import dataclass from datetime import datetime, timedelta, timezone from logging import getLogger @@ -38,9 +37,9 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): Requests for explicitly configured domains are routed into dedicated sub-managers at insertion time — each request lives in exactly one manager, eliminating duplication and simplifying deduplication. - When `fetch_next_request()` is called, it returns requests from the sub-manager whose domain has been waiting the - longest. If all configured domains are throttled, it falls back to the inner manager for non-throttled domains. If - the inner manager is also empty and all sub-managers are throttled, it sleeps until the earliest cooldown expires. + `fetch_next_request()` takes from the sub-manager whose domain has been waiting the longest, skipping domains in a + cooldown, and falls back to the inner manager when no sub-manager yields a request. If nothing can be dispatched + right now, it returns `None` rather than waiting, so the caller's task slot is released. Delay sources: - HTTP 429 responses (via `record_domain_delay`) @@ -101,9 +100,6 @@ def __init__( self._request_manager_opener = request_manager_opener self._domain_states: dict[str, _DomainState] = {d.lower(): _DomainState(domain=d.lower()) for d in domains if d} self._sub_managers: dict[str, TRequestManager] = {} - self._new_work_event = asyncio.Event() - """Set whenever a request is added or reclaimed. Lets `fetch_next_request` wake from a throttle - wait early when fresh work appears, instead of sleeping for the full computed cooldown.""" @property def inner(self) -> TRequestManager: @@ -140,12 +136,9 @@ async def add_request(self, request: str | Request, *, forefront: bool = False) if domain in self._domain_states: sm = await self._get_or_create_sub_manager(domain) - result = await sm.add_request(request, forefront=forefront) - else: - result = await self._inner.add_request(request, forefront=forefront) + return await sm.add_request(request, forefront=forefront) - self._signal_new_work() - return result + return await self._inner.add_request(request, forefront=forefront) @override async def add_requests( @@ -192,68 +185,25 @@ async def add_requests( wait_for_all_requests_to_be_added_timeout=wait_for_all_requests_to_be_added_timeout, ) - if inner_requests or domain_requests: - self._signal_new_work() - @override async def fetch_next_request(self) -> Request | None: """Fetch the next request, respecting per-domain delays. - Sub-managers are checked in order of longest-overdue domain first (sorted by `throttled_until` ascending). If - all configured domains are throttled, falls back to the inner manager for non-throttled domains. If the inner - manager is also empty and all sub-managers are throttled, waits until either the earliest domain becomes - available or new work is added (whichever comes first). + Sub-managers are checked in order of longest-overdue domain first, then the inner manager. Throttled domains + are skipped rather than waited for, so a caller holding a concurrency slot gets it back instead of sleeping. """ - while True: - # Clear the event before checking the queues. Any add/reclaim that races with this iteration will set the - # event again, so the wait at the end of the loop returns immediately rather than blocking until the - # throttle expires. - self._new_work_event.clear() - - now = datetime.now(timezone.utc) - available_domains = sorted( - ( - domain - for domain, state in self._domain_states.items() - if domain in self._sub_managers and now >= state.throttled_until - ), - key=lambda d: self._domain_states[d].throttled_until, - ) - - for domain in available_domains: - req = await self._sub_managers[domain].fetch_next_request() - if req: - self._mark_domain_dispatched(domain) - return req - - request = await self._inner.fetch_next_request() - if request is not None: + for domain in self._fetchable_domains(): + request = await self._sub_managers[domain].fetch_next_request() + if request: + self._mark_domain_dispatched(domain) return request - if not self._sub_managers: - return None - - sub_managers_empty = await asyncio.gather(*(sm.is_empty() for sm in self._sub_managers.values())) - if all(sub_managers_empty): - return None - - earliest = self._get_earliest_available_time(now) - sleep_duration = max( - (earliest - now).total_seconds(), - 0.1, # Avoid tight loops if a throttle expired during the previous iteration. - ) - logger.debug( - f'All configured domains are throttled and inner manager is empty. ' - f'Waiting up to {sleep_duration:.1f}s for earliest domain to become available or new work.' - ) - await self._wait_for_new_work_or_timeout(sleep_duration) + return await self._inner.fetch_next_request() @override async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None: manager = self._select_manager(request.url) - result = await manager.reclaim_request(request, forefront=forefront) - self._signal_new_work() - return result + return await manager.reclaim_request(request, forefront=forefront) @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: @@ -278,7 +228,13 @@ async def get_total_count(self) -> int: @override async def is_empty(self) -> bool: - results = await asyncio.gather(self._inner.is_empty(), *(sm.is_empty() for sm in self._sub_managers.values())) + """Report whether anything can be dispatched right now. + + Requests queued for a domain in a cooldown do not count. They still count towards `is_finished`, so the crawl + waits for them. + """ + fetchable = (self._sub_managers[domain] for domain in self._fetchable_domains()) + results = await asyncio.gather(self._inner.is_empty(), *(sm.is_empty() for sm in fetchable)) return all(results) @override @@ -385,15 +341,18 @@ def _is_domain_throttled(self, domain: str) -> bool: return False return datetime.now(timezone.utc) < state.throttled_until - def _get_earliest_available_time(self, now: datetime) -> datetime: - """Get the earliest time any throttled domain becomes available.""" - earliest = now + self._max_delay - - for state in self._domain_states.values(): - if now < state.throttled_until < earliest: - earliest = state.throttled_until - - return earliest + def _fetchable_domains(self) -> list[str]: + """Return the configured domains that are not in a cooldown right now, longest-overdue first.""" + now = datetime.now(timezone.utc) + available = [ + domain + for domain, state in self._domain_states.items() + # The sub-manager check guards the lookups in the callers: a domain can be given throttle state before its + # first request creates the sub-manager. + if domain in self._sub_managers and now >= state.throttled_until + ] + available.sort(key=lambda domain: self._domain_states[domain].throttled_until) + return available def _mark_domain_dispatched(self, domain: str) -> None: """Record that a request to this domain was just dispatched. @@ -404,10 +363,6 @@ def _mark_domain_dispatched(self, domain: str) -> None: if state is not None and state.crawl_delay is not None: state.throttled_until = datetime.now(timezone.utc) + state.crawl_delay - def _signal_new_work(self) -> None: - """Wake `fetch_next_request` if it is sleeping inside a throttle wait.""" - self._new_work_event.set() - def _select_manager(self, url: str) -> TRequestManager: """Return the manager that owns the given URL — its sub-manager if one exists, otherwise the inner.""" domain = self._extract_domain(url) @@ -415,16 +370,6 @@ def _select_manager(self, url: str) -> TRequestManager: return self._sub_managers[domain] return self._inner - async def _wait_for_new_work_or_timeout(self, timeout: float) -> None: - """Wait until new work is signaled or `timeout` seconds elapse, whichever comes first. - - The signal is set by `add_request`, `add_requests`, and `reclaim_request`, allowing `fetch_next_request` to wake - up immediately when fresh work appears during a throttle wait instead of sleeping for the full computed cooldown - (up to `max_delay`). - """ - with contextlib.suppress(asyncio.TimeoutError): - await asyncio.wait_for(self._new_work_event.wait(), timeout=timeout) - class _RequestManagerOpener(Protocol[TRequestManager]): """Callable that opens a `RequestManager` instance. diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index 0451297fff..ea8ef21805 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -4,8 +4,6 @@ import asyncio from datetime import datetime, timedelta, timezone -from typing import Any -from unittest.mock import AsyncMock, patch import pytest @@ -17,6 +15,7 @@ from crawlee.storages import RequestQueue THROTTLED_DOMAIN = 'throttled.com' +SECOND_THROTTLED_DOMAIN = 'slow.com' NON_THROTTLED_DOMAIN = 'free.com' TEST_DOMAINS = [THROTTLED_DOMAIN] @@ -50,6 +49,20 @@ async def manager(inner_queue: RequestQueue, service_locator: ServiceLocator) -> ) +@pytest.fixture +async def two_domain_manager( + inner_queue: RequestQueue, + service_locator: ServiceLocator, +) -> ThrottlingRequestManager[RequestQueue]: + """Create a ThrottlingRequestManager with two throttled domains.""" + return ThrottlingRequestManager( + inner_queue, + domains=[THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN], + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + def _make_request(url: str) -> Request: """Helper to create a Request object.""" return Request.from_url(url) @@ -307,66 +320,73 @@ async def test_fetch_skips_throttled_sub_manager( assert result.url == free_url -async def test_sleep_when_all_throttled(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """When all domains are throttled and inner is empty, should wait and retry.""" +async def test_fetch_returns_none_when_all_throttled(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """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_domain_delay(url, retry_after=timedelta(seconds=10)) + assert await asyncio.wait_for(manager.fetch_next_request(), timeout=1.0) is None - target = ( - 'crawlee.request_loaders._throttling_request_manager.ThrottlingRequestManager._wait_for_new_work_or_timeout' - ) - 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 - # clock resolution or scheduling jitter on slow CI runners. - manager._domain_states[THROTTLED_DOMAIN].throttled_until = datetime.now(timezone.utc) - timedelta(seconds=1) +async def test_throttled_domain_reads_as_empty_but_not_finished( + manager: ThrottlingRequestManager[RequestQueue], +) -> None: + """Requests waiting out a cooldown are nothing to dispatch, but the crawl must not end on them either.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) - mock_wait.side_effect = wait_side_effect + assert await manager.is_empty() is True + assert await manager.is_finished() is False - result = await manager.fetch_next_request() - mock_wait.assert_called() - assert result is not None - assert result.url == url +async def test_crawl_delay_hides_queued_requests(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A crawl-delay armed on dispatch keeps the domain's remaining requests out of `is_empty`.""" + manager.set_crawl_delay(f'https://{THROTTLED_DOMAIN}/', 30) + await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1') + await manager.add_request(f'https://{THROTTLED_DOMAIN}/page2') + assert await manager.fetch_next_request() is not None -async def test_fetch_wakes_when_request_added_during_throttle_wait( + assert await manager.is_empty() is True + assert await manager.is_finished() is False + + +async def test_expired_throttle_makes_the_domain_fetchable_again( manager: ThrottlingRequestManager[RequestQueue], ) -> None: - """When all sub-managers are throttled and inner is empty, fetch should wake up immediately - when a new request is added rather than blocking until the throttle expires.""" - # Throttle the only configured domain for a long time so a naive sleep would block here. - throttled_url = f'https://{THROTTLED_DOMAIN}/page1' - await manager.add_request(throttled_url) - manager.record_domain_delay(throttled_url, retry_after=timedelta(seconds=60)) - - free_url = f'https://{NON_THROTTLED_DOMAIN}/page1' + """Once the cooldown passes, the domain counts again for both dispatching and emptiness.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + assert await manager.is_empty() is True - # Wrap the wait helper so we can synchronize with the moment fetch enters the wait state. - wait_entered = asyncio.Event() - original_wait = manager._wait_for_new_work_or_timeout + manager._domain_states[THROTTLED_DOMAIN].throttled_until = datetime.now(timezone.utc) - timedelta(seconds=1) - async def signaling_wait(timeout: float) -> None: - wait_entered.set() - await original_wait(timeout) + assert await manager.is_empty() is False + result = await manager.fetch_next_request() + assert result is not None + assert result.url == url - manager._wait_for_new_work_or_timeout = signaling_wait # ty: ignore[invalid-assignment] - fetch_task = asyncio.create_task(manager.fetch_next_request()) +async def test_fetch_prefers_longest_overdue_domain( + two_domain_manager: ThrottlingRequestManager[RequestQueue], +) -> None: + """With several domains free, the one whose cooldown expired earliest is dispatched first.""" + recent_url = f'https://{THROTTLED_DOMAIN}/page1' + overdue_url = f'https://{SECOND_THROTTLED_DOMAIN}/page1' + await two_domain_manager.add_request(recent_url) + await two_domain_manager.add_request(overdue_url) - # Wait until fetch is suspended inside the wait, then add fresh non-throttled work. - await wait_entered.wait() - await manager.add_request(free_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) - # If the wake-up signal works, fetch returns the freshly-added request well within the 2s wait_for budget; otherwise - # it would still be blocked on the 60s throttle. - result = await asyncio.wait_for(fetch_task, timeout=2.0) + result = await two_domain_manager.fetch_next_request() assert result is not None - assert result.url == free_url + assert result.url == overdue_url # ── Delegation Tests ──────────────────────────────────── From f4d84c557ba22d9c9a19bcd49bcafc4f01e29aab Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 20 Aug 2026 13:59:37 +0200 Subject: [PATCH 2/4] docs: describe concurrency slot release in the request throttling guide --- docs/guides/request_throttling.mdx | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/guides/request_throttling.mdx b/docs/guides/request_throttling.mdx index e4c99a6c8d..9afcc8b3e2 100644 --- a/docs/guides/request_throttling.mdx +++ b/docs/guides/request_throttling.mdx @@ -18,7 +18,7 @@ The `ThrottlingRequestManager` Date: Thu, 20 Aug 2026 13:59:38 +0200 Subject: [PATCH 3/4] refactor(throttling): clarify the is_empty and is_finished contract --- .../_throttling_request_manager.py | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index 2775b4d563..b86d928b69 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -39,7 +39,9 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): `fetch_next_request()` takes from the sub-manager whose domain has been waiting the longest, skipping domains in a cooldown, and falls back to the inner manager when no sub-manager yields a request. If nothing can be dispatched - right now, it returns `None` rather than waiting, so the caller's task slot is released. + right now, it returns `None` rather than waiting, so the caller's task slot is released. `is_empty()` reports the + same view and reads as empty while every remaining request sits in a cooldown, whereas `is_finished()` counts those + requests, so the crawl idles until they are dispatchable instead of ending early. Delay sources: - HTTP 429 responses (via `record_domain_delay`) @@ -189,12 +191,18 @@ async def add_requests( async def fetch_next_request(self) -> Request | None: """Fetch the next request, respecting per-domain delays. - Sub-managers are checked in order of longest-overdue domain first, then the inner manager. Throttled domains - are skipped rather than waited for, so a caller holding a concurrency slot gets it back instead of sleeping. + Sub-managers are checked in order of longest-overdue domain first, then the inner manager. Domains in a + cooldown are skipped, so the call returns `None` when nothing is dispatchable right now. + + Note: + Unlike the `RequestLoader.fetch_next_request` contract, a `None` result does not imply that `is_finished()` + is `True` - it only means nothing is dispatchable right now. Since the manager never waits out a cooldown + itself, the dispatch cadence is only as precise as the caller's polling interval: a cooldown expiring + between two polls is picked up on the next one. """ for domain in self._fetchable_domains(): request = await self._sub_managers[domain].fetch_next_request() - if request: + if request is not None: self._mark_domain_dispatched(domain) return request @@ -233,8 +241,9 @@ async def is_empty(self) -> bool: Requests queued for a domain in a cooldown do not count. They still count towards `is_finished`, so the crawl waits for them. """ - fetchable = (self._sub_managers[domain] for domain in self._fetchable_domains()) - results = await asyncio.gather(self._inner.is_empty(), *(sm.is_empty() for sm in fetchable)) + results = await asyncio.gather( + self._inner.is_empty(), *(self._sub_managers[d].is_empty() for d in self._fetchable_domains()) + ) return all(results) @override @@ -347,8 +356,8 @@ def _fetchable_domains(self) -> list[str]: available = [ domain for domain, state in self._domain_states.items() - # The sub-manager check guards the lookups in the callers: a domain can be given throttle state before its - # first request creates the sub-manager. + # Every configured domain has state from construction, but sub-managers are created lazily on first + # insertion, so this check keeps the `_sub_managers[domain]` lookups in the callers safe. if domain in self._sub_managers and now >= state.throttled_until ] available.sort(key=lambda domain: self._domain_states[domain].throttled_until) From 8840a33e8513a6ebcf1be5d2879f173b501d705a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 20 Aug 2026 13:59:39 +0200 Subject: [PATCH 4/4] test(throttling): cover a throttled crawl finishing without ending early --- .../crawlers/_basic/test_basic_crawler.py | 35 +++++++++++++++++++ tests/unit/test_throttling_request_manager.py | 2 +- 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 56ba257e86..416ae257a2 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -2502,3 +2502,38 @@ 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) + + +async def test_throttled_domain_waits_out_backoff_without_ending_the_crawl() -> None: + """A domain in a cooldown reads as empty to the autoscaled pool, yet its queued requests are still crawled.""" + storage_client = MemoryStorageClient() + # The throttler opens its sub-managers through the global service locator, so point that at the same client. + service_locator.set_storage_client(storage_client) + inner = await RequestQueue.open(name='test-inner-backoff', storage_client=storage_client) + throttler = ThrottlingRequestManager( + inner, + domains=['throttled.placeholder.com'], + request_manager_opener=RequestQueue.open, + ) + # A single worker slot makes the ordering deterministic: the second dispatch cannot start before the first + # handler has armed the backoff. + crawler = BasicCrawler( + request_manager=throttler, + configure_logging=False, + concurrency_settings=ConcurrencySettings(desired_concurrency=1, max_concurrency=1), + ) + dispatched_at = list[float]() + empty_during_cooldown = list[bool]() + + @crawler.router.default_handler + 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)) + empty_during_cooldown.append(await throttler.is_empty()) + + await crawler.run(['https://throttled.placeholder.com/a', 'https://throttled.placeholder.com/b']) + + assert empty_during_cooldown == [True] + assert len(dispatched_at) == 2 + assert dispatched_at[1] - dispatched_at[0] >= 0.5 diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index ea8ef21805..e49a6891d7 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -489,7 +489,7 @@ async def test_get_total_count_aggregates(manager: ThrottlingRequestManager[Requ async def test_is_empty_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """is_empty should return False if any manager has requests.""" + """is_empty should return False if any manager that is not in a cooldown has requests.""" assert await manager.is_empty() is True await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1')