From d80c60c93da144d5e6c54f8f41f0b873d7a5af2c Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Thu, 15 Oct 2026 21:45:18 +0000 Subject: [PATCH 1/4] replace `is_empty` and `is_finished` with `check_readiness` --- .../code_examples/run_parallel_crawlers.py | 3 +- docs/guides/request_loaders.mdx | 3 +- docs/upgrading/upgrading_to_v2.md | 29 +++ src/crawlee/crawlers/_basic/_basic_crawler.py | 15 +- src/crawlee/request_loaders/__init__.py | 14 ++ src/crawlee/request_loaders/_request_list.py | 16 +- .../request_loaders/_request_loader.py | 17 +- .../_request_manager_tandem.py | 36 ++- .../_sitemap_request_loader.py | 29 ++- .../_throttling_request_manager.py | 124 +++++----- src/crawlee/request_loaders/_types.py | 49 ++++ src/crawlee/request_loaders/_utils.py | 38 +++ .../_base/_request_queue_client.py | 3 +- .../_file_system/_request_queue_client.py | 6 +- .../_memory/_request_queue_client.py | 4 + src/crawlee/storages/_request_queue.py | 39 ++-- .../crawlers/_basic/test_basic_crawler.py | 83 ++++--- .../unit/crawlers/_http/test_http_crawler.py | 21 +- .../unit/request_loaders/test_request_list.py | 48 ++-- .../test_request_source_status.py | 60 +++++ .../test_sitemap_request_loader.py | 87 ++++--- .../storages/test_request_manager_tandem.py | 62 ++++- tests/unit/storages/test_request_queue.py | 97 ++++---- tests/unit/test_throttling_request_manager.py | 220 +++++++++++------- 24 files changed, 720 insertions(+), 383 deletions(-) create mode 100644 src/crawlee/request_loaders/_types.py create mode 100644 src/crawlee/request_loaders/_utils.py create mode 100644 tests/unit/request_loaders/test_request_source_status.py diff --git a/docs/examples/code_examples/run_parallel_crawlers.py b/docs/examples/code_examples/run_parallel_crawlers.py index 5ce94a58fa..608c45a968 100644 --- a/docs/examples/code_examples/run_parallel_crawlers.py +++ b/docs/examples/code_examples/run_parallel_crawlers.py @@ -7,6 +7,7 @@ PlaywrightCrawler, PlaywrightCrawlingContext, ) +from crawlee.request_loaders import FinishedStatus from crawlee.sessions import SessionPool from crawlee.storages import RequestQueue @@ -79,7 +80,7 @@ async def handle_parsel(context: ParselCrawlingContext) -> None: await parsel_crawler.run(['https://crawlee.dev/blog']) # Wait for the Playwright crawler to finish processing all requests - while not await playwright_rq.is_empty(): + while not isinstance(await playwright_rq.check_readiness(), FinishedStatus): playwright_crawler.log.info('Waiting for Playwright crawler to finish...') await asyncio.sleep(5) diff --git a/docs/guides/request_loaders.mdx b/docs/guides/request_loaders.mdx index a63393142d..f2dd135fd3 100644 --- a/docs/guides/request_loaders.mdx +++ b/docs/guides/request_loaders.mdx @@ -62,8 +62,7 @@ class RequestLoader { + total_count + fetch_next_request() + mark_request_as_handled() - + is_empty() - + is_finished() + + check_readiness() + to_tandem() } diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 894d2a8567..ff42ed1b08 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -167,3 +167,32 @@ current concurrency 3, desired 5, overloaded: cpu 58% of the time (limit 40%) To read the concurrency values in code, use `crawler.concurrency_system`. The autoscaling loggers moved together with the package. `crawlee._autoscaling.autoscaled_pool`, `crawlee._autoscaling.snapshotter` and `crawlee._autoscaling.system_status` are now `crawlee.autoscaling._autoscaled_pool`, `crawlee.autoscaling._snapshotter` and `crawlee.autoscaling._system_status`. The state log, along with the messages about scaling and free capacity, now comes from the new `crawlee.autoscaling._concurrency_system` logger. To configure all of them at once, use the `crawlee.autoscaling` logger. + +## `check_readiness` replaces `is_empty` and `is_finished` + +`RequestLoader` and `RequestManager` no longer have `is_empty` and `is_finished`. A single `check_readiness` method replaces them and returns one of four statuses: + +- `ReadyStatus`: the next `fetch_next_request` call is expected to return a request. +- `WaitingStatus`: nothing can be fetched right now, but the source isn't done. Requests are in progress, are being added in the background, or are held back until `ready_at`. +- `StalledStatus`: the source holds requests it can't make progress on, as explained in `reason`. `ThrottlingRequestManager` reports it, and `RequestManagerTandem` passes it on from the manager it wraps. +- `FinishedStatus`: every request has been handled. + +The change covers `RequestQueue`, `RequestList`, `SitemapRequestLoader`, `RequestManagerTandem` and `ThrottlingRequestManager`. `ThrottlingRequestManager.get_stall_reason` is removed too, since a stall is now reported as `StalledStatus`. Storage clients keep `is_empty` and `is_finished`. + +```python +from crawlee.request_loaders import FinishedStatus, ReadyStatus + +# v1 +has_requests = not await rq.is_empty() +done = await rq.is_finished() + +# v2 +has_requests = isinstance(await rq.check_readiness(), ReadyStatus) +done = isinstance(await rq.check_readiness(), FinishedStatus) +``` + +If you have a custom request loader or manager, replace `is_empty` and `is_finished` with `check_readiness`. Return `ReadyStatus` as soon as a request can be fetched, before checking anything else, since the crawler calls the method often. Never return `FinishedStatus` while requests are still pending or in progress. A loader doesn't return `StalledStatus`. + +A `None` from `fetch_next_request` no longer means the loader is finished. It only means nothing is pending right now. `RequestManagerTandem` now moves a request from its loader only when the loader is ready, so a `SitemapRequestLoader` that's still loading sitemaps no longer holds up the requests already in the manager. + +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. diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index c7af459092..cfcb9fe81f 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/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 FinishedStatus, ReadyStatus, StalledStatus, ThrottlingRequestManager from crawlee.router import Router from crawlee.sessions import SessionPool from crawlee.statistics import Statistics, StatisticsState @@ -1499,22 +1499,23 @@ async def __is_finished_function(self) -> bool: if self._keep_alive: return False + request_manager = await self.get_request_manager() + status = await request_manager.check_readiness() + # Check for a stall only when the pool is idle. An in-flight request can still enqueue new work or end the # stall. The pool's own requests are what counts, not those of another crawler sharing the budget. if ( - isinstance(self._request_manager, ThrottlingRequestManager) + isinstance(status, StalledStatus) and self._autoscaled_pool is not None and self._autoscaled_pool.own_concurrency == 0 - and (reason := await self._request_manager.get_stall_reason()) is not None ): - error = PersistentRateLimitError(f'Giving up: {reason}') + error = PersistentRateLimitError(f'Giving up: {status.reason}') self._logger.error( 'Giving up the crawl because a domain keeps rate-limiting every request.', exc_info=error ) raise error - request_manager = await self.get_request_manager() - return await request_manager.is_finished() + return isinstance(status, FinishedStatus) async def __is_task_ready_function(self) -> bool: self._stop_if_max_requests_count_exceeded() @@ -1526,7 +1527,7 @@ async def __is_task_ready_function(self) -> bool: return False request_manager = await self.get_request_manager() - return not await request_manager.is_empty() + return isinstance(await request_manager.check_readiness(), ReadyStatus) async def __run_task_function(self, task_id: str) -> None: # noqa: ARG002 request_manager = await self.get_request_manager() diff --git a/src/crawlee/request_loaders/__init__.py b/src/crawlee/request_loaders/__init__.py index 6dd8cccfab..3237ea5b72 100644 --- a/src/crawlee/request_loaders/__init__.py +++ b/src/crawlee/request_loaders/__init__.py @@ -4,12 +4,26 @@ from ._request_manager_tandem import RequestManagerTandem from ._sitemap_request_loader import SitemapRequestLoader from ._throttling_request_manager import ThrottlingRequestManager +from ._types import ( + FinishedStatus, + ReadyStatus, + RequestLoaderStatus, + RequestSourceStatus, + StalledStatus, + WaitingStatus, +) __all__ = [ + 'FinishedStatus', + 'ReadyStatus', 'RequestList', 'RequestLoader', + 'RequestLoaderStatus', 'RequestManager', 'RequestManagerTandem', + 'RequestSourceStatus', 'SitemapRequestLoader', + 'StalledStatus', 'ThrottlingRequestManager', + 'WaitingStatus', ] diff --git a/src/crawlee/request_loaders/_request_list.py b/src/crawlee/request_loaders/_request_list.py index cd8893d4ce..18d2f71d85 100644 --- a/src/crawlee/request_loaders/_request_list.py +++ b/src/crawlee/request_loaders/_request_list.py @@ -4,7 +4,7 @@ import contextlib from collections.abc import AsyncGenerator, AsyncIterable, AsyncIterator, Iterable from logging import getLogger -from typing import Annotated +from typing import TYPE_CHECKING, Annotated from pydantic import BaseModel, ConfigDict, Field, ValidationError from pydantic.alias_generators import to_camel @@ -14,6 +14,11 @@ from crawlee._utils.docs import docs_group from crawlee.request_loaders._request_loader import RequestLoader +from ._types import FinishedStatus, ReadyStatus, WaitingStatus + +if TYPE_CHECKING: + from ._types import RequestLoaderStatus + logger = getLogger(__name__) @@ -155,14 +160,13 @@ async def get_total_count(self) -> int: return self._assumed_total_count @override - async def is_empty(self) -> bool: + async def check_readiness(self) -> RequestLoaderStatus: await self._ensure_next_request() - return self._next[0] is None + if self._next[0] is not None: + return ReadyStatus() - @override - async def is_finished(self) -> bool: state = await self._get_state() - return len(state.in_progress) == 0 and await self.is_empty() + return WaitingStatus() if state.in_progress else FinishedStatus() @override async def fetch_next_request(self) -> Request | None: diff --git a/src/crawlee/request_loaders/_request_loader.py b/src/crawlee/request_loaders/_request_loader.py index 200339a46d..a0cac2fbc2 100644 --- a/src/crawlee/request_loaders/_request_loader.py +++ b/src/crawlee/request_loaders/_request_loader.py @@ -12,6 +12,8 @@ from crawlee.request_loaders import RequestManager, RequestManagerTandem from crawlee.storage_clients.models import ProcessedRequest + from ._types import RequestSourceStatus + @docs_group('Request loaders') class RequestLoader(ABC): @@ -34,19 +36,18 @@ async def get_total_count(self) -> int: """Get an offline approximation of the total number of requests in the loader (i.e. pending + handled).""" @abstractmethod - async def is_empty(self) -> bool: - """Return True if there are no more requests in the loader (there might still be unfinished requests).""" + async def check_readiness(self) -> RequestSourceStatus: + """Report whether the loader has a request to hand over, is waiting for one, or is done. - @abstractmethod - async def is_finished(self) -> bool: - """Return True if all requests have been handled.""" + The crawler's task loop calls this often, so answer `ReadyStatus` before evaluating anything else. + `FinishedStatus` may arrive late, but never early. + """ @abstractmethod async def fetch_next_request(self) -> Request | None: - """Return the next request to be processed, or `None` if there are no more pending requests. + """Return the next request to be processed, or `None` if no request is pending right now. - The method should return `None` if and only if `is_finished` would return `True`. In other cases, the method - should wait until a request appears. + `None` doesn't mean the loader is finished. Use `check_readiness` to tell a finished loader from a waiting one. """ @abstractmethod diff --git a/src/crawlee/request_loaders/_request_manager_tandem.py b/src/crawlee/request_loaders/_request_manager_tandem.py index a605c25d25..8a992dc6b4 100644 --- a/src/crawlee/request_loaders/_request_manager_tandem.py +++ b/src/crawlee/request_loaders/_request_manager_tandem.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from datetime import timedelta from logging import getLogger from typing import TYPE_CHECKING @@ -9,6 +10,9 @@ from crawlee._utils.docs import docs_group from crawlee.request_loaders import RequestManager +from ._types import ReadyStatus +from ._utils import join_request_source_statuses + if TYPE_CHECKING: from collections.abc import Sequence @@ -16,6 +20,8 @@ from crawlee.request_loaders import RequestLoader from crawlee.storage_clients.models import ProcessedRequest + from ._types import RequestSourceStatus + logger = getLogger(__name__) @@ -41,12 +47,11 @@ async def get_total_count(self) -> int: return (await self._read_only_loader.get_total_count()) + (await self._read_write_manager.get_total_count()) @override - async def is_empty(self) -> bool: - return (await self._read_only_loader.is_empty()) and (await self._read_write_manager.is_empty()) - - @override - async def is_finished(self) -> bool: - return (await self._read_only_loader.is_finished()) and (await self._read_write_manager.is_finished()) + async def check_readiness(self) -> RequestSourceStatus: + loader_status, manager_status = await asyncio.gather( + self._read_only_loader.check_readiness(), self._read_write_manager.check_readiness() + ) + return join_request_source_statuses(loader_status, manager_status) @override async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None: @@ -74,13 +79,22 @@ async def add_requests( @override async def fetch_next_request(self) -> Request | None: - if await self._read_only_loader.is_finished(): - return await self._read_write_manager.fetch_next_request() + # Only the loader's own readiness decides the transfer, so a loader that's still loading doesn't block the + # manager's requests. + if isinstance(await self._read_only_loader.check_readiness(), ReadyStatus): + transferred = await self._transfer_next_request() + # The request was dropped, so skip the manager this round. The next call picks up where this one left off. + if not transferred: + return None + + return await self._read_write_manager.fetch_next_request() + async def _transfer_next_request(self) -> bool: + """Move the loader's next request into the manager, returning `False` if it had to be dropped.""" request = await self._read_only_loader.fetch_next_request() if not request: - return await self._read_write_manager.fetch_next_request() + return True try: await self._read_write_manager.add_request(request, forefront=True) @@ -90,13 +104,13 @@ async def fetch_next_request(self) -> Request | None: extra={'url': request.url, 'unique_key': request.unique_key}, ) - return None + return False finally: # Mark it as processed so that the `request` doesn't get stuck in the `in_progress` status # in `RequestLoader` await self._read_only_loader.mark_request_as_handled(request) - return await self._read_write_manager.fetch_next_request() + return True @override async def reclaim_request(self, request: Request, *, forefront: bool = False) -> None: diff --git a/src/crawlee/request_loaders/_sitemap_request_loader.py b/src/crawlee/request_loaders/_sitemap_request_loader.py index 4d07c089d5..4b4a47c3b3 100644 --- a/src/crawlee/request_loaders/_sitemap_request_loader.py +++ b/src/crawlee/request_loaders/_sitemap_request_loader.py @@ -26,6 +26,8 @@ from crawlee._utils.urls import filter_url from crawlee.request_loaders._request_loader import RequestLoader +from ._types import FinishedStatus, ReadyStatus, WaitingStatus + if TYPE_CHECKING: import re from collections.abc import Callable, Sequence @@ -37,6 +39,8 @@ from crawlee.proxy_configuration import ProxyInfo from crawlee.storage_clients.models import ProcessedRequest + from ._types import RequestLoaderStatus + logger = getLogger(__name__) @@ -206,23 +210,26 @@ async def get_handled_count(self) -> int: return state.handled_count @override - async def is_empty(self) -> bool: - """Check if there are no more URLs to process.""" + async def check_readiness(self) -> RequestLoaderStatus: state = await self._get_state() - return not state.url_queue + if state.url_queue: + return ReadyStatus() - @override - async def is_finished(self) -> bool: - """Check if all URLs have been processed.""" - state = await self._get_state() - return not state.url_queue and len(state.in_progress) == 0 and self._loading_task.done() + # More URLs may arrive while the sitemaps are still being loaded. + if not self._loading_task.done(): + return WaitingStatus() + + return WaitingStatus() if state.in_progress else FinishedStatus() @override async def fetch_next_request(self) -> Request | None: """Fetch the next request to process.""" - while not (await self.is_finished()): + while True: state = await self._get_state() if not state.url_queue: + # More URLs may arrive while the sitemaps are still being loaded. + if self._loading_task.done(): + return None await asyncio.sleep(0.1) continue @@ -250,8 +257,6 @@ async def fetch_next_request(self) -> Request | None: return request - return None - @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: """Mark a request as successfully handled.""" @@ -449,6 +454,6 @@ async def _load_sitemaps(self) -> None: state.completed = True except Exception: - # Not re-raised: nothing retrieves the task's exception (`is_finished` only checks doneness), so it + # Not re-raised: nothing retrieves the task's exception (`check_readiness` only checks doneness), so it # would only resurface as an "exception was never retrieved" asyncio error when the task is collected. logger.exception('Error loading sitemaps') diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index 60e09d5f1b..a98bb73483 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -15,6 +15,9 @@ from crawlee._utils.docs import docs_group from crawlee.request_loaders._request_manager import RequestManager +from ._types import FinishedStatus, ReadyStatus, StalledStatus, WaitingStatus +from ._utils import join_request_source_statuses + if TYPE_CHECKING: from collections.abc import Sequence @@ -23,6 +26,8 @@ from crawlee.storage_clients import StorageClient from crawlee.storage_clients.models import ProcessedRequest + from ._types import RequestSourceStatus + logger = getLogger(__name__) TRequestManager = TypeVar('TRequestManager', bound=RequestManager) @@ -53,9 +58,8 @@ 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. `is_empty()` reads as - empty while every remaining request sits in a cooldown or belongs to a stalled domain, whereas `is_finished()` - counts those requests, so the crawl idles until they are dispatchable instead of ending early. + right now, it returns `None` rather than waiting, so the caller's task slot is released. `check_readiness()` then + reports the crawl as waiting until the earliest cooldown ends, so it idles instead of ending early. Delay sources: - HTTP 429 responses (via `record_domain_delay`) @@ -248,10 +252,8 @@ async def fetch_next_request(self) -> Request | None: 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. + 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. """ await self._ensure_sub_managers() @@ -304,43 +306,17 @@ async def get_total_count(self) -> int: return sum(counts) - self._migrated_from_inner @override - async def is_empty(self) -> bool: - """Report whether anything can be dispatched right now. - - Requests queued for a domain in a cooldown don't count. They still count towards `is_finished`, so the crawl - waits for them. Requests of a domain that has rate-limited every request for longer than `max_domain_stall` - don't count either. - """ - await self._ensure_sub_managers() - now = datetime.now(UTC) - results = await asyncio.gather( - self._inner.is_empty(), - *( - self._sub_managers[d].is_empty() - for d in self._fetchable_domains() - if not self._domain_states[d].is_stall_candidate(now, self._max_domain_stall) - ), - ) - return all(results) - - @override - async def is_finished(self) -> bool: - await self._ensure_sub_managers() - results = await asyncio.gather( - self._inner.is_finished(), *(sm.is_finished() for sm in self._sub_managers.values()) - ) - return all(results) - - async def get_stall_reason(self) -> str | None: - """Explain why the crawl can't make progress, or return `None` if it can. + async def check_readiness(self) -> RequestSourceStatus: + """Report whether anything can be dispatched now, and otherwise when it can be or why it never will. - A domain is stalled once it has rate-limited every request for longer than `max_domain_stall`. It's reported - only when nothing else can be dispatched and no domain in a cooldown has requests left. + Requests of a domain in a cooldown count as outstanding work, so the crawl waits until the earliest cooldown + ends. A domain is stalled once it has rate-limited every request for longer than `max_domain_stall`. It's + reported only when nothing else can be dispatched and no domain in a cooldown has requests left. """ await self._ensure_sub_managers() now = datetime.now(UTC) dispatchable: list[RequestManager] = [self._inner] - cooling: list[RequestManager] = [] + cooling: list[str] = [] candidates: list[str] = [] for domain, state in self._domain_states.items(): @@ -351,36 +327,56 @@ async def get_stall_reason(self) -> str | None: elif now >= state.throttled_until: dispatchable.append(self._sub_managers[domain]) else: - cooling.append(self._sub_managers[domain]) + cooling.append(domain) - if not candidates: - return None + status: RequestSourceStatus = FinishedStatus() + for manager_status in await asyncio.gather(*(manager.check_readiness() for manager in dispatchable)): + status = join_request_source_statuses(status, manager_status) + if isinstance(status, ReadyStatus): + return status - all_dispatchable_empty = all(await asyncio.gather(*(manager.is_empty() for manager in dispatchable))) - if not all_dispatchable_empty: - return None + cooling_statuses = await asyncio.gather(*(self._sub_managers[domain].check_readiness() for domain in cooling)) - # A domain waiting out a cooldown with work left will still make progress, so the crawl isn't stuck yet. - all_cooling_finished = all(await asyncio.gather(*(manager.is_finished() for manager in cooling))) - if not all_cooling_finished: - return None - - candidates_empty = await asyncio.gather(*(self._sub_managers[domain].is_empty() for domain in candidates)) - stalled = [domain for domain, is_empty in zip(candidates, candidates_empty, strict=True) if not is_empty] - if not stalled: - return None - - summary = ', '.join( - f'"{domain}" ({(now - since).total_seconds():.0f}s)' - for domain in stalled - if (since := self._domain_states[domain].rate_limited_since) is not None + ready_at = min( + ( + self._domain_states[domain].throttled_until + for domain, cooling_status in zip(cooling, cooling_statuses, strict=True) + if not isinstance(cooling_status, FinishedStatus) + ), + default=None, ) - window = self._max_domain_stall.total_seconds() - return ( - f'{summary} rate-limited every request for longer than `max_domain_stall` ({window:.0f}s). Waiting ' - "longer will not help - lower the crawler's concurrency, or drop these domains. Their requests are " - 'still queued, so re-running with `purge_on_start` disabled will resume them if the rate limit lifts.' + + # A domain waiting out a cooldown with work left will still make progress, so the crawl isn't stuck yet. + if ready_at is not None: + return join_request_source_statuses(status, WaitingStatus(ready_at=ready_at)) + + candidate_statuses = await asyncio.gather( + *(self._sub_managers[domain].check_readiness() for domain in candidates) ) + stalled = [ + domain + for domain, candidate_status in zip(candidates, candidate_statuses, strict=True) + if isinstance(candidate_status, ReadyStatus) + ] + + if stalled: + summary = ', '.join( + f'"{domain}" ({(now - since).total_seconds():.0f}s)' + for domain in stalled + if (since := self._domain_states[domain].rate_limited_since) is not None + ) + window = self._max_domain_stall.total_seconds() + return StalledStatus( + f'{summary} rate-limited every request for longer than `max_domain_stall` ({window:.0f}s). Waiting ' + "longer will not help - lower the crawler's concurrency, or drop these domains. Their requests are " + 'still queued, so re-running with `purge_on_start` disabled will resume them if the rate limit lifts.' + ) + + # A candidate's requests that are in progress or still being added in the background aren't finished yet. + for candidate_status in candidate_statuses: + status = join_request_source_statuses(status, candidate_status) + + return status 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. diff --git a/src/crawlee/request_loaders/_types.py b/src/crawlee/request_loaders/_types.py new file mode 100644 index 0000000000..80faf687ad --- /dev/null +++ b/src/crawlee/request_loaders/_types.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING + +from crawlee._utils.docs import docs_group + +if TYPE_CHECKING: + from datetime import datetime + + +@docs_group('Request loaders') +@dataclass(frozen=True) +class ReadyStatus: + """The next `fetch_next_request` call is expected to return a request.""" + + +@docs_group('Request loaders') +@dataclass(frozen=True, kw_only=True) +class WaitingStatus: + """Nothing to fetch right now, but the source isn't done. + + Requests are in progress, are being added in the background, or are held back until `ready_at`. + """ + + ready_at: datetime | None = None + """When the source expects to become ready, as a timezone-aware UTC time. `None` when the wait has no clock.""" + + +@docs_group('Request loaders') +@dataclass(frozen=True) +class StalledStatus: + """The source holds requests it can't make progress on. Only a manager that paces its own dispatch reports it.""" + + reason: str + """Why the source can't make progress.""" + + +@docs_group('Request loaders') +@dataclass(frozen=True) +class FinishedStatus: + """Every request has been handled.""" + + +RequestLoaderStatus = ReadyStatus | WaitingStatus | FinishedStatus +"""Readiness of a request loader. Loaders never stall.""" + +RequestSourceStatus = ReadyStatus | WaitingStatus | StalledStatus | FinishedStatus +"""Readiness of a request source, as reported by `RequestLoader.check_readiness`.""" diff --git a/src/crawlee/request_loaders/_utils.py b/src/crawlee/request_loaders/_utils.py new file mode 100644 index 0000000000..3ab98a1870 --- /dev/null +++ b/src/crawlee/request_loaders/_utils.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from ._types import ReadyStatus, StalledStatus, WaitingStatus + +if TYPE_CHECKING: + from ._types import RequestSourceStatus + + +def join_request_source_statuses( + first_status: RequestSourceStatus, + second_status: RequestSourceStatus, +) -> RequestSourceStatus: + """Combine two statuses with the precedence ready > waiting > stalled > finished. + + Of two waiting statuses, the one with the earlier known `ready_at` wins. + """ + if isinstance(first_status, ReadyStatus): + return first_status + if isinstance(second_status, ReadyStatus): + return second_status + + if isinstance(first_status, WaitingStatus): + if not isinstance(second_status, WaitingStatus) or second_status.ready_at is None: + return first_status + if first_status.ready_at is not None and first_status.ready_at <= second_status.ready_at: + return first_status + return second_status + + if isinstance(second_status, WaitingStatus): + return second_status + + if isinstance(first_status, StalledStatus): + return first_status + + # `first_status` is finished, so `second_status` decides. + return second_status diff --git a/src/crawlee/storage_clients/_base/_request_queue_client.py b/src/crawlee/storage_clients/_base/_request_queue_client.py index 768733f2da..e9efda0d9c 100644 --- a/src/crawlee/storage_clients/_base/_request_queue_client.py +++ b/src/crawlee/storage_clients/_base/_request_queue_client.py @@ -78,8 +78,7 @@ async def fetch_next_request(self) -> Request | None: in another call to the `fetch_next_request` method. Note that the `None` return value does not mean the queue processing finished, it means there are currently - no pending requests. To check whether all requests in queue were finished, use `RequestQueue.is_finished` - instead. + no pending requests. To check whether all requests in queue were finished, use `is_finished` instead. Returns: The request or `None` if there are no more pending requests. diff --git a/src/crawlee/storage_clients/_file_system/_request_queue_client.py b/src/crawlee/storage_clients/_file_system/_request_queue_client.py index 2ba57709bc..bb846627cf 100644 --- a/src/crawlee/storage_clients/_file_system/_request_queue_client.py +++ b/src/crawlee/storage_clients/_file_system/_request_queue_client.py @@ -488,8 +488,12 @@ async def fetch_next_request(self) -> Request | None: state.in_progress_requests.add(next_request.unique_key) # `in_progress_requests` is updated, so we need to invalidate the `is_empty` cache. self._is_empty_cache = None + return next_request - return next_request + # Yield to the event loop like the other clients do, so a caller polling an empty queue can't starve the + # tasks that would fill it. + await asyncio.sleep(0) + return None @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: diff --git a/src/crawlee/storage_clients/_memory/_request_queue_client.py b/src/crawlee/storage_clients/_memory/_request_queue_client.py index fae1aeda21..640bf43ce1 100644 --- a/src/crawlee/storage_clients/_memory/_request_queue_client.py +++ b/src/crawlee/storage_clients/_memory/_request_queue_client.py @@ -1,5 +1,6 @@ from __future__ import annotations +import asyncio from collections import OrderedDict from datetime import UTC, datetime from logging import getLogger @@ -212,6 +213,9 @@ async def add_batch_of_requests( @override async def fetch_next_request(self) -> Request | None: if not self._pending_requests: + # Yield to the event loop like the other clients do, so a caller polling an empty queue can't starve the + # tasks that would fill it. + await asyncio.sleep(0) return None _, request = self._pending_requests.popitem(last=False) diff --git a/src/crawlee/storages/_request_queue.py b/src/crawlee/storages/_request_queue.py index b614928a2b..11887f0d58 100644 --- a/src/crawlee/storages/_request_queue.py +++ b/src/crawlee/storages/_request_queue.py @@ -10,7 +10,7 @@ from crawlee import Request, service_locator from crawlee._utils.docs import docs_group from crawlee._utils.wait import wait_for_all_tasks_to_finish -from crawlee.request_loaders import RequestManager +from crawlee.request_loaders import FinishedStatus, ReadyStatus, RequestManager, WaitingStatus from crawlee.storage_clients.models import AddRequestsResponse from ._base import Storage @@ -21,6 +21,7 @@ from crawlee import Request from crawlee.configuration import Configuration + from crawlee.request_loaders import RequestLoaderStatus from crawlee.storage_clients import StorageClient from crawlee.storage_clients._base import RequestQueueClient from crawlee.storage_clients.models import ProcessedRequest, RequestQueueMetadata @@ -257,7 +258,7 @@ async def fetch_next_request(self) -> Request | None: in another call to the `fetch_next_request` method. Note that the `None` return value does not mean the queue processing finished, it means there are currently - no pending requests. To check whether all requests in queue were finished, use `RequestQueue.is_finished` + no pending requests. To check whether all requests in queue were finished, use `RequestQueue.check_readiness` instead. Returns: @@ -313,36 +314,24 @@ async def reclaim_request( """ return await self._client.reclaim_request(request, forefront=forefront) - async def is_empty(self) -> bool: - """Check if the request queue is empty. - - An empty queue means that there are no requests currently available to fetch. However, this does not - necessarily mean that the crawling operation is finished, as there still might be requests being processed - or tasks that could add additional requests to the queue. + @override + async def check_readiness(self) -> RequestLoaderStatus: + """Report whether the queue has a request to hand over, is waiting for one, or is done. - Returns: - True if the request queue is empty, False otherwise. + The queue waits while requests are in progress, possibly in another client sharing it, or while requests are + still being added in the background. Neither has a clock, so `WaitingStatus.ready_at` stays `None`. """ - return await self._client.is_empty() - - async def is_finished(self) -> bool: - """Check if the request queue is finished. + # A fetchable request outranks everything below, so this is the only backend call a ready queue needs. + if not await self._client.is_empty(): + return ReadyStatus() - A finished queue means that all requests have been processed and there are no more tasks that could add - additional requests to the queue. This is the definitive way to check if a crawling operation is complete. - - Returns: - True if the request queue is finished and no pending add operations, False otherwise. - """ if self._add_requests_tasks: - logger.debug('Background add requests tasks are still in progress.') - return False + return WaitingStatus() if await self._client.is_finished(): - logger.debug('The request queue is finished.') - return True + return FinishedStatus() - return False + return WaitingStatus() async def _process_batch( self, diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 3ef451b259..4e23ab8db8 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -38,7 +38,16 @@ ) from crawlee.events import Event, EventCrawlerStatusData, LocalEventManager from crawlee.http_clients import HttpClient -from crawlee.request_loaders import RequestList, RequestManagerTandem, ThrottlingRequestManager +from crawlee.request_loaders import ( + FinishedStatus, + ReadyStatus, + RequestList, + RequestManagerTandem, + RequestSourceStatus, + StalledStatus, + ThrottlingRequestManager, + WaitingStatus, +) from crawlee.sessions import Session, SessionPool from crawlee.statistics import FinalStatistics, Statistics, StatisticsState from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient @@ -70,16 +79,12 @@ async def handler(context: BasicCrawlingContext) -> None: assert calls == ['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com'] -@pytest.mark.parametrize( - 'method', - [pytest.param('is_empty', id='is_empty'), pytest.param('is_finished', id='is_finished')], -) @pytest.mark.parametrize( 'error_type', [pytest.param(RuntimeError, id='runtime error'), pytest.param(TimeoutError, id='timeout error')], ) async def test_propagates_request_queue_status_errors( - monkeypatch: pytest.MonkeyPatch, method: str, error_type: type[Exception] + monkeypatch: pytest.MonkeyPatch, error_type: type[Exception] ) -> None: """A request manager failure reaches the caller, and the crawler can be run again once the manager recovers.""" queue = await RequestQueue.open() @@ -93,17 +98,17 @@ async def handler(context: BasicCrawlingContext) -> None: error = error_type('Queue status unavailable') with monkeypatch.context() as monkey: - monkey.setattr(queue, method, AsyncMock(side_effect=error)) + monkey.setattr(queue, 'check_readiness', AsyncMock(side_effect=error)) with pytest.raises(error_type, match='Queue status unavailable') as exc_info: await crawler.run() assert exc_info.value is error assert handled_urls == [] - assert not await queue.is_finished() + assert await queue.check_readiness() == ReadyStatus() await crawler.run() assert handled_urls == ['https://a.placeholder.com'] - assert await queue.is_finished() + assert await queue.check_readiness() == FinishedStatus() async def test_crawler_is_usable_after_a_failed_run_setup(monkeypatch: pytest.MonkeyPatch) -> None: @@ -139,7 +144,7 @@ async def handler(context: BasicCrawlingContext) -> None: assert handled_urls == ['https://a.placeholder.com'] with monkeypatch.context() as monkey: - monkey.setattr(queue, 'is_empty', AsyncMock(side_effect=RuntimeError('Queue status unavailable'))) + monkey.setattr(queue, 'check_readiness', AsyncMock(side_effect=RuntimeError('Queue status unavailable'))) with pytest.raises(RuntimeError, match='Queue status unavailable'): await crawler.run(['https://b.placeholder.com']) @@ -160,7 +165,7 @@ async def handler(context: BasicCrawlingContext) -> None: await crawler.run(['https://a.placeholder.com']) with monkeypatch.context() as monkey: - monkey.setattr(queue, 'is_empty', AsyncMock(side_effect=RuntimeError('Queue status unavailable'))) + monkey.setattr(queue, 'check_readiness', AsyncMock(side_effect=RuntimeError('Queue status unavailable'))) with pytest.raises(RuntimeError, match='Queue status unavailable'): await crawler.run(['https://b.placeholder.com']) @@ -2719,8 +2724,8 @@ async def test_warn_unconfigured_throttle_domain_once_per_domain(caplog: pytest. 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.""" +async def test_crawl_waits_out_backoff() -> None: + """Test that a domain in a cooldown reads as waiting, 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) @@ -2738,18 +2743,20 @@ async def test_throttled_domain_waits_out_backoff_without_ending_the_crawl() -> concurrency_settings=ConcurrencySettings(desired_concurrency=1, max_concurrency=1), ) dispatched_at = list[float]() - empty_during_cooldown = list[bool]() + statuses_during_cooldown = list[RequestSourceStatus]() @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()) + statuses_during_cooldown.append(await throttler.check_readiness()) await crawler.run(['https://throttled.placeholder.com/a', 'https://throttled.placeholder.com/b']) - assert empty_during_cooldown == [True] + assert len(statuses_during_cooldown) == 1 + assert isinstance(statuses_during_cooldown[0], WaitingStatus) + assert statuses_during_cooldown[0].ready_at is not None assert len(dispatched_at) == 2 assert dispatched_at[1] - dispatched_at[0] >= 0.5 @@ -2764,6 +2771,11 @@ async def handler(context: BasicCrawlingContext) -> None: MAX_HANDLER_CALLS = 50 +def is_stalled(status: RequestSourceStatus) -> bool: + """Tell whether a request manager reported a stall.""" + return isinstance(status, StalledStatus) + + async def open_throttler() -> ThrottlingRequestManager[RequestQueue]: """Open a throttler for `THROTTLED_URL` with short delays and a short stall window.""" return ThrottlingRequestManager( @@ -2861,7 +2873,7 @@ async def test_keep_alive_ignores_stall() -> None: """A crawler running with `keep_alive` keeps waiting on a stalled domain instead of raising.""" throttler = await open_throttler() crawler = BasicCrawler(request_manager=throttler, keep_alive=True) - stall_reasons = list[str | None]() + stall_statuses = list[RequestSourceStatus]() @crawler.router.default_handler async def handler(context: BasicCrawlingContext) -> None: @@ -2869,7 +2881,7 @@ async def handler(context: BasicCrawlingContext) -> None: raise RequestThrottledError async def stop_after_stall() -> None: - stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, timeout=10)) + stall_statuses.append(await poll_until_condition(throttler.check_readiness, is_stalled, timeout=10)) # Past the pool's next check, where a crawler without `keep_alive` raises. await asyncio.sleep(0.6) crawler.stop() @@ -2881,7 +2893,7 @@ async def stop_after_stall() -> None: finally: stopper.cancel() - assert stall_reasons[0] is not None + assert isinstance(stall_statuses[0], StalledStatus) async def test_stall_waits_for_in_flight_work() -> None: @@ -2894,7 +2906,7 @@ async def test_stall_waits_for_in_flight_work() -> None: concurrency_settings=ConcurrencySettings(min_concurrency=2, desired_concurrency=2), ) visited = list[str]() - stall_reasons = list[str | None]() + stall_statuses = list[RequestSourceStatus]() @crawler.router.default_handler async def handler(context: BasicCrawlingContext) -> None: @@ -2905,19 +2917,20 @@ async def handler(context: BasicCrawlingContext) -> None: throttler.record_domain_delay(context.request.url) raise RequestThrottledError if context.request.url == 'https://free.placeholder.com/slow': - stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, timeout=10)) + stall_statuses.append(await poll_until_condition(throttler.check_readiness, is_stalled, timeout=10)) # Two more pool checks, so at least one ran entirely while the stall was live and this request in flight. checks = is_finished.await_count await poll_until_condition(lambda: is_finished.await_count >= checks + 2, timeout=10) await context.add_requests(['https://free.placeholder.com/next']) + is_finished_function = cast('Any', crawler)._BasicCrawler__is_finished_function with ( - patch.object(throttler, 'is_finished', wraps=throttler.is_finished) as is_finished, + patch.object(crawler, '_BasicCrawler__is_finished_function', wraps=is_finished_function) as is_finished, pytest.raises(PersistentRateLimitError), ): await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) - assert stall_reasons[0] is not None + assert isinstance(stall_statuses[0], StalledStatus) assert 'https://free.placeholder.com/next' in visited @@ -2932,7 +2945,7 @@ async def test_max_requests_per_crawl_precedes_stall() -> None: concurrency_settings=ConcurrencySettings(min_concurrency=2, desired_concurrency=2), ) calls = 0 - stall_reasons = list[str | None]() + stall_statuses = list[RequestSourceStatus]() @crawler.router.default_handler async def handler(context: BasicCrawlingContext) -> None: @@ -2943,11 +2956,11 @@ async def handler(context: BasicCrawlingContext) -> None: if context.request.url == THROTTLED_URL: throttler.record_domain_delay(context.request.url) raise RequestThrottledError - stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, timeout=10)) + stall_statuses.append(await poll_until_condition(throttler.check_readiness, is_stalled, timeout=10)) stats = await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) - assert stall_reasons[0] is not None + assert isinstance(stall_statuses[0], StalledStatus) assert stats.requests_finished == 1 @@ -2982,6 +2995,18 @@ async def handler(context: BasicCrawlingContext) -> None: assert set(paced_urls) <= set(visited) +async def test_stall_behind_tandem_raises() -> None: + """Test that a stall reported through a `RequestManagerTandem` ends the crawl.""" + queue = await RequestQueue.open() + crawler = BasicCrawler(request_manager=RequestManagerTandem(RequestList([]), queue)) + + with ( + patch.object(queue, 'check_readiness', return_value=StalledStatus('stuck')), + pytest.raises(PersistentRateLimitError, match='Giving up: stuck'), + ): + await crawler.run() + + async def test_inner_held_429_is_deferred_and_stalls() -> None: """A rate-limited request held by `inner` moves into its domain's sub-manager and can stall the crawl.""" throttler = await open_throttler() @@ -3001,8 +3026,8 @@ async def handler(context: BasicCrawlingContext) -> None: with pytest.raises(PersistentRateLimitError): await crawler.run() - assert await throttler.inner.is_finished() is True - assert await throttler.is_finished() is False + assert await throttler.inner.check_readiness() == FinishedStatus() + assert await throttler.check_readiness() != FinishedStatus() @pytest.mark.parametrize( @@ -3165,7 +3190,7 @@ async def test_stall_ignores_shared_concurrency() -> None: with ( patch.object(crawler, '_autoscaled_pool', pool), - patch.object(throttler, 'get_stall_reason', AsyncMock(return_value='Domain stalled')), + patch.object(throttler, 'check_readiness', AsyncMock(return_value=StalledStatus('Domain stalled'))), pytest.raises(PersistentRateLimitError), ): await cast('Any', crawler)._BasicCrawler__is_finished_function() diff --git a/tests/unit/crawlers/_http/test_http_crawler.py b/tests/unit/crawlers/_http/test_http_crawler.py index 6102944f26..97cd626b38 100644 --- a/tests/unit/crawlers/_http/test_http_crawler.py +++ b/tests/unit/crawlers/_http/test_http_crawler.py @@ -3,7 +3,7 @@ import json import logging from datetime import UTC, datetime, timedelta -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from unittest.mock import AsyncMock, Mock from urllib.parse import parse_qs, urlencode @@ -13,7 +13,7 @@ from crawlee.crawlers import HttpCrawler from crawlee.errors import PersistentRateLimitError from crawlee.http_clients import HttpClient, HttpCrawlingResult -from crawlee.request_loaders import ThrottlingRequestManager, _throttling_request_manager +from crawlee.request_loaders import FinishedStatus, ThrottlingRequestManager, _throttling_request_manager from crawlee.sessions import SessionPool from crawlee.statistics import Statistics from crawlee.storages import RequestQueue @@ -770,13 +770,6 @@ async def test_persistent_429_raises_after_stall( max_delay=backoff, max_domain_stall=backoff * 1.5, ) - get_stall_reason = throttler.get_stall_reason - - async def get_stall_reason_after_backoff() -> str | None: - clock.now.return_value += backoff - return await get_stall_reason() - - monkeypatch.setattr(throttler, 'get_stall_reason', get_stall_reason_after_backoff) failed_request_handler = AsyncMock() crawler = HttpCrawler( request_handler=AsyncMock(), @@ -786,12 +779,20 @@ async def get_stall_reason_after_backoff() -> str | None: max_session_rotations=0, ) crawler.failed_request_handler(failed_request_handler) + is_finished = cast('Any', crawler)._BasicCrawler__is_finished_function + + async def is_finished_after_backoff() -> bool: + if crawler._autoscaled_pool is not None and crawler._autoscaled_pool.own_concurrency == 0: + clock.now.return_value += backoff + return await is_finished() + + monkeypatch.setattr(crawler, '_BasicCrawler__is_finished_function', is_finished_after_backoff) with caplog.at_level(logging.ERROR), pytest.raises(PersistentRateLimitError, match='Giving up: '): await crawler.run([str(server_url / 'status/429')]) assert any(record.levelno == logging.ERROR and record.exc_info for record in caplog.records) - assert await throttler.is_finished() is False + assert await throttler.check_readiness() != FinishedStatus() failed_request_handler.assert_not_called() # A failed run keeps the queue, and the stall clock still runs, so a re-run gives up again. diff --git a/tests/unit/request_loaders/test_request_list.py b/tests/unit/request_loaders/test_request_list.py index 5c1dfd3639..b70c14df2a 100644 --- a/tests/unit/request_loaders/test_request_list.py +++ b/tests/unit/request_loaders/test_request_list.py @@ -1,22 +1,26 @@ from collections.abc import AsyncGenerator +from crawlee.request_loaders import FinishedStatus, ReadyStatus, WaitingStatus from crawlee.request_loaders._request_list import RequestList from crawlee.storages import KeyValueStore async def test_sync_traversal() -> None: + """Test that a list built from a sync iterable is traversed to the end.""" request_list = RequestList(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com']) - while not await request_list.is_finished(): + while not isinstance(await request_list.check_readiness(), FinishedStatus): item = await request_list.fetch_next_request() assert item is not None await request_list.mark_request_as_handled(item) - assert await request_list.is_empty() + assert await request_list.get_handled_count() == 3 async def test_async_traversal() -> None: + """Test that a list built from an async generator is traversed to the end.""" + async def generator() -> AsyncGenerator[str]: yield 'https://a.placeholder.com' yield 'https://b.placeholder.com' @@ -24,39 +28,32 @@ async def generator() -> AsyncGenerator[str]: request_list = RequestList(generator()) - while not await request_list.is_finished(): + while not isinstance(await request_list.check_readiness(), FinishedStatus): item = await request_list.fetch_next_request() assert item is not None await request_list.mark_request_as_handled(item) - assert await request_list.is_empty() + assert await request_list.get_handled_count() == 3 -async def test_is_empty_does_not_depend_on_fetch_next_request() -> None: +async def test_readiness_during_traversal() -> None: + """Test that the list waits for fetched requests and finishes once they're all handled.""" request_list = RequestList(['https://a.placeholder.com', 'https://b.placeholder.com', 'https://c.placeholder.com']) - item_1 = await request_list.fetch_next_request() - assert item_1 is not None - assert not await request_list.is_finished() - - item_2 = await request_list.fetch_next_request() - assert item_2 is not None - assert not await request_list.is_finished() - - item_3 = await request_list.fetch_next_request() - assert item_3 is not None - assert not await request_list.is_finished() + items = [] + for _ in range(3): + assert await request_list.check_readiness() == ReadyStatus() + item = await request_list.fetch_next_request() + assert item is not None + items.append(item) - assert await request_list.is_empty() - assert not await request_list.is_finished() + assert await request_list.check_readiness() == WaitingStatus() - await request_list.mark_request_as_handled(item_1) - await request_list.mark_request_as_handled(item_2) - await request_list.mark_request_as_handled(item_3) + for item in items: + await request_list.mark_request_as_handled(item) - assert await request_list.is_empty() - assert await request_list.is_finished() + assert await request_list.check_readiness() == FinishedStatus() async def test_persist_requests_key_with_sync_iterable() -> None: @@ -86,9 +83,8 @@ async def test_persist_requests_key_with_empty_iterator() -> None: # Create request list with empty iterator request_list = RequestList([], persist_requests_key=persist_key) - # Should be empty immediately - assert await request_list.is_empty() - assert await request_list.is_finished() + # Should be finished immediately + assert await request_list.check_readiness() == FinishedStatus() # Check that empty requests were persisted kvs = await KeyValueStore.open() diff --git a/tests/unit/request_loaders/test_request_source_status.py b/tests/unit/request_loaders/test_request_source_status.py new file mode 100644 index 0000000000..76357289a4 --- /dev/null +++ b/tests/unit/request_loaders/test_request_source_status.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest + +from crawlee.request_loaders import FinishedStatus, ReadyStatus, RequestSourceStatus, StalledStatus, WaitingStatus +from crawlee.request_loaders._utils import join_request_source_statuses + +READY = ReadyStatus() +STALLED = StalledStatus('example.com rate-limited every request') +WAITING = WaitingStatus() +FINISHED = FinishedStatus() + +EARLY = datetime(2026, 1, 1, tzinfo=UTC) +LATE = EARLY + timedelta(seconds=10) + + +@pytest.mark.parametrize( + ('first_status', 'second_status', 'expected'), + [ + pytest.param(READY, READY, READY, id='ready+ready'), + pytest.param(READY, STALLED, READY, id='ready+stalled'), + pytest.param(READY, WAITING, READY, id='ready+waiting'), + pytest.param(READY, FINISHED, READY, id='ready+finished'), + pytest.param(STALLED, STALLED, STALLED, id='stalled+stalled'), + pytest.param(WAITING, STALLED, WAITING, id='waiting+stalled'), + pytest.param(STALLED, FINISHED, STALLED, id='stalled+finished'), + pytest.param(WAITING, WAITING, WAITING, id='waiting+waiting'), + pytest.param(WAITING, FINISHED, WAITING, id='waiting+finished'), + pytest.param(FINISHED, FINISHED, FINISHED, id='finished+finished'), + ], +) +def test_join_precedence( + first_status: RequestSourceStatus, + second_status: RequestSourceStatus, + expected: RequestSourceStatus, +) -> None: + """Test that the join ranks ready over waiting over stalled over finished, in either order.""" + assert join_request_source_statuses(first_status, second_status) == expected + assert join_request_source_statuses(second_status, first_status) == expected + + +def test_join_keeps_earliest_ready_at() -> None: + """Test that the earlier of two known wake-up times wins.""" + early = WaitingStatus(ready_at=EARLY) + late = WaitingStatus(ready_at=LATE) + + assert join_request_source_statuses(early, late) == early + assert join_request_source_statuses(late, early) == early + + +def test_join_keeps_known_ready_at() -> None: + """Test that a known wake-up time survives a source that has none.""" + known = WaitingStatus(ready_at=EARLY) + + assert join_request_source_statuses(WAITING, known) == known + assert join_request_source_statuses(known, WAITING) == known + assert join_request_source_statuses(known, FINISHED) == known + assert join_request_source_statuses(STALLED, known) == known diff --git a/tests/unit/request_loaders/test_sitemap_request_loader.py b/tests/unit/request_loaders/test_sitemap_request_loader.py index d88b84164d..84f632f704 100644 --- a/tests/unit/request_loaders/test_sitemap_request_loader.py +++ b/tests/unit/request_loaders/test_sitemap_request_loader.py @@ -1,3 +1,4 @@ +import asyncio import base64 import gzip from contextlib import asynccontextmanager @@ -12,6 +13,7 @@ from crawlee._utils.sitemap import DEFAULT_MAX_DEPTH from crawlee.crawlers import BasicCrawler from crawlee.http_clients._base import HttpClient, HttpResponse +from crawlee.request_loaders import FinishedStatus, ReadyStatus, RequestSourceStatus, WaitingStatus from crawlee.request_loaders._sitemap_request_loader import SitemapRequestLoader from crawlee.storages import KeyValueStore from tests.unit.utils import ( @@ -28,6 +30,11 @@ from crawlee._types import JsonSerializable +def finished(status: RequestSourceStatus) -> bool: + """Tell whether a loader reported that every request has been handled.""" + return isinstance(status, FinishedStatus) + + def compress_gzip(data: str) -> bytes: """Compress a string using gzip.""" return gzip.compress(data.encode()) @@ -63,7 +70,7 @@ async def read_stream() -> 'AsyncIterator[bytes]': client.stream = stream loader = SitemapRequestLoader([f'{DEFAULT_URL}sitemap-0.xml'], http_client=client) - assert await poll_until_condition(loader.is_finished) + assert finished(await poll_until_condition(loader.check_readiness, finished)) assert len(fetched) == DEFAULT_MAX_DEPTH + 1 @@ -74,14 +81,13 @@ async def test_sitemap_traversal(server_url: URL, http_client: HttpClient) -> No ) sitemap_loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, enqueue_strategy='all') - while not await sitemap_loader.is_finished(): + while not finished(await sitemap_loader.check_readiness()): item = await sitemap_loader.fetch_next_request() if item: await sitemap_loader.mark_request_as_handled(item) - assert await sitemap_loader.is_empty() - assert await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == FinishedStatus() assert await sitemap_loader.get_total_count() == 5 assert await sitemap_loader.get_handled_count() == 5 @@ -95,7 +101,7 @@ async def test_sitemap_http_error_is_retried_before_loading_requests(monkeypatch ) loader = SitemapRequestLoader([sitemap_url], http_client=client) - while not await loader.is_finished(): + while not finished(await loader.check_readiness()): request = await loader.fetch_next_request() if request: await loader.mark_request_as_handled(request) @@ -135,11 +141,11 @@ async def test_sitemap_loader_drains_requests_after_later_source_fails(monkeypat assert request is not None requests.append(request) - assert not await loader.is_finished() + assert not finished(await loader.check_readiness()) for request in requests: await loader.mark_request_as_handled(request) - assert await poll_until_condition(loader.is_finished) + assert finished(await poll_until_condition(loader.check_readiness, finished)) async def test_sitemap_loader_continues_after_earlier_source_fails(monkeypatch: pytest.MonkeyPatch) -> None: @@ -154,7 +160,7 @@ async def test_sitemap_loader_continues_after_earlier_source_fails(monkeypatch: loader = SitemapRequestLoader([f'{DEFAULT_URL}broken.xml', f'{DEFAULT_URL}sitemap.xml'], http_client=client) urls = [] - while not await loader.is_finished(): + while not finished(await loader.check_readiness()): request = await loader.fetch_next_request() if request: urls.append(request.url) @@ -187,7 +193,8 @@ async def handler(context: BasicCrawlingContext) -> None: assert stats.requests_failed == 0 -async def test_is_empty_does_not_depend_on_fetch_next_request(server_url: URL, http_client: HttpClient) -> None: +async def test_readiness_during_traversal(server_url: URL, http_client: HttpClient) -> None: + """Test that the loader waits for fetched requests and finishes once they're all handled.""" sitemap_url = (server_url / 'sitemap.xml').with_query( base64=encode_base64(get_basic_sitemap(url=server_url).encode()) ) @@ -198,18 +205,30 @@ async def test_is_empty_does_not_depend_on_fetch_next_request(server_url: URL, h for _ in range(5): item = await sitemap_loader.fetch_next_request() assert item is not None - assert not await sitemap_loader.is_finished() + assert not finished(await sitemap_loader.check_readiness()) items.append(item) - assert await sitemap_loader.is_empty() - assert not await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == WaitingStatus() for item in items: await sitemap_loader.mark_request_as_handled(item) - assert await sitemap_loader.is_empty() + assert await sitemap_loader.check_readiness() != ReadyStatus() + assert finished(await poll_until_condition(sitemap_loader.check_readiness, finished)) + + +async def test_fetch_none_while_in_progress(server_url: URL, http_client: HttpClient) -> None: + """Test that a fetch after the last URL returns `None` while that URL is still in progress.""" + sitemap_url = (server_url / 'sitemap.xml').with_query( + base64=encode_base64(get_basic_sitemap(url=server_url).encode()) + ) + sitemap_loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, enqueue_strategy='all') + + for _ in range(5): + assert await sitemap_loader.fetch_next_request() is not None - assert await poll_until_condition(sitemap_loader.is_finished) + assert await asyncio.wait_for(sitemap_loader.fetch_next_request(), timeout=5) is None + assert await sitemap_loader.check_readiness() == WaitingStatus() async def test_abort_sitemap_loading(server_url: URL, http_client: HttpClient) -> None: @@ -224,8 +243,7 @@ async def test_abort_sitemap_loading(server_url: URL, http_client: HttpClient) - assert item is not None await sitemap_loader.mark_request_as_handled(item) - assert not await sitemap_loader.is_empty() - assert not await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == ReadyStatus() await sitemap_loader.abort_loading() @@ -233,7 +251,7 @@ async def test_abort_sitemap_loading(server_url: URL, http_client: HttpClient) - assert item is not None await sitemap_loader.mark_request_as_handled(item) - assert await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == FinishedStatus() async def test_create_persist_state_for_sitemap_loading( @@ -246,7 +264,7 @@ async def test_create_persist_state_for_sitemap_loading( sitemap_loader = SitemapRequestLoader( [str(sitemap_url)], http_client=http_client, persist_state_key=persist_key, enqueue_strategy='all' ) - assert await sitemap_loader.is_finished() is False + assert not finished(await sitemap_loader.check_readiness()) await sitemap_loader.close() @@ -268,7 +286,7 @@ async def test_data_persistence_for_sitemap_loading( ) # Give time to load. - await poll_until_condition(sitemap_loader.is_empty, lambda is_empty: not is_empty, timeout=10) + await poll_until_condition(sitemap_loader.check_readiness, lambda status: status == ReadyStatus(), timeout=10) await sitemap_loader.close() @@ -332,7 +350,7 @@ def transform_request(request_options: RequestOptions) -> RequestOptions | Reque extracted_urls = set() - while not await sitemap_loader.is_finished(): + while not finished(await sitemap_loader.check_readiness()): request = await sitemap_loader.fetch_next_request() if request: @@ -361,7 +379,7 @@ def transform_request(_request_options: RequestOptions) -> RequestOptions | Requ enqueue_strategy='all', ) - while not await sitemap_loader.is_finished(): + while not finished(await sitemap_loader.check_readiness()): request = await sitemap_loader.fetch_next_request() if request: @@ -369,8 +387,7 @@ def transform_request(_request_options: RequestOptions) -> RequestOptions | Requ # Even though the sitemap had URLs, all were skipped, so the loader should be empty and finished with # 0 handled requests. - assert await sitemap_loader.is_empty() - assert await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == FinishedStatus() assert await sitemap_loader.get_total_count() == 0 assert await sitemap_loader.get_handled_count() == 0 @@ -386,17 +403,15 @@ async def test_sitemap_loader_to_tandem( sitemap_loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, enqueue_strategy='all') request_manager = await sitemap_loader.to_tandem() - while not await sitemap_loader.is_finished(): + while not finished(await sitemap_loader.check_readiness()): request = await request_manager.fetch_next_request() if request: await request_manager.mark_request_as_handled(request) - assert await sitemap_loader.is_empty() - assert await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == FinishedStatus() - assert await request_manager.is_empty() - assert await request_manager.is_finished() + assert await request_manager.check_readiness() == FinishedStatus() async def test_sitemap_loader_to_tandem_with_request_dropped( @@ -417,17 +432,15 @@ async def test_sitemap_loader_to_tandem_with_request_dropped( with patch.object( request_manager._read_write_manager, 'add_request', side_effect=Exception('Failed to add request') ): - while not await sitemap_loader.is_finished(): + while not finished(await sitemap_loader.check_readiness()): request = await request_manager.fetch_next_request() if request: await request_manager.mark_request_as_handled(request) - assert await sitemap_loader.is_empty() - assert await sitemap_loader.is_finished() + assert await sitemap_loader.check_readiness() == FinishedStatus() - assert await request_manager.is_empty() - assert await request_manager.is_finished() + assert await request_manager.check_readiness() == FinishedStatus() def _make_urlset(urls: list[str]) -> str: @@ -462,7 +475,7 @@ async def test_sitemap_loader_filters_cross_host_urls(server_url: URL, http_clie loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client) fetched: list[str] = [] - while not await loader.is_finished(): + while not finished(await loader.check_readiness()): request = await loader.fetch_next_request() if request is not None: fetched.append(request.url) @@ -482,7 +495,7 @@ async def test_sitemap_loader_filters_cross_host_nested_sitemap(server_url: URL, loader = SitemapRequestLoader([index_url], http_client=http_client) fetched: list[str] = [] - while not await loader.is_finished(): + while not finished(await loader.check_readiness()): request = await loader.fetch_next_request() if request is not None: fetched.append(request.url) @@ -513,7 +526,7 @@ async def test_sitemap_loader_strategy_all_disables_filtering(server_url: URL, h loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, enqueue_strategy='all') fetched: list[str] = [] - while not await loader.is_finished(): + while not finished(await loader.check_readiness()): request = await loader.fetch_next_request() if request is not None: fetched.append(request.url) @@ -535,7 +548,7 @@ async def test_sitemap_loader_drops_non_http_scheme_under_strategy_all( loader = SitemapRequestLoader([str(sitemap_url)], http_client=http_client, enqueue_strategy='all') fetched: list[str] = [] - while not await loader.is_finished(): + while not finished(await loader.check_readiness()): request = await loader.fetch_next_request() if request is not None: fetched.append(request.url) diff --git a/tests/unit/storages/test_request_manager_tandem.py b/tests/unit/storages/test_request_manager_tandem.py index 69bd944348..710c213aa2 100644 --- a/tests/unit/storages/test_request_manager_tandem.py +++ b/tests/unit/storages/test_request_manager_tandem.py @@ -1,12 +1,22 @@ from __future__ import annotations from dataclasses import dataclass +from datetime import UTC, datetime from unittest.mock import create_autospec import pytest from crawlee import Request -from crawlee.request_loaders import RequestLoader, RequestManagerTandem +from crawlee.request_loaders import ( + FinishedStatus, + ReadyStatus, + RequestLoader, + RequestManager, + RequestManagerTandem, + RequestSourceStatus, + StalledStatus, + WaitingStatus, +) from crawlee.storages import RequestQueue @@ -57,6 +67,7 @@ class TestInput: ], ) async def test_basic_functionality(test_input: TestInput) -> None: + """Test that the tandem serves requests from both the loader and the manager.""" request_queue = await RequestQueue.open() if test_input.request_manager_items: @@ -64,12 +75,14 @@ async def test_basic_functionality(test_input: TestInput) -> None: mock_request_loader = create_autospec(RequestLoader, instance=True, spec_set=True) mock_request_loader.fetch_next_request.side_effect = lambda: test_input.request_loader_items.pop(0) - mock_request_loader.is_finished.side_effect = lambda: len(test_input.request_loader_items) == 0 + mock_request_loader.check_readiness.side_effect = lambda: ( + ReadyStatus() if test_input.request_loader_items else FinishedStatus() + ) tandem = RequestManagerTandem(mock_request_loader, request_queue) processed = set[str]() - while not await tandem.is_finished(): + while not isinstance(await tandem.check_readiness(), FinishedStatus): request = await tandem.fetch_next_request() assert request is not None processed.add(request.url) @@ -80,3 +93,46 @@ async def test_basic_functionality(test_input: TestInput) -> None: await tandem.mark_request_as_handled(request) assert processed == test_input.expected_result + + +READY_AT = datetime(2026, 1, 1, tzinfo=UTC) + + +@pytest.mark.parametrize( + ('loader_status', 'manager_status', 'expected'), + [ + pytest.param(ReadyStatus(), FinishedStatus(), ReadyStatus(), id='ready loader'), + pytest.param( + FinishedStatus(), WaitingStatus(ready_at=READY_AT), WaitingStatus(ready_at=READY_AT), id='waiting' + ), + pytest.param(WaitingStatus(), StalledStatus('stuck'), WaitingStatus(), id='waiting masks stalled'), + pytest.param(FinishedStatus(), StalledStatus('stuck'), StalledStatus('stuck'), id='stalled'), + pytest.param(FinishedStatus(), FinishedStatus(), FinishedStatus(), id='finished'), + ], +) +async def test_readiness_joins_sources( + loader_status: RequestSourceStatus, + manager_status: RequestSourceStatus, + expected: RequestSourceStatus, +) -> None: + """Test that the tandem reports the loader and the manager as one source.""" + loader = create_autospec(RequestLoader, instance=True, spec_set=True) + loader.check_readiness.return_value = loader_status + manager = create_autospec(RequestManager, instance=True, spec_set=True) + manager.check_readiness.return_value = manager_status + + assert await RequestManagerTandem(loader, manager).check_readiness() == expected + + +async def test_transfers_only_when_loader_ready() -> None: + """Test that a loader that isn't ready doesn't block the manager's requests.""" + loader = create_autospec(RequestLoader, instance=True, spec_set=True) + loader.check_readiness.return_value = WaitingStatus() + manager = create_autospec(RequestManager, instance=True, spec_set=True) + manager.fetch_next_request.return_value = Request.from_url('https://a.placeholder.com') + + request = await RequestManagerTandem(loader, manager).fetch_next_request() + + assert request is not None + assert request.url == 'https://a.placeholder.com' + loader.fetch_next_request.assert_not_awaited() diff --git a/tests/unit/storages/test_request_queue.py b/tests/unit/storages/test_request_queue.py index 112adfaa96..42edba3510 100644 --- a/tests/unit/storages/test_request_queue.py +++ b/tests/unit/storages/test_request_queue.py @@ -3,11 +3,13 @@ import asyncio from datetime import timedelta from typing import TYPE_CHECKING +from unittest.mock import patch import pytest from crawlee import Request, service_locator from crawlee.configuration import Configuration +from crawlee.request_loaders import FinishedStatus, ReadyStatus, WaitingStatus from crawlee.storage_clients import MemoryStorageClient, StorageClient from crawlee.storage_clients.models import AddRequestsResponse, ProcessedRequest, UnprocessedRequest from crawlee.storages import RequestQueue @@ -544,33 +546,52 @@ async def test_reclaim_request_with_forefront(rq: RequestQueue) -> None: assert next_request.url == 'https://example.com/first' -async def test_is_empty_and_is_finished(rq: RequestQueue) -> None: - """Test checking if a request queue is empty and finished.""" - # Initially the queue should be empty and finished - assert await rq.is_empty() is True - assert await rq.is_finished() is True +async def test_check_readiness(rq: RequestQueue) -> None: + """Test that the readiness follows a request from adding to handling.""" + assert await rq.check_readiness() == FinishedStatus() - # Add a request await rq.add_request('https://example.com') - assert await rq.is_empty() is False - assert await rq.is_finished() is False + assert await rq.check_readiness() == ReadyStatus() - # Fetch the request request = await rq.fetch_next_request() - assert request is not None - # Queue is empty, because there is no request for fetching - assert await rq.is_empty() is True - # Queue is not finished, because there is a request being processed - assert await rq.is_finished() is False + # Nothing to fetch, but the request is still in progress. + assert await rq.check_readiness() == WaitingStatus() - # Mark the request as handled await rq.mark_request_as_handled(request) + assert await rq.check_readiness() == FinishedStatus() + + +async def test_readiness_waits_for_background_adds(rq: RequestQueue) -> None: + """Test that a drained queue waits while a batch is still being added in the background.""" + release = asyncio.Event() + add_batch = rq._client.add_batch_of_requests + + async def add_batch_after_release(requests: Sequence[Request], *, forefront: bool = False) -> AddRequestsResponse: + if requests[0].url == 'https://example.com/2': + await release.wait() + return await add_batch(requests, forefront=forefront) + + with patch.object(rq._client, 'add_batch_of_requests', side_effect=add_batch_after_release): + await rq.add_requests(['https://example.com/1', 'https://example.com/2'], batch_size=1) - # Queue should be empty and finished again - assert await rq.is_empty() is True - assert await rq.is_finished() is True + request = await rq.fetch_next_request() + assert request is not None + await rq.mark_request_as_handled(request) + assert await rq.check_readiness() == WaitingStatus() + + release.set() + assert await poll_until_condition(rq.check_readiness, lambda status: status == ReadyStatus()) == ReadyStatus() + + +async def test_empty_fetch_yields(rq: RequestQueue) -> None: + """Test that a fetch from an empty queue lets other tasks run.""" + ran = asyncio.Event() + asyncio.get_running_loop().call_soon(ran.set) + + assert await rq.fetch_next_request() is None + assert ran.is_set() @pytest.mark.parametrize( @@ -607,38 +628,6 @@ async def test_add_requests_wait_for_all( assert await rq.get_total_count() == 15 -async def test_is_finished(rq: RequestQueue) -> None: - """Test checking if a request queue is finished.""" - # Initially the queue should be finished (empty and no background tasks) - assert await rq.is_finished() is True - - # Add a request - await rq.add_request('https://example.com') - assert await rq.is_finished() is False - - # Add requests in the background - await rq.add_requests( - ['https://example.com/1', 'https://example.com/2'], - wait_for_all_requests_to_be_added=False, - ) - - # Queue shouldn't be finished while background tasks are running - assert await rq.is_finished() is False - - # Wait for the background add task to finish. - await poll_until_condition(lambda: not rq._add_requests_tasks) - - # Process all requests - while True: - request = await rq.fetch_next_request() - if request is None: - break - await rq.mark_request_as_handled(request) - - # Now queue should be finished - assert await rq.is_finished() is True - - async def test_mark_non_existent_request_as_handled(rq: RequestQueue) -> None: """Test marking a non-existent request as handled.""" # Create a request that hasn't been added to the queue @@ -681,7 +670,7 @@ async def test_drop( ) # Verify the queue is empty - assert await new_rq.is_empty() is True + assert await new_rq.check_readiness() == FinishedStatus() metadata = await new_rq.get_metadata() assert metadata.total_request_count == 0 assert metadata.pending_request_count == 0 @@ -785,7 +774,7 @@ async def test_purge( assert metadata.total_request_count == 0 assert metadata.pending_request_count == 0 assert metadata.handled_request_count == 0 - assert await rq.is_empty() is True + assert await rq.check_readiness() != ReadyStatus() # Verify we can add new requests after purging await rq.add_request('https://example.com/new-after-purge') @@ -1043,7 +1032,7 @@ async def test_alias_request_operations( # Test fetching and handling requests processed_urls = [] - while not await rq.is_empty(): + while await rq.check_readiness() == ReadyStatus(): request = await rq.fetch_next_request() if request: processed_urls.append(request.url) @@ -1057,7 +1046,7 @@ async def test_alias_request_operations( metadata = await rq.get_metadata() assert metadata.pending_request_count == 0 assert metadata.handled_request_count == 3 - assert await rq.is_empty() is True + assert await rq.check_readiness() == FinishedStatus() # Clean up await rq.drop() diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index 88fc10bc9d..ce5fb764bd 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -15,6 +15,7 @@ from crawlee._service_locator import ServiceLocator from crawlee._utils.http import parse_retry_after_header from crawlee.configuration import Configuration +from crawlee.request_loaders import FinishedStatus, ReadyStatus, StalledStatus, WaitingStatus from crawlee.request_loaders._throttling_request_manager import ThrottlingRequestManager from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient from crawlee.storages import RequestQueue @@ -126,7 +127,7 @@ async def test_add_request_routes_listed_domain_to_sub_manager( await manager.add_request(request) # Inner manager should be empty — the request went to a sub-manager. - assert await inner_queue.is_empty() + assert await inner_queue.check_readiness() == FinishedStatus() assert THROTTLED_DOMAIN in manager._sub_managers # The sub-manager should have the request. @@ -141,7 +142,7 @@ async def test_add_request_routes_non_listed_domain_to_inner( request = _make_request(f'https://{NON_THROTTLED_DOMAIN}/page1') await manager.add_request(request) - assert not await inner_queue.is_empty() + assert await inner_queue.check_readiness() == ReadyStatus() assert NON_THROTTLED_DOMAIN not in manager._sub_managers @@ -170,7 +171,7 @@ async def test_domain_matching_is_case_insensitive( await manager.add_request('https://EXAMPLE.com/page1') - assert await inner_queue.is_empty() + assert await inner_queue.check_readiness() == FinishedStatus() assert 'example.com' in manager._sub_managers assert await manager._sub_managers['example.com'].get_total_count() == 1 @@ -277,7 +278,7 @@ async def test_add_requests_routes_mixed_domains( await manager.add_requests([throttled_req, free_req]) # Inner gets the non-listed domain request. - assert not await inner_queue.is_empty() + assert await inner_queue.check_readiness() == ReadyStatus() # Sub-manager gets the listed domain request. assert THROTTLED_DOMAIN in manager._sub_managers @@ -596,47 +597,76 @@ async def test_fetch_returns_none_when_all_throttled(manager: ThrottlingRequestM assert await asyncio.wait_for(manager.fetch_next_request(), timeout=1.0) is None -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.""" +async def test_cooldown_reads_as_waiting(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that requests waiting out a cooldown make the manager wait until the cooldown ends.""" 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 - assert await manager.is_finished() is False + ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until + assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) -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`.""" +async def test_crawl_delay_reads_as_waiting(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """A crawl-delay armed on dispatch makes the manager wait before the domain's next request.""" 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 - assert await manager.is_empty() is True - assert await manager.is_finished() is False + ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until + assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) async def test_expired_throttle_makes_the_domain_fetchable_again( manager: ThrottlingRequestManager[RequestQueue], ) -> None: - """Once the cooldown passes, the domain counts again for both dispatching and emptiness.""" + """Once the cooldown passes, the domain counts again for both dispatching and readiness.""" 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 - manager._domain_states[THROTTLED_DOMAIN].backoff_until = datetime.now(UTC) - timedelta(seconds=1) + with _frozen_clock() as clock: + manager.record_domain_delay(url, retry_after=timedelta(seconds=60)) + assert isinstance(await manager.check_readiness(), WaitingStatus) + + clock.now.return_value += timedelta(seconds=61) + assert await manager.check_readiness() == ReadyStatus() + result = await manager.fetch_next_request() - assert await manager.is_empty() is False - result = await manager.fetch_next_request() assert result is not None assert result.url == url +async def test_waits_for_earliest_cooldown(two_domain_manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that the manager waits until the earliest cooldown of a domain with work left ends.""" + late_url = f'https://{THROTTLED_DOMAIN}/page1' + early_url = f'https://{SECOND_THROTTLED_DOMAIN}/page1' + await two_domain_manager.add_request(late_url) + await two_domain_manager.add_request(early_url) + + with _frozen_clock(): + two_domain_manager.record_domain_delay(late_url, retry_after=timedelta(seconds=30)) + two_domain_manager.record_domain_delay(early_url, retry_after=timedelta(seconds=10)) + status = await two_domain_manager.check_readiness() + + assert status == WaitingStatus(ready_at=CLOCK_START + timedelta(seconds=10)) + + +async def test_empty_cooldown_not_waited(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that a domain in a cooldown with nothing left to do doesn't keep the manager from finishing.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + manager.set_crawl_delay(url, 30) + await manager.add_request(url) + + request = await manager.fetch_next_request() + assert request is not None + await manager.mark_request_as_handled(request) + + assert manager.is_throttled(url) + assert await manager.check_readiness() == FinishedStatus() + + async def test_fetch_prefers_longest_overdue_domain( two_domain_manager: ThrottlingRequestManager[RequestQueue], ) -> None: @@ -673,7 +703,7 @@ async def test_reclaim_request_routes_to_sub_manager( await manager.reclaim_request(request) # Should be back in the sub-manager. - assert not await manager._sub_managers[THROTTLED_DOMAIN].is_empty() + assert await manager._sub_managers[THROTTLED_DOMAIN].check_readiness() == ReadyStatus() async def test_reclaim_request_routes_to_inner( @@ -690,7 +720,7 @@ async def test_reclaim_request_routes_to_inner( await manager.reclaim_request(request) # Should be back in inner manager. - assert not await inner_queue.is_empty() + assert await inner_queue.check_readiness() == ReadyStatus() async def test_mark_request_as_handled_routes_to_sub_manager( @@ -737,8 +767,8 @@ async def test_reclaim_returns_inner_request_to_inner( await manager.reclaim_request(request) - assert not await inner_queue.is_empty() - assert await manager._sub_managers[THROTTLED_DOMAIN].is_empty() + assert await inner_queue.check_readiness() == ReadyStatus() + assert await manager._sub_managers[THROTTLED_DOMAIN].check_readiness() == FinishedStatus() async def test_handled_inner_request_finishes_inner( @@ -753,8 +783,8 @@ async def test_handled_inner_request_finishes_inner( await manager.mark_request_as_handled(request) - assert await inner_queue.is_finished() is True - assert await manager.is_finished() is True + assert await inner_queue.check_readiness() == FinishedStatus() + assert await manager.check_readiness() == FinishedStatus() async def test_purge_forgets_in_flight_inner_requests( @@ -774,7 +804,7 @@ async def test_purge_forgets_in_flight_inner_requests( assert request is not None await manager.mark_request_as_handled(request) - assert await manager._sub_managers[THROTTLED_DOMAIN].is_finished() is True + assert await manager._sub_managers[THROTTLED_DOMAIN].check_readiness() == FinishedStatus() async def test_failed_completion_keeps_its_inner_routing_for_the_retry( @@ -799,8 +829,8 @@ async def test_failed_completion_keeps_its_inner_routing_for_the_retry( monkeypatch.undo() await manager.mark_request_as_handled(request) - assert await inner_queue.is_finished() is True - assert await manager.is_finished() is True + assert await inner_queue.check_readiness() == FinishedStatus() + assert await manager.check_readiness() == FinishedStatus() async def test_shared_unique_key_does_not_reroute_sub_manager_request( @@ -826,8 +856,8 @@ async def test_shared_unique_key_does_not_reroute_sub_manager_request( await manager.reclaim_request(sub_request) # The reclaim has to land in the sub-manager; inner keeps its own request in flight. - assert not await manager._sub_managers[THROTTLED_DOMAIN].is_empty() - assert await inner_queue.is_empty() is True + assert await manager._sub_managers[THROTTLED_DOMAIN].check_readiness() == ReadyStatus() + assert await inner_queue.check_readiness() == WaitingStatus() async def test_get_handled_count_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -861,27 +891,19 @@ async def test_get_total_count_aggregates(manager: ThrottlingRequestManager[Requ assert await manager.get_total_count() == 2 -async def test_is_empty_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """is_empty should return False if any manager that is not in a cooldown has requests.""" - assert await manager.is_empty() is True +async def test_readiness_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that the readiness follows a request through the sub-manager it's routed to.""" + assert await manager.check_readiness() == FinishedStatus() await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1') - assert await manager.is_empty() is False - - -async def test_is_finished_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """is_finished should return True only when all managers are finished.""" - assert await manager.is_finished() is True - - url = f'https://{THROTTLED_DOMAIN}/page1' - await manager.add_request(url) - assert await manager.is_finished() is False + assert await manager.check_readiness() == ReadyStatus() request = await manager.fetch_next_request() assert request is not None - await manager.mark_request_as_handled(request) + assert await manager.check_readiness() == WaitingStatus() - assert await manager.is_finished() is True + await manager.mark_request_as_handled(request) + assert await manager.check_readiness() == FinishedStatus() async def test_drop_clears_all( @@ -913,8 +935,8 @@ async def test_purge_clears_requests_and_resets_throttle_state( # Domain configuration is preserved. assert THROTTLED_DOMAIN in manager._domain_states - # Manager is empty. - assert await manager.is_empty() + # Manager is finished. + assert await manager.check_readiness() == FinishedStatus() # Transient backoff state is reset. assert state.consecutive_429_count == 0 assert not manager._is_domain_throttled(THROTTLED_DOMAIN) @@ -935,7 +957,7 @@ async def test_sub_managers_opened_for_every_configured_domain( assert manager._sub_managers == {} - await manager.is_empty() + await manager.check_readiness() assert set(manager._sub_managers) == set(domains) @@ -954,7 +976,7 @@ async def test_sub_managers_opened_once( service_locator=service_locator, ) - await asyncio.gather(manager.is_empty(), manager.is_finished(), manager.fetch_next_request()) + await asyncio.gather(manager.check_readiness(), manager.check_readiness(), manager.fetch_next_request()) assert opener.await_count == len(domains) @@ -983,9 +1005,9 @@ async def open_once_failing(*, alias: str, **kwargs: Any) -> RequestQueue: ) with pytest.raises(RuntimeError, match='storage unavailable'): - await manager.is_empty() + await manager.check_readiness() - assert await manager.is_empty() is True + assert await manager.check_readiness() == FinishedStatus() # The domain that opened before its sibling failed is reused, not opened a second time. opened_aliases = [call.kwargs['alias'] for call in opener.await_args_list] @@ -1001,7 +1023,7 @@ async def test_read_path_after_drop_reopens_sub_managers( await manager.drop() assert manager._sub_managers == {} - assert await manager.is_empty() is True + assert await manager.check_readiness() == FinishedStatus() assert set(manager._sub_managers) == set(TEST_DOMAINS) @@ -1013,8 +1035,7 @@ async def test_read_paths_reopen_persisted_sub_queues(fs_service_locator: Servic restarted = await _restart_fs_manager(fs_service_locator) - assert await restarted.is_finished() is False - assert await restarted.is_empty() is False + assert await restarted.check_readiness() == ReadyStatus() assert await restarted.get_total_count() == 2 request = await restarted.fetch_next_request() @@ -1031,12 +1052,12 @@ async def test_default_purge_on_start_empties_persisted_sub_queues(fs_service_lo fs_service_locator.storage_instance_manager.clear_cache() purging_locator = ServiceLocator(configuration=Configuration(), storage_client=FileSystemStorageClient()) purged = await _open_fs_manager(purging_locator) - assert await purged.is_empty() is True + assert await purged.check_readiness() == FinishedStatus() # Reopen without purging: an empty queue proves the requests were deleted, not just hidden. restarted = await _restart_fs_manager(fs_service_locator) assert await restarted.get_total_count() == 0 - assert await restarted.is_finished() is True + assert await restarted.check_readiness() == FinishedStatus() async def test_purge_empties_sub_queues_from_a_previous_run(fs_service_locator: ServiceLocator) -> None: @@ -1082,8 +1103,8 @@ async def test_reclaim_routes_to_sub_manager_after_restart(fs_service_locator: S await restarted.reclaim_request(request) - assert not await restarted._sub_managers[THROTTLED_DOMAIN].is_empty() - assert await restarted.inner.is_empty() + assert await restarted._sub_managers[THROTTLED_DOMAIN].check_readiness() == ReadyStatus() + assert await restarted.inner.check_readiness() == FinishedStatus() def make_stalled(manager: ThrottlingRequestManager[Any], clock: MagicMock, url: str) -> None: @@ -1100,15 +1121,16 @@ async def test_stall_reported_after_window(manager: ThrottlingRequestManager[Req with _frozen_clock() as clock: manager.record_domain_delay(url) - assert await manager.get_stall_reason() is None + ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until + assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) manager.record_domain_delay(url) - reason = await manager.get_stall_reason() + status = await manager.check_readiness() - assert reason is not None - assert f'"{THROTTLED_DOMAIN}" (901s)' in reason - assert '`max_domain_stall` (900s)' in reason + assert isinstance(status, StalledStatus) + assert f'"{THROTTLED_DOMAIN}" (901s)' in status.reason + assert '`max_domain_stall` (900s)' in status.reason async def test_suppressed_429_extends_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1134,7 +1156,8 @@ async def test_first_429_after_idle_not_stalled(manager: ThrottlingRequestManage with _frozen_clock() as clock: clock.now.return_value += manager._max_domain_stall * 2 manager.record_domain_delay(url) - assert await manager.get_stall_reason() is None + ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until + assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) async def test_old_rate_limit_not_stalled(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1145,7 +1168,7 @@ async def test_old_rate_limit_not_stalled(manager: ThrottlingRequestManager[Requ with _frozen_clock() as clock: manager.record_domain_delay(url) clock.now.return_value += manager._max_domain_stall + timedelta(seconds=1) - assert await manager.get_stall_reason() is None + assert await manager.check_readiness() == ReadyStatus() @pytest.mark.parametrize( @@ -1171,10 +1194,11 @@ async def test_handled_request_clears_stall( await manager.add_request(f'https://{THROTTLED_DOMAIN}/page2') make_stalled(manager, clock, url) - assert await manager.get_stall_reason() is not None + assert isinstance(await manager.check_readiness(), StalledStatus) await manager.mark_request_as_handled(request) - assert await manager.get_stall_reason() is None + ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until + assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) async def test_reclaim_keeps_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1188,7 +1212,7 @@ async def test_reclaim_keeps_stall(manager: ThrottlingRequestManager[RequestQueu make_stalled(manager, clock, url) await manager.reclaim_request(request) - assert await manager.get_stall_reason() is not None + assert isinstance(await manager.check_readiness(), StalledStatus) async def test_record_success_keeps_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1199,7 +1223,7 @@ async def test_record_success_keeps_stall(manager: ThrottlingRequestManager[Requ with _frozen_clock() as clock: make_stalled(manager, clock, url) manager.record_success(url) - assert await manager.get_stall_reason() is not None + assert isinstance(await manager.check_readiness(), StalledStatus) @pytest.mark.parametrize( @@ -1219,7 +1243,7 @@ async def test_ready_work_masks_stall( with _frozen_clock() as clock: make_stalled(two_domain_manager, clock, url) - assert await two_domain_manager.get_stall_reason() is None + assert await two_domain_manager.check_readiness() == ReadyStatus() async def test_waiting_domain_masks_stall(two_domain_manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1232,7 +1256,8 @@ async def test_waiting_domain_masks_stall(two_domain_manager: ThrottlingRequestM with _frozen_clock() as clock: make_stalled(two_domain_manager, clock, url) two_domain_manager.record_domain_delay(waiting_url, retry_after=timedelta(seconds=60)) - assert await two_domain_manager.get_stall_reason() is None + ready_at = two_domain_manager._domain_states[SECOND_THROTTLED_DOMAIN].throttled_until + assert await two_domain_manager.check_readiness() == WaitingStatus(ready_at=ready_at) async def test_empty_waiting_domain_does_not_mask_stall( @@ -1245,11 +1270,11 @@ async def test_empty_waiting_domain_does_not_mask_stall( with _frozen_clock() as clock: make_stalled(two_domain_manager, clock, url) two_domain_manager.record_domain_delay(f'https://{SECOND_THROTTLED_DOMAIN}/page1') - reason = await two_domain_manager.get_stall_reason() + status = await two_domain_manager.check_readiness() - assert reason is not None - assert THROTTLED_DOMAIN in reason - assert SECOND_THROTTLED_DOMAIN not in reason + assert isinstance(status, StalledStatus) + assert THROTTLED_DOMAIN in status.reason + assert SECOND_THROTTLED_DOMAIN not in status.reason async def test_lapsed_backoff_does_not_mask_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: @@ -1260,18 +1285,18 @@ async def test_lapsed_backoff_does_not_mask_stall(manager: ThrottlingRequestMana with _frozen_clock() as clock: make_stalled(manager, clock, url) clock.now.return_value = manager._domain_states[THROTTLED_DOMAIN].throttled_until + timedelta(seconds=1) - assert await manager.get_stall_reason() is not None + assert isinstance(await manager.check_readiness(), StalledStatus) async def test_empty_domain_not_stalled(manager: ThrottlingRequestManager[RequestQueue]) -> None: """A domain with no requests left is finished, not stalled.""" with _frozen_clock() as clock: make_stalled(manager, clock, f'https://{THROTTLED_DOMAIN}/page1') - assert await manager.get_stall_reason() is None + assert await manager.check_readiness() == FinishedStatus() -async def test_is_empty_skips_stall_candidate(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """A stalled domain doesn't count for `is_empty`, but its request can still be fetched.""" +async def test_stall_candidate_not_ready(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that a stalled domain doesn't read as ready, but its request can still be fetched.""" url = f'https://{THROTTLED_DOMAIN}/page1' await manager.add_request(url) @@ -1279,13 +1304,38 @@ async def test_is_empty_skips_stall_candidate(manager: ThrottlingRequestManager[ make_stalled(manager, clock, url) clock.now.return_value = manager._domain_states[THROTTLED_DOMAIN].throttled_until + timedelta(seconds=1) - assert await manager.is_empty() is True + assert isinstance(await manager.check_readiness(), StalledStatus) request = await manager.fetch_next_request() assert request is not None assert request.url == url +async def test_stall_ignores_work_in_flight(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that a request in progress elsewhere doesn't keep a stall from being reported.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(f'https://{NON_THROTTLED_DOMAIN}/page1') + assert await manager.fetch_next_request() is not None + await manager.add_request(url) + + with _frozen_clock() as clock: + make_stalled(manager, clock, url) + assert isinstance(await manager.check_readiness(), StalledStatus) + + +async def test_stall_candidate_adding_in_background(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that requests still being added for a stalled domain keep the manager from finishing.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await manager.add_request(url) + sub_manager = manager._sub_managers[THROTTLED_DOMAIN] + + with _frozen_clock() as clock, patch.object(sub_manager, 'check_readiness', return_value=WaitingStatus()): + make_stalled(manager, clock, url) + status = await manager.check_readiness() + + assert status == WaitingStatus() + + async def test_purge_resets_stall_and_migrations( manager: ThrottlingRequestManager[RequestQueue], inner_queue: RequestQueue, @@ -1341,11 +1391,11 @@ async def test_inner_request_migrates_while_throttled( assert await manager.fetch_next_request() is None - assert await inner_queue.is_finished() is True + assert await inner_queue.check_readiness() == FinishedStatus() assert await manager._sub_managers[THROTTLED_DOMAIN].get_total_count() == 1 assert await manager.get_total_count() == 1 assert await manager.get_handled_count() == 0 - assert await manager.is_finished() is False + assert isinstance(await manager.check_readiness(), WaitingStatus) async def test_inner_request_dispatched_when_not_throttled( @@ -1442,10 +1492,10 @@ async def test_stall_sees_migrated_inner_work( assert request is not None make_stalled(manager, clock, url) await manager.reclaim_request(request) - assert await manager.get_stall_reason() is None + assert not isinstance(await manager.check_readiness(), StalledStatus) assert await manager.fetch_next_request() is None - assert await manager.get_stall_reason() is not None + assert isinstance(await manager.check_readiness(), StalledStatus) # ── Utility Tests ────────────────────────────────────── From 6899836d0218f443bdba6c0c18fa7daf5b99d94e Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Sun, 18 Oct 2026 18:46:54 +0000 Subject: [PATCH 2/4] add `status` literal in statuses --- .../code_examples/run_parallel_crawlers.py | 3 +- docs/upgrading/upgrading_to_v2.md | 11 +++-- src/crawlee/crawlers/_basic/_basic_crawler.py | 16 +++---- .../_request_manager_tandem.py | 3 +- .../_throttling_request_manager.py | 28 +++++------ src/crawlee/request_loaders/_types.py | 14 +++++- src/crawlee/request_loaders/_utils.py | 46 +++++++++---------- .../test_request_source_status.py | 14 ++++++ 8 files changed, 78 insertions(+), 57 deletions(-) diff --git a/docs/examples/code_examples/run_parallel_crawlers.py b/docs/examples/code_examples/run_parallel_crawlers.py index 608c45a968..18ef309892 100644 --- a/docs/examples/code_examples/run_parallel_crawlers.py +++ b/docs/examples/code_examples/run_parallel_crawlers.py @@ -7,7 +7,6 @@ PlaywrightCrawler, PlaywrightCrawlingContext, ) -from crawlee.request_loaders import FinishedStatus from crawlee.sessions import SessionPool from crawlee.storages import RequestQueue @@ -80,7 +79,7 @@ async def handle_parsel(context: ParselCrawlingContext) -> None: await parsel_crawler.run(['https://crawlee.dev/blog']) # Wait for the Playwright crawler to finish processing all requests - while not isinstance(await playwright_rq.check_readiness(), FinishedStatus): + while (await playwright_rq.check_readiness()).status != 'finished': playwright_crawler.log.info('Waiting for Playwright crawler to finish...') await asyncio.sleep(5) diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index ff42ed1b08..3eed91cc76 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -177,22 +177,23 @@ The autoscaling loggers moved together with the package. `crawlee._autoscaling.a - `StalledStatus`: the source holds requests it can't make progress on, as explained in `reason`. `ThrottlingRequestManager` reports it, and `RequestManagerTandem` passes it on from the manager it wraps. - `FinishedStatus`: every request has been handled. +Each of these classes also has a `status` attribute set to `'ready'`, `'waiting'`, `'stalled'` or `'finished'`, so a check doesn't need an import. + The change covers `RequestQueue`, `RequestList`, `SitemapRequestLoader`, `RequestManagerTandem` and `ThrottlingRequestManager`. `ThrottlingRequestManager.get_stall_reason` is removed too, since a stall is now reported as `StalledStatus`. Storage clients keep `is_empty` and `is_finished`. ```python -from crawlee.request_loaders import FinishedStatus, ReadyStatus - # v1 has_requests = not await rq.is_empty() done = await rq.is_finished() # v2 -has_requests = isinstance(await rq.check_readiness(), ReadyStatus) -done = isinstance(await rq.check_readiness(), FinishedStatus) +readiness = await rq.check_readiness() +has_requests = readiness.status == 'ready' +done = readiness.status == 'finished' ``` If you have a custom request loader or manager, replace `is_empty` and `is_finished` with `check_readiness`. Return `ReadyStatus` as soon as a request can be fetched, before checking anything else, since the crawler calls the method often. Never return `FinishedStatus` while requests are still pending or in progress. A loader doesn't return `StalledStatus`. A `None` from `fetch_next_request` no longer means the loader is finished. It only means nothing is pending right now. `RequestManagerTandem` now moves a request from its loader only when the loader is ready, so a `SitemapRequestLoader` that's still loading sitemaps no longer holds up the requests already in the manager. -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. +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. The error it 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." diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index cfcb9fe81f..e6c907f4af 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/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 FinishedStatus, ReadyStatus, StalledStatus, ThrottlingRequestManager +from crawlee.request_loaders import ThrottlingRequestManager from crawlee.router import Router from crawlee.sessions import SessionPool from crawlee.statistics import Statistics, StatisticsState @@ -1500,22 +1500,20 @@ async def __is_finished_function(self) -> bool: return False request_manager = await self.get_request_manager() - status = await request_manager.check_readiness() + readiness = await request_manager.check_readiness() # Check for a stall only when the pool is idle. An in-flight request can still enqueue new work or end the # stall. The pool's own requests are what counts, not those of another crawler sharing the budget. if ( - isinstance(status, StalledStatus) + readiness.status == 'stalled' and self._autoscaled_pool is not None and self._autoscaled_pool.own_concurrency == 0 ): - error = PersistentRateLimitError(f'Giving up: {status.reason}') - self._logger.error( - 'Giving up the crawl because a domain keeps rate-limiting every request.', exc_info=error - ) + error = PersistentRateLimitError(f'Giving up: {readiness.reason}') + self._logger.error("Giving up the crawl because the request manager can't make progress.", exc_info=error) raise error - return isinstance(status, FinishedStatus) + return readiness.status == 'finished' async def __is_task_ready_function(self) -> bool: self._stop_if_max_requests_count_exceeded() @@ -1527,7 +1525,7 @@ async def __is_task_ready_function(self) -> bool: return False request_manager = await self.get_request_manager() - return isinstance(await request_manager.check_readiness(), ReadyStatus) + return (await request_manager.check_readiness()).status == 'ready' async def __run_task_function(self, task_id: str) -> None: # noqa: ARG002 request_manager = await self.get_request_manager() diff --git a/src/crawlee/request_loaders/_request_manager_tandem.py b/src/crawlee/request_loaders/_request_manager_tandem.py index 8a992dc6b4..b41e847e91 100644 --- a/src/crawlee/request_loaders/_request_manager_tandem.py +++ b/src/crawlee/request_loaders/_request_manager_tandem.py @@ -10,7 +10,6 @@ from crawlee._utils.docs import docs_group from crawlee.request_loaders import RequestManager -from ._types import ReadyStatus from ._utils import join_request_source_statuses if TYPE_CHECKING: @@ -81,7 +80,7 @@ async def add_requests( async def fetch_next_request(self) -> Request | None: # Only the loader's own readiness decides the transfer, so a loader that's still loading doesn't block the # manager's requests. - if isinstance(await self._read_only_loader.check_readiness(), ReadyStatus): + if (await self._read_only_loader.check_readiness()).status == 'ready': transferred = await self._transfer_next_request() # The request was dropped, so skip the manager this round. The next call picks up where this one left off. if not transferred: diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index a98bb73483..86685aacc3 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -15,7 +15,7 @@ from crawlee._utils.docs import docs_group from crawlee.request_loaders._request_manager import RequestManager -from ._types import FinishedStatus, ReadyStatus, StalledStatus, WaitingStatus +from ._types import FinishedStatus, StalledStatus, WaitingStatus from ._utils import join_request_source_statuses if TYPE_CHECKING: @@ -329,34 +329,34 @@ async def check_readiness(self) -> RequestSourceStatus: else: cooling.append(domain) - status: RequestSourceStatus = FinishedStatus() - for manager_status in await asyncio.gather(*(manager.check_readiness() for manager in dispatchable)): - status = join_request_source_statuses(status, manager_status) - if isinstance(status, ReadyStatus): - return status + probed: RequestSourceStatus = FinishedStatus() + for readiness in await asyncio.gather(*(manager.check_readiness() for manager in dispatchable)): + probed = join_request_source_statuses(probed, readiness) + if probed.status == 'ready': + return probed cooling_statuses = await asyncio.gather(*(self._sub_managers[domain].check_readiness() for domain in cooling)) ready_at = min( ( self._domain_states[domain].throttled_until - for domain, cooling_status in zip(cooling, cooling_statuses, strict=True) - if not isinstance(cooling_status, FinishedStatus) + for domain, readiness in zip(cooling, cooling_statuses, strict=True) + if readiness.status != 'finished' ), default=None, ) # A domain waiting out a cooldown with work left will still make progress, so the crawl isn't stuck yet. if ready_at is not None: - return join_request_source_statuses(status, WaitingStatus(ready_at=ready_at)) + return join_request_source_statuses(probed, WaitingStatus(ready_at=ready_at)) candidate_statuses = await asyncio.gather( *(self._sub_managers[domain].check_readiness() for domain in candidates) ) stalled = [ domain - for domain, candidate_status in zip(candidates, candidate_statuses, strict=True) - if isinstance(candidate_status, ReadyStatus) + for domain, readiness in zip(candidates, candidate_statuses, strict=True) + if readiness.status == 'ready' ] if stalled: @@ -373,10 +373,10 @@ async def check_readiness(self) -> RequestSourceStatus: ) # A candidate's requests that are in progress or still being added in the background aren't finished yet. - for candidate_status in candidate_statuses: - status = join_request_source_statuses(status, candidate_status) + for readiness in candidate_statuses: + probed = join_request_source_statuses(probed, readiness) - return status + return probed 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. diff --git a/src/crawlee/request_loaders/_types.py b/src/crawlee/request_loaders/_types.py index 80faf687ad..3f30dfeb8b 100644 --- a/src/crawlee/request_loaders/_types.py +++ b/src/crawlee/request_loaders/_types.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar, Literal from crawlee._utils.docs import docs_group @@ -14,6 +14,9 @@ class ReadyStatus: """The next `fetch_next_request` call is expected to return a request.""" + status: ClassVar[Literal['ready']] = 'ready' + """The status name, so a check doesn't need to import the class.""" + @docs_group('Request loaders') @dataclass(frozen=True, kw_only=True) @@ -23,6 +26,9 @@ class WaitingStatus: Requests are in progress, are being added in the background, or are held back until `ready_at`. """ + status: ClassVar[Literal['waiting']] = 'waiting' + """The status name, so a check doesn't need to import the class.""" + ready_at: datetime | None = None """When the source expects to become ready, as a timezone-aware UTC time. `None` when the wait has no clock.""" @@ -32,6 +38,9 @@ class WaitingStatus: class StalledStatus: """The source holds requests it can't make progress on. Only a manager that paces its own dispatch reports it.""" + status: ClassVar[Literal['stalled']] = 'stalled' + """The status name, so a check doesn't need to import the class.""" + reason: str """Why the source can't make progress.""" @@ -41,6 +50,9 @@ class StalledStatus: class FinishedStatus: """Every request has been handled.""" + status: ClassVar[Literal['finished']] = 'finished' + """The status name, so a check doesn't need to import the class.""" + RequestLoaderStatus = ReadyStatus | WaitingStatus | FinishedStatus """Readiness of a request loader. Loaders never stall.""" diff --git a/src/crawlee/request_loaders/_utils.py b/src/crawlee/request_loaders/_utils.py index 3ab98a1870..e86634b51c 100644 --- a/src/crawlee/request_loaders/_utils.py +++ b/src/crawlee/request_loaders/_utils.py @@ -2,37 +2,35 @@ from typing import TYPE_CHECKING -from ._types import ReadyStatus, StalledStatus, WaitingStatus - if TYPE_CHECKING: from ._types import RequestSourceStatus def join_request_source_statuses( - first_status: RequestSourceStatus, - second_status: RequestSourceStatus, + first: RequestSourceStatus, + second: RequestSourceStatus, ) -> RequestSourceStatus: """Combine two statuses with the precedence ready > waiting > stalled > finished. Of two waiting statuses, the one with the earlier known `ready_at` wins. """ - if isinstance(first_status, ReadyStatus): - return first_status - if isinstance(second_status, ReadyStatus): - return second_status - - if isinstance(first_status, WaitingStatus): - if not isinstance(second_status, WaitingStatus) or second_status.ready_at is None: - return first_status - if first_status.ready_at is not None and first_status.ready_at <= second_status.ready_at: - return first_status - return second_status - - if isinstance(second_status, WaitingStatus): - return second_status - - if isinstance(first_status, StalledStatus): - return first_status - - # `first_status` is finished, so `second_status` decides. - return second_status + if first.status == 'ready': + return first + if second.status == 'ready': + return second + + if first.status == 'waiting': + if second.status != 'waiting' or second.ready_at is None: + return first + if first.ready_at is not None and first.ready_at <= second.ready_at: + return first + return second + + if second.status == 'waiting': + return second + + if first.status == 'stalled': + return first + + # `first` is finished, so `second` decides. + return second diff --git a/tests/unit/request_loaders/test_request_source_status.py b/tests/unit/request_loaders/test_request_source_status.py index 76357289a4..891d804789 100644 --- a/tests/unit/request_loaders/test_request_source_status.py +++ b/tests/unit/request_loaders/test_request_source_status.py @@ -58,3 +58,17 @@ def test_join_keeps_known_ready_at() -> None: assert join_request_source_statuses(known, WAITING) == known assert join_request_source_statuses(known, FINISHED) == known assert join_request_source_statuses(STALLED, known) == known + + +@pytest.mark.parametrize( + ('readiness', 'name'), + [ + pytest.param(READY, 'ready', id='ready'), + pytest.param(WAITING, 'waiting', id='waiting'), + pytest.param(STALLED, 'stalled', id='stalled'), + pytest.param(FINISHED, 'finished', id='finished'), + ], +) +def test_status_name(readiness: RequestSourceStatus, name: str) -> None: + """Test that each status carries its own name.""" + assert readiness.status == name From c4e5eb59bede432aaa28876781e9b2d2f3d690cd Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Wed, 21 Oct 2026 16:59:24 +0000 Subject: [PATCH 3/4] fixes --- .../_request_manager_tandem.py | 28 ++++++++----------- .../_throttling_request_manager.py | 16 +++++++---- src/crawlee/request_loaders/_types.py | 9 ++++-- src/crawlee/request_loaders/_utils.py | 1 + .../crawlers/_basic/test_basic_crawler.py | 20 +++++++++---- .../storages/test_request_manager_tandem.py | 10 +++++++ tests/unit/test_throttling_request_manager.py | 10 ++++--- 7 files changed, 60 insertions(+), 34 deletions(-) diff --git a/src/crawlee/request_loaders/_request_manager_tandem.py b/src/crawlee/request_loaders/_request_manager_tandem.py index b41e847e91..006ec9477a 100644 --- a/src/crawlee/request_loaders/_request_manager_tandem.py +++ b/src/crawlee/request_loaders/_request_manager_tandem.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio from datetime import timedelta from logging import getLogger from typing import TYPE_CHECKING @@ -47,10 +46,12 @@ async def get_total_count(self) -> int: @override async def check_readiness(self) -> RequestSourceStatus: - loader_status, manager_status = await asyncio.gather( - self._read_only_loader.check_readiness(), self._read_write_manager.check_readiness() - ) - return join_request_source_statuses(loader_status, manager_status) + loader_readiness = await self._read_only_loader.check_readiness() + if loader_readiness.status == 'ready': + return loader_readiness + + manager_readiness = await self._read_write_manager.check_readiness() + return join_request_source_statuses(loader_readiness, manager_readiness) @override async def add_request(self, request: str | Request, *, forefront: bool = False) -> ProcessedRequest | None: @@ -80,20 +81,13 @@ async def add_requests( async def fetch_next_request(self) -> Request | None: # Only the loader's own readiness decides the transfer, so a loader that's still loading doesn't block the # manager's requests. - if (await self._read_only_loader.check_readiness()).status == 'ready': - transferred = await self._transfer_next_request() - # The request was dropped, so skip the manager this round. The next call picks up where this one left off. - if not transferred: - return None + if (await self._read_only_loader.check_readiness()).status != 'ready': + return await self._read_write_manager.fetch_next_request() - return await self._read_write_manager.fetch_next_request() - - async def _transfer_next_request(self) -> bool: - """Move the loader's next request into the manager, returning `False` if it had to be dropped.""" request = await self._read_only_loader.fetch_next_request() if not request: - return True + return await self._read_write_manager.fetch_next_request() try: await self._read_write_manager.add_request(request, forefront=True) @@ -103,13 +97,13 @@ async def _transfer_next_request(self) -> bool: extra={'url': request.url, 'unique_key': request.unique_key}, ) - return False + return None finally: # Mark it as processed so that the `request` doesn't get stuck in the `in_progress` status # in `RequestLoader` await self._read_only_loader.mark_request_as_handled(request) - return True + return await self._read_write_manager.fetch_next_request() @override async def reclaim_request(self, request: Request, *, forefront: bool = False) -> None: diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index 86685aacc3..9d202a7999 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -311,7 +311,8 @@ async def check_readiness(self) -> RequestSourceStatus: Requests of a domain in a cooldown count as outstanding work, so the crawl waits until the earliest cooldown ends. A domain is stalled once it has rate-limited every request for longer than `max_domain_stall`. It's - reported only when nothing else can be dispatched and no domain in a cooldown has requests left. + reported only when nothing else is ready or waiting, such as a domain in a cooldown with requests left, or + requests in progress or still being added. """ await self._ensure_sub_managers() now = datetime.now(UTC) @@ -359,6 +360,11 @@ async def check_readiness(self) -> RequestSourceStatus: if readiness.status == 'ready' ] + # A candidate's requests that are in progress or still being added in the background aren't finished yet. + for readiness in candidate_statuses: + if readiness.status != 'ready': + probed = join_request_source_statuses(probed, readiness) + if stalled: summary = ', '.join( f'"{domain}" ({(now - since).total_seconds():.0f}s)' @@ -366,15 +372,13 @@ async def check_readiness(self) -> RequestSourceStatus: if (since := self._domain_states[domain].rate_limited_since) is not None ) window = self._max_domain_stall.total_seconds() - return StalledStatus( + stall = StalledStatus( f'{summary} rate-limited every request for longer than `max_domain_stall` ({window:.0f}s). Waiting ' "longer will not help - lower the crawler's concurrency, or drop these domains. Their requests are " 'still queued, so re-running with `purge_on_start` disabled will resume them if the rate limit lifts.' ) - - # A candidate's requests that are in progress or still being added in the background aren't finished yet. - for readiness in candidate_statuses: - probed = join_request_source_statuses(probed, readiness) + # Requests still in progress or being added elsewhere may yet bring work, so they outrank the stall. + return join_request_source_statuses(probed, stall) return probed diff --git a/src/crawlee/request_loaders/_types.py b/src/crawlee/request_loaders/_types.py index 3f30dfeb8b..6c59249478 100644 --- a/src/crawlee/request_loaders/_types.py +++ b/src/crawlee/request_loaders/_types.py @@ -30,13 +30,18 @@ class WaitingStatus: """The status name, so a check doesn't need to import the class.""" ready_at: datetime | None = None - """When the source expects to become ready, as a timezone-aware UTC time. `None` when the wait has no clock.""" + """When the source expects to become ready, as a timezone-aware UTC time. It may become ready earlier, so a consumer + keeps polling either way. `None` when the wait has no clock. + """ @docs_group('Request loaders') @dataclass(frozen=True) class StalledStatus: - """The source holds requests it can't make progress on. Only a manager that paces its own dispatch reports it.""" + """The source holds requests it can't make progress on. + + Only a manager that paces its own dispatch, or one that wraps such a manager, reports it. + """ status: ClassVar[Literal['stalled']] = 'stalled' """The status name, so a check doesn't need to import the class.""" diff --git a/src/crawlee/request_loaders/_utils.py b/src/crawlee/request_loaders/_utils.py index e86634b51c..4937e7f964 100644 --- a/src/crawlee/request_loaders/_utils.py +++ b/src/crawlee/request_loaders/_utils.py @@ -20,6 +20,7 @@ def join_request_source_statuses( return second if first.status == 'waiting': + # A known `ready_at` lets a consumer schedule a wake-up on top of regular polling, so it beats an unknown one. if second.status != 'waiting' or second.ready_at is None: return first if first.ready_at is not None and first.ready_at <= second.ready_at: diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 4e23ab8db8..a1068e973f 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -2916,7 +2916,9 @@ async def handler(context: BasicCrawlingContext) -> None: if context.request.url == THROTTLED_URL: throttler.record_domain_delay(context.request.url) raise RequestThrottledError - if context.request.url == 'https://free.placeholder.com/slow': + # The slow request belongs to the stalled domain, since a request in progress elsewhere would keep the stall + # from being reported. + if context.request.url == 'https://throttled.placeholder.com/slow': stall_statuses.append(await poll_until_condition(throttler.check_readiness, is_stalled, timeout=10)) # Two more pool checks, so at least one ran entirely while the stall was live and this request in flight. checks = is_finished.await_count @@ -2928,7 +2930,8 @@ async def handler(context: BasicCrawlingContext) -> None: patch.object(crawler, '_BasicCrawler__is_finished_function', wraps=is_finished_function) as is_finished, pytest.raises(PersistentRateLimitError), ): - await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) + # The slow request goes first, since a deferred `THROTTLED_URL` returns to the front of the domain's queue. + await crawler.run(['https://throttled.placeholder.com/slow', THROTTLED_URL]) assert isinstance(stall_statuses[0], StalledStatus) assert 'https://free.placeholder.com/next' in visited @@ -2945,7 +2948,8 @@ async def test_max_requests_per_crawl_precedes_stall() -> None: concurrency_settings=ConcurrencySettings(min_concurrency=2, desired_concurrency=2), ) calls = 0 - stall_statuses = list[RequestSourceStatus]() + stall_live = list[bool]() + state = throttler._domain_states['throttled.placeholder.com'] @crawler.router.default_handler async def handler(context: BasicCrawlingContext) -> None: @@ -2956,11 +2960,17 @@ async def handler(context: BasicCrawlingContext) -> None: if context.request.url == THROTTLED_URL: throttler.record_domain_delay(context.request.url) raise RequestThrottledError - stall_statuses.append(await poll_until_condition(throttler.check_readiness, is_stalled, timeout=10)) + # This request keeps the stall from being reported while in flight, so watch the stall clock. It's of another + # domain, so handling it doesn't reset that clock and the stall is live once the limit is reached. + stall_live.append( + await poll_until_condition( + lambda: state.is_stall_candidate(datetime.now(UTC), MAX_DOMAIN_STALL), timeout=10 + ) + ) stats = await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) - assert isinstance(stall_statuses[0], StalledStatus) + assert stall_live == [True] assert stats.requests_finished == 1 diff --git a/tests/unit/storages/test_request_manager_tandem.py b/tests/unit/storages/test_request_manager_tandem.py index 710c213aa2..d0db897e8a 100644 --- a/tests/unit/storages/test_request_manager_tandem.py +++ b/tests/unit/storages/test_request_manager_tandem.py @@ -124,6 +124,16 @@ async def test_readiness_joins_sources( assert await RequestManagerTandem(loader, manager).check_readiness() == expected +async def test_ready_loader_skips_manager() -> None: + """Test that a ready loader answers without probing the manager.""" + loader = create_autospec(RequestLoader, instance=True, spec_set=True) + loader.check_readiness.return_value = ReadyStatus() + manager = create_autospec(RequestManager, instance=True, spec_set=True) + + assert await RequestManagerTandem(loader, manager).check_readiness() == ReadyStatus() + manager.check_readiness.assert_not_awaited() + + async def test_transfers_only_when_loader_ready() -> None: """Test that a loader that isn't ready doesn't block the manager's requests.""" loader = create_autospec(RequestLoader, instance=True, spec_set=True) diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index ce5fb764bd..9dbbf16890 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -1194,9 +1194,11 @@ async def test_handled_request_clears_stall( await manager.add_request(f'https://{THROTTLED_DOMAIN}/page2') make_stalled(manager, clock, url) - assert isinstance(await manager.check_readiness(), StalledStatus) + state = manager._domain_states[THROTTLED_DOMAIN] + assert state.is_stall_candidate(clock.now.return_value, manager._max_domain_stall) await manager.mark_request_as_handled(request) + assert not state.is_stall_candidate(clock.now.return_value, manager._max_domain_stall) ready_at = manager._domain_states[THROTTLED_DOMAIN].throttled_until assert await manager.check_readiness() == WaitingStatus(ready_at=ready_at) @@ -1311,8 +1313,8 @@ async def test_stall_candidate_not_ready(manager: ThrottlingRequestManager[Reque assert request.url == url -async def test_stall_ignores_work_in_flight(manager: ThrottlingRequestManager[RequestQueue]) -> None: - """Test that a request in progress elsewhere doesn't keep a stall from being reported.""" +async def test_work_in_progress_masks_stall(manager: ThrottlingRequestManager[RequestQueue]) -> None: + """Test that a request in progress elsewhere keeps a stall from being reported.""" url = f'https://{THROTTLED_DOMAIN}/page1' await manager.add_request(f'https://{NON_THROTTLED_DOMAIN}/page1') assert await manager.fetch_next_request() is not None @@ -1320,7 +1322,7 @@ async def test_stall_ignores_work_in_flight(manager: ThrottlingRequestManager[Re with _frozen_clock() as clock: make_stalled(manager, clock, url) - assert isinstance(await manager.check_readiness(), StalledStatus) + assert await manager.check_readiness() == WaitingStatus() async def test_stall_candidate_adding_in_background(manager: ThrottlingRequestManager[RequestQueue]) -> None: From 09c255746b73970a839859c798c3618deedd8a08 Mon Sep 17 00:00:00 2001 From: Max Bohomolov Date: Wed, 21 Oct 2026 18:44:06 +0000 Subject: [PATCH 4/4] update upgrade guide --- docs/upgrading/upgrading_to_v2.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 3eed91cc76..cbcd1a584e 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -197,3 +197,5 @@ If you have a custom request loader or manager, replace `is_empty` and `is_finis A `None` from `fetch_next_request` no longer means the loader is finished. It only means nothing is pending right now. `RequestManagerTandem` now moves a request from its loader only when the loader is ready, so a `SitemapRequestLoader` that's still loading sitemaps no longer holds up the requests already in the manager. 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. The error it 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.