diff --git a/docs/examples/code_examples/run_parallel_crawlers.py b/docs/examples/code_examples/run_parallel_crawlers.py index 5ce94a58fa..18ef309892 100644 --- a/docs/examples/code_examples/run_parallel_crawlers.py +++ b/docs/examples/code_examples/run_parallel_crawlers.py @@ -79,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 await playwright_rq.is_empty(): + 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/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 4bfa59ae51..081e64f83e 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -95,3 +95,33 @@ The `Request.forefront` property is removed. Proxy tiers were the only code that The storage clients now write their JSON data with camelCase keys, the same as Crawlee JS. This covers the metadata, request and record metadata files of `FileSystemStorageClient`, the request and record metadata JSON stored by `RedisStorageClient`, and the request JSON stored by `SqlStorageClient`. For example, `item_count` in a dataset's `__metadata__.json` becomes `itemCount`. The storage metadata documents of `RedisStorageClient` keep their snake_case keys. Data written by v1 still loads, so an existing storage directory, Redis instance or database keeps working. Any tooling that reads these files directly needs to handle the camelCase keys. + +## `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. + +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 +# v1 +has_requests = not await rq.is_empty() +done = await rq.is_finished() + +# v2 +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 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 f091c1dff0..6feab528a7 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -1453,21 +1453,17 @@ async def __is_finished_function(self) -> bool: if self._keep_alive: return False + request_manager = await self.get_request_manager() + 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. - if ( - isinstance(self._request_manager, ThrottlingRequestManager) - and self._autoscaled_pool.current_concurrency == 0 - and (reason := await self._request_manager.get_stall_reason()) is not None - ): - error = PersistentRateLimitError(f'Giving up: {reason}') - self._logger.error( - 'Giving up the crawl because a domain keeps rate-limiting every request.', exc_info=error - ) + if readiness.status == 'stalled' and self._autoscaled_pool.current_concurrency == 0: + 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 - request_manager = await self.get_request_manager() - return await request_manager.is_finished() + return readiness.status == 'finished' async def __is_task_ready_function(self) -> bool: self._stop_if_max_requests_count_exceeded() @@ -1479,7 +1475,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 (await request_manager.check_readiness()).status == 'ready' async def __run_task_function(self) -> None: 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..b41e847e91 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,8 @@ from crawlee._utils.docs import docs_group from crawlee.request_loaders import RequestManager +from ._utils import join_request_source_statuses + if TYPE_CHECKING: from collections.abc import Sequence @@ -16,6 +19,8 @@ from crawlee.request_loaders import RequestLoader from crawlee.storage_clients.models import ProcessedRequest + from ._types import RequestSourceStatus + logger = getLogger(__name__) @@ -41,12 +46,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 +78,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 (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 + + 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 +103,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..86685aacc3 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, 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 + 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 - 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, readiness in zip(cooling, cooling_statuses, strict=True) + if readiness.status != 'finished' + ), + 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(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, readiness in zip(candidates, candidate_statuses, strict=True) + if readiness.status == 'ready' + ] + + 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 readiness in candidate_statuses: + probed = join_request_source_statuses(probed, readiness) + + 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 new file mode 100644 index 0000000000..3f30dfeb8b --- /dev/null +++ b/src/crawlee/request_loaders/_types.py @@ -0,0 +1,61 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, ClassVar, Literal + +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.""" + + 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) +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`. + """ + + 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.""" + + +@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.""" + + 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.""" + + +@docs_group('Request loaders') +@dataclass(frozen=True) +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.""" + +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..e86634b51c --- /dev/null +++ b/src/crawlee/request_loaders/_utils.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from ._types import RequestSourceStatus + + +def join_request_source_statuses( + 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 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/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 cb80b41837..e0c092ee6d 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -37,7 +37,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, StatisticsState from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient @@ -69,16 +78,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() @@ -92,17 +97,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: @@ -138,7 +143,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']) @@ -159,7 +164,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']) @@ -2679,8 +2684,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) @@ -2698,18 +2703,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 @@ -2724,6 +2731,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( @@ -2821,7 +2833,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: @@ -2829,7 +2841,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() @@ -2841,7 +2853,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: @@ -2854,7 +2866,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: @@ -2865,19 +2877,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']) + pool = crawler._autoscaled_pool with ( - patch.object(throttler, 'is_finished', wraps=throttler.is_finished) as is_finished, + patch.object(pool, '_is_finished_function', wraps=pool._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 @@ -2892,7 +2905,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: @@ -2903,11 +2916,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 @@ -2942,6 +2955,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() @@ -2961,8 +2986,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( diff --git a/tests/unit/crawlers/_http/test_http_crawler.py b/tests/unit/crawlers/_http/test_http_crawler.py index 6102944f26..debf2e2304 100644 --- a/tests/unit/crawlers/_http/test_http_crawler.py +++ b/tests/unit/crawlers/_http/test_http_crawler.py @@ -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 = crawler._autoscaled_pool._is_finished_function + + async def is_finished_after_backoff() -> bool: + if crawler._autoscaled_pool.current_concurrency == 0: + clock.now.return_value += backoff + return await is_finished() + + monkeypatch.setattr(crawler._autoscaled_pool, '_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..891d804789 --- /dev/null +++ b/tests/unit/request_loaders/test_request_source_status.py @@ -0,0 +1,74 @@ +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 + + +@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 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 ──────────────────────────────────────