diff --git a/docs/guides/request_throttling.mdx b/docs/guides/request_throttling.mdx index e4c99a6c8d..60dd6298a2 100644 --- a/docs/guides/request_throttling.mdx +++ b/docs/guides/request_throttling.mdx @@ -40,6 +40,21 @@ To use request throttling, create a `. All of them are opened the first time you use the manager, so a domain that never receives a request still gets an empty store. + +Opening the sub-managers up front also makes requests that a previous run left behind visible again. Whether they're resumed or discarded depends on `Configuration.purge_on_start`: + +- With the default `purge_on_start=True`, the leftover requests are purged when the sub-manager opens, just like the requests in an unnamed inner queue. +- With `purge_on_start=False`, the leftover requests are picked up and crawled. + +:::warning + +Named storages are exempt from `purge_on_start`, but aliased ones aren't. If you give the inner `RequestQueue` a `name` to make it persistent, the inner queue keeps its requests across a restart while the per-domain stores are still purged. To keep the requests in both, set `purge_on_start=False`. + +::: + :::tip The `ThrottlingRequestManager` is an opt-in feature. If you don't pass it to your crawler, requests are processed normally without any per-domain throttling. diff --git a/src/crawlee/request_loaders/_throttling_request_manager.py b/src/crawlee/request_loaders/_throttling_request_manager.py index d48c49ec92..c7b5389327 100644 --- a/src/crawlee/request_loaders/_throttling_request_manager.py +++ b/src/crawlee/request_loaders/_throttling_request_manager.py @@ -35,8 +35,9 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): """A request manager that wraps another and enforces per-domain delays. - Requests for explicitly configured domains are routed into dedicated sub-managers at insertion time — each request - lives in exactly one manager, eliminating duplication and simplifying deduplication. + Requests for explicitly configured domains are routed into dedicated sub-managers, so each request lives in exactly + one store and is deduplicated there. A request that reached `inner` before its domain was configured stays and is + completed there, without the domain's delay. When `fetch_next_request()` is called, it returns requests from the sub-manager whose domain has been waiting the longest. If all configured domains are throttled, it falls back to the inner manager for non-throttled domains. If @@ -46,10 +47,14 @@ class ThrottlingRequestManager(RequestManager, Generic[TRequestManager]): - HTTP 429 responses (via `record_domain_delay`) - robots.txt crawl-delay directives (via `set_crawl_delay`) - The class is generic over the wrapped manager type. The `request_manager_opener` callback is used to construct - per-domain sub-managers at insertion time, so every sub-manager shares the same `RequestManager` subclass and - backing store as `inner`. The opener must accept `alias`, `storage_client`, and `configuration` keyword arguments - (as `RequestQueue.open` does) and return the same concrete subclass as `inner`. + The class is generic over the wrapped manager type. The first asynchronous operation opens one sub-manager per + configured domain through `request_manager_opener`, so all of them share the subclass and backing store of `inner`; + the synchronous delay methods never open anything. The opener must accept `alias`, `storage_client`, and + `configuration` keyword arguments (as `RequestQueue.open` does) and return the same concrete subclass as `inner`. + + Requests a previous run left in a persistent store become visible again at open. The default `purge_on_start=True` + empties them; `purge_on_start=False` resumes them. Named stores are exempt from that purge and aliased ones are not, + so a named `inner` keeps its requests while the per-domain stores are emptied. ### Usage @@ -86,9 +91,9 @@ def __init__( domains: Explicit list of domain hostnames to throttle. Only requests matching these domains will be routed to per-domain sub-managers. Matching is case-insensitive (hostnames are lowercased) and exact: subdomain wildcards such as `*.example.com` are not supported — list each subdomain explicitly if needed. - request_manager_opener: Async callable used to create per-domain sub-managers at insertion time. Must - accept `alias`, `storage_client`, and `configuration` keyword arguments and return the same concrete - subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`). + request_manager_opener: Async callable used to open one sub-manager per configured domain on first use. + Must accept `alias`, `storage_client`, and `configuration` keyword arguments and return the same + concrete subclass as `inner` (e.g. `RequestQueue.open` when `inner` is a `RequestQueue`). service_locator: Service locator for creating sub-managers. If not provided, defaults to the global service locator, ensuring consistency with the crawler's storage backend. base_delay: Initial delay after the first 429 response from a domain. @@ -101,6 +106,14 @@ def __init__( self._request_manager_opener = request_manager_opener self._domain_states: dict[str, _DomainState] = {d.lower(): _DomainState(domain=d.lower()) for d in domains if d} self._sub_managers: dict[str, TRequestManager] = {} + self._sub_managers_ready = False + self._sub_managers_lock = asyncio.Lock() + self._in_flight_from_inner: set[tuple[str, str]] = set() + """`(unique_key, url)` pairs of configured-domain requests that `fetch_next_request` took from `inner`, where + they live if they were added before their domain was listed, and where they must be completed. The URL is part + of the key because an explicit `unique_key` is only unique per store. Identical pairs held by `inner` and by a + sub-manager are indistinguishable, so their completions can cross; both stores hold the key, so the cost is a + duplicate crawl and a retry without the domain's delay.""" self._new_work_event = asyncio.Event() """Set whenever a request is added or reclaimed. Lets `fetch_next_request` wake from a throttle wait early when fresh work appears, instead of sleeping for the full computed cooldown.""" @@ -112,18 +125,22 @@ def inner(self) -> TRequestManager: @override async def drop(self) -> None: + await self._ensure_sub_managers() await asyncio.gather(self._inner.drop(), *(sm.drop() for sm in self._sub_managers.values())) self._sub_managers.clear() + self._sub_managers_ready = False + self._in_flight_from_inner.clear() @override async def purge(self) -> None: """Empty the inner manager and all sub-managers, and reset transient per-domain throttle state. - The configured domain list and any robots.txt-derived `crawl_delay` are preserved; only the dynamic backoff - state (consecutive 429 counter and `throttled_until`) is cleared. Sub-managers are kept around so they don't - need to be re-opened on the next request — they're just emptied. + The configured domain list and any robots.txt-derived `crawl_delay` are preserved. Only the dynamic backoff + state (consecutive 429 counter and `throttled_until`) is cleared. Sub-managers stay open; they're just emptied. """ + await self._ensure_sub_managers() await asyncio.gather(self._inner.purge(), *(sm.purge() for sm in self._sub_managers.values())) + self._in_flight_from_inner.clear() for state in self._domain_states.values(): state.consecutive_429_count = 0 state.throttled_until = _NEVER_THROTTLED @@ -135,12 +152,13 @@ async def add_request(self, request: str | Request, *, forefront: bool = False) Requests for explicitly configured domains are routed directly to their per-domain sub-manager. All other requests go to the inner manager. """ + await self._ensure_sub_managers() + url = self._get_url_from_request(request) domain = self._extract_domain(url) if domain in self._domain_states: - sm = await self._get_or_create_sub_manager(domain) - result = await sm.add_request(request, forefront=forefront) + result = await self._sub_managers[domain].add_request(request, forefront=forefront) else: result = await self._inner.add_request(request, forefront=forefront) @@ -159,6 +177,8 @@ async def add_requests( wait_for_all_requests_to_be_added_timeout: timedelta | None = None, ) -> None: """Add multiple requests, routing each to the appropriate manager.""" + await self._ensure_sub_managers() + inner_requests: list[str | Request] = [] domain_requests: dict[str, list[str | Request]] = {} @@ -182,8 +202,7 @@ async def add_requests( ) for domain, reqs in domain_requests.items(): - sm = await self._get_or_create_sub_manager(domain) - await sm.add_requests( + await self._sub_managers[domain].add_requests( reqs, forefront=forefront, batch_size=batch_size, @@ -204,6 +223,8 @@ async def fetch_next_request(self) -> Request | None: manager is also empty and all sub-managers are throttled, waits until either the earliest domain becomes available or new work is added (whichever comes first). """ + await self._ensure_sub_managers() + while True: # Clear the event before checking the queues. Any add/reclaim that races with this iteration will set the # event again, so the wait at the end of the loop returns immediately rather than blocking until the @@ -212,11 +233,7 @@ async def fetch_next_request(self) -> Request | None: now = datetime.now(timezone.utc) available_domains = sorted( - ( - domain - for domain, state in self._domain_states.items() - if domain in self._sub_managers and now >= state.throttled_until - ), + (domain for domain, state in self._domain_states.items() if now >= state.throttled_until), key=lambda d: self._domain_states[d].throttled_until, ) @@ -228,11 +245,10 @@ async def fetch_next_request(self) -> Request | None: request = await self._inner.fetch_next_request() if request is not None: + if self._extract_domain(request.url) in self._domain_states: + self._in_flight_from_inner.add((request.unique_key, request.url)) return request - if not self._sub_managers: - return None - sub_managers_empty = await asyncio.gather(*(sm.is_empty() for sm in self._sub_managers.values())) if all(sub_managers_empty): return None @@ -250,20 +266,25 @@ async def fetch_next_request(self) -> Request | None: @override async def reclaim_request(self, request: Request, *, forefront: bool = False) -> ProcessedRequest | None: - manager = self._select_manager(request.url) + await self._ensure_sub_managers() + manager = self._fetch_owner(request) result = await manager.reclaim_request(request, forefront=forefront) + self._clear_fetch_owner(request) self._signal_new_work() return result @override async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None: - manager = self._select_manager(request.url) + await self._ensure_sub_managers() + manager = self._fetch_owner(request) result = await manager.mark_request_as_handled(request) + self._clear_fetch_owner(request) self.record_success(request.url) return result @override async def get_handled_count(self) -> int: + await self._ensure_sub_managers() counts = await asyncio.gather( self._inner.get_handled_count(), *(sm.get_handled_count() for sm in self._sub_managers.values()) ) @@ -271,6 +292,7 @@ async def get_handled_count(self) -> int: @override async def get_total_count(self) -> int: + await self._ensure_sub_managers() counts = await asyncio.gather( self._inner.get_total_count(), *(sm.get_total_count() for sm in self._sub_managers.values()) ) @@ -278,11 +300,13 @@ async def get_total_count(self) -> int: @override async def is_empty(self) -> bool: + await self._ensure_sub_managers() results = await asyncio.gather(self._inner.is_empty(), *(sm.is_empty() for sm in self._sub_managers.values())) 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()) ) @@ -368,15 +392,34 @@ def _get_domain_state(self, url: str) -> _DomainState | None: domain = self._extract_domain(url) return self._domain_states.get(domain) if domain else None - async def _get_or_create_sub_manager(self, domain: str) -> TRequestManager: - """Get or create a per-domain sub-manager using the configured `request_manager_opener`.""" - if domain not in self._sub_managers: - self._sub_managers[domain] = await self._request_manager_opener( - alias=f'throttled-{domain}', - storage_client=self._service_locator.get_storage_client(), - configuration=self._service_locator.get_configuration(), + async def _open_sub_manager(self, domain: str) -> None: + """Open the sub-manager for a single domain using the configured `request_manager_opener`.""" + self._sub_managers[domain] = await self._request_manager_opener( + alias=f'throttled-{domain}', + storage_client=self._service_locator.get_storage_client(), + configuration=self._service_locator.get_configuration(), + ) + + async def _ensure_sub_managers(self) -> None: + """Open a sub-manager for every configured domain, once; a retry opens only what is still missing.""" + if self._sub_managers_ready: + return + + async with self._sub_managers_lock: + if self._sub_managers_ready: + return + + # All attempts must settle before the lock is released: openers left running would write into + # `_sub_managers` after a retry has already replaced that domain's manager. + missing = [domain for domain in self._domain_states if domain not in self._sub_managers] + results = await asyncio.gather( + *(self._open_sub_manager(domain) for domain in missing), return_exceptions=True ) - return self._sub_managers[domain] + for result in results: + if isinstance(result, BaseException): + raise result + + self._sub_managers_ready = True def _is_domain_throttled(self, domain: str) -> bool: """Check if a domain is currently throttled.""" @@ -408,12 +451,18 @@ def _signal_new_work(self) -> None: """Wake `fetch_next_request` if it is sleeping inside a throttle wait.""" self._new_work_event.set() - def _select_manager(self, url: str) -> TRequestManager: - """Return the manager that owns the given URL — its sub-manager if one exists, otherwise the inner.""" - domain = self._extract_domain(url) - if domain in self._sub_managers: - return self._sub_managers[domain] - return self._inner + def _fetch_owner(self, request: Request) -> TRequestManager: + """Return the manager the request must be given back to, leaving its in-flight record in place. + + `_clear_fetch_owner` drops the record only once the completion is accepted, so a retry resolves the same way. + """ + if (request.unique_key, request.url) in self._in_flight_from_inner: + return self._inner + return self._sub_managers.get(self._extract_domain(request.url), self._inner) + + def _clear_fetch_owner(self, request: Request) -> None: + """Drop the in-flight record of a request whose completion was accepted.""" + self._in_flight_from_inner.discard((request.unique_key, request.url)) async def _wait_for_new_work_or_timeout(self, timeout: float) -> None: """Wait until new work is signaled or `timeout` seconds elapse, whichever comes first. diff --git a/tests/unit/test_throttling_request_manager.py b/tests/unit/test_throttling_request_manager.py index 0451297fff..162f5b3e6e 100644 --- a/tests/unit/test_throttling_request_manager.py +++ b/tests/unit/test_throttling_request_manager.py @@ -4,6 +4,7 @@ import asyncio from datetime import datetime, timedelta, timezone +from pathlib import Path from typing import Any from unittest.mock import AsyncMock, patch @@ -12,11 +13,13 @@ from crawlee._request import Request from crawlee._service_locator import ServiceLocator from crawlee._utils.http import parse_retry_after_header +from crawlee.configuration import Configuration from crawlee.request_loaders._throttling_request_manager import ThrottlingRequestManager -from crawlee.storage_clients import MemoryStorageClient +from crawlee.storage_clients import FileSystemStorageClient, MemoryStorageClient from crawlee.storages import RequestQueue THROTTLED_DOMAIN = 'throttled.com' +SECOND_THROTTLED_DOMAIN = 'slow.com' NON_THROTTLED_DOMAIN = 'free.com' TEST_DOMAINS = [THROTTLED_DOMAIN] @@ -50,11 +53,38 @@ async def manager(inner_queue: RequestQueue, service_locator: ServiceLocator) -> ) +@pytest.fixture +def fs_service_locator() -> ServiceLocator: + """Create a ServiceLocator backed by the file system, so storages survive a simulated restart.""" + return ServiceLocator(configuration=Configuration(purge_on_start=False), storage_client=FileSystemStorageClient()) + + def _make_request(url: str) -> Request: """Helper to create a Request object.""" return Request.from_url(url) +async def _open_fs_manager(service_locator: ServiceLocator) -> ThrottlingRequestManager[RequestQueue]: + """Open a throttling manager over the persistent storage directory, as a fresh process would.""" + inner_queue = await RequestQueue.open( + name='persistent-inner', + storage_client=service_locator.get_storage_client(), + configuration=service_locator.get_configuration(), + ) + return ThrottlingRequestManager( + inner_queue, + domains=TEST_DOMAINS, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + +async def _restart_fs_manager(service_locator: ServiceLocator) -> ThrottlingRequestManager[RequestQueue]: + """Simulate a process restart by dropping cached storage instances and reopening the manager.""" + service_locator.storage_instance_manager.clear_cache() + return await _open_fs_manager(service_locator) + + # ── Request Routing Tests ───────────────────────────────── @@ -437,6 +467,112 @@ async def test_mark_request_as_handled_routes_to_inner( assert await inner_queue.get_handled_count() == 1 +async def test_reclaim_returns_inner_request_to_inner( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """A request fetched from inner is reclaimed back into inner, even when its domain is configured.""" + # A domain configured only after the request had already been stored in inner. + await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1') + + request = await manager.fetch_next_request() + assert request is not None + + await manager.reclaim_request(request) + + assert not await inner_queue.is_empty() + assert await manager._sub_managers[THROTTLED_DOMAIN].is_empty() + + +async def test_handled_inner_request_finishes_inner( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """A request fetched from inner is marked as handled in inner, so inner can finish.""" + await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1') + + request = await manager.fetch_next_request() + assert request is not None + + await manager.mark_request_as_handled(request) + + assert await inner_queue.is_finished() is True + assert await manager.is_finished() is True + + +async def test_purge_forgets_in_flight_inner_requests( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """A purge drops the requests still in flight, so their routing must not outlive it.""" + url = f'https://{THROTTLED_DOMAIN}/page1' + await inner_queue.add_request(url) + assert await manager.fetch_next_request() is not None + + await manager.purge() + + # The same URL yields the same unique key, so a stale record would send it back to inner. + await manager.add_request(url) + request = await manager.fetch_next_request() + assert request is not None + await manager.mark_request_as_handled(request) + + assert await manager._sub_managers[THROTTLED_DOMAIN].is_finished() is True + + +async def test_failed_completion_keeps_its_inner_routing_for_the_retry( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A completion that fails keeps its inner routing, so a retried completion is not diverted to the sub-manager.""" + await inner_queue.add_request(f'https://{THROTTLED_DOMAIN}/page1') + request = await manager.fetch_next_request() + assert request is not None + + # `BasicCrawler` retries this, so a failed attempt must not consume the routing record. + monkeypatch.setattr( + inner_queue, + 'mark_request_as_handled', + AsyncMock(side_effect=RuntimeError('transient storage failure')), + ) + with pytest.raises(RuntimeError, match='transient storage failure'): + await manager.mark_request_as_handled(request) + + monkeypatch.undo() + await manager.mark_request_as_handled(request) + + assert await inner_queue.is_finished() is True + assert await manager.is_finished() is True + + +async def test_shared_unique_key_does_not_reroute_sub_manager_request( + manager: ThrottlingRequestManager[RequestQueue], + inner_queue: RequestQueue, +) -> None: + """A unique key shared with an in-flight inner request must not send a sub-manager request back to inner.""" + shared_key = 'shared-unique-key' + # A leftover from before the domain was configured, sharing its explicit unique key with a fresh request. + await inner_queue.add_request( + Request.from_url(f'https://{THROTTLED_DOMAIN}/page1', unique_key=shared_key), + ) + await manager.add_request(Request.from_url(f'https://{THROTTLED_DOMAIN}/page2', unique_key=shared_key)) + + # Sub-managers are drained before inner, so the first fetch is the sub-manager's request. + sub_request = await manager.fetch_next_request() + inner_request = await manager.fetch_next_request() + assert sub_request is not None + assert inner_request is not None + assert sub_request.url == f'https://{THROTTLED_DOMAIN}/page2' + assert inner_request.url == f'https://{THROTTLED_DOMAIN}/page1' + + 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 + + async def test_get_handled_count_aggregates(manager: ThrottlingRequestManager[RequestQueue]) -> None: """get_handled_count should sum inner and all sub-managers.""" throttled_url = f'https://{THROTTLED_DOMAIN}/page1' @@ -527,6 +663,172 @@ async def test_purge_clears_requests_and_resets_throttle_state( assert not manager._is_domain_throttled(THROTTLED_DOMAIN) +async def test_sub_managers_opened_for_every_configured_domain( + inner_queue: RequestQueue, + service_locator: ServiceLocator, +) -> None: + """Every configured domain gets a sub-manager on first use, not during construction.""" + domains = [THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN] + manager = ThrottlingRequestManager( + inner_queue, + domains=domains, + request_manager_opener=RequestQueue.open, + service_locator=service_locator, + ) + + assert manager._sub_managers == {} + + await manager.is_empty() + + assert set(manager._sub_managers) == set(domains) + + +async def test_sub_managers_opened_once( + inner_queue: RequestQueue, + service_locator: ServiceLocator, +) -> None: + """Concurrent first calls open each sub-manager exactly once.""" + domains = [THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN] + opener = AsyncMock(side_effect=RequestQueue.open) + manager: ThrottlingRequestManager[RequestQueue] = ThrottlingRequestManager( + inner_queue, + domains=domains, + request_manager_opener=opener, + service_locator=service_locator, + ) + + await asyncio.gather(manager.is_empty(), manager.is_finished(), manager.fetch_next_request()) + + assert opener.await_count == len(domains) + + +async def test_failed_open_keeps_successful_sub_managers( + inner_queue: RequestQueue, + service_locator: ServiceLocator, +) -> None: + """A failing opener leaves the sub-managers that did open in place, so a retry opens only what is missing.""" + domains = [THROTTLED_DOMAIN, SECOND_THROTTLED_DOMAIN] + failing_alias = f'throttled-{SECOND_THROTTLED_DOMAIN}' + pending_failures = {failing_alias} + + async def open_once_failing(*, alias: str, **kwargs: Any) -> RequestQueue: + if alias in pending_failures: + pending_failures.discard(alias) + raise RuntimeError('storage unavailable') + return await RequestQueue.open(alias=alias, **kwargs) + + opener = AsyncMock(side_effect=open_once_failing) + manager: ThrottlingRequestManager[RequestQueue] = ThrottlingRequestManager( + inner_queue, + domains=domains, + request_manager_opener=opener, + service_locator=service_locator, + ) + + with pytest.raises(RuntimeError, match='storage unavailable'): + await manager.is_empty() + + assert await manager.is_empty() is True + + # 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] + assert opened_aliases.count(f'throttled-{THROTTLED_DOMAIN}') == 1 + assert opened_aliases.count(failing_alias) == 2 + + +async def test_read_path_after_drop_reopens_sub_managers( + manager: ThrottlingRequestManager[RequestQueue], +) -> None: + """After drop(), the next read reopens the sub-managers.""" + await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1') + await manager.drop() + assert manager._sub_managers == {} + + assert await manager.is_empty() is True + assert set(manager._sub_managers) == set(TEST_DOMAINS) + + +async def test_read_paths_reopen_persisted_sub_queues(fs_service_locator: ServiceLocator) -> None: + """A restarted manager sees the requests a previous run left in a persisted sub-queue.""" + urls = [f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2'] + manager = await _open_fs_manager(fs_service_locator) + await manager.add_requests(urls) + + 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.get_total_count() == 2 + + request = await restarted.fetch_next_request() + assert request is not None + assert request.url in urls + + +async def test_default_purge_on_start_empties_persisted_sub_queues(fs_service_locator: ServiceLocator) -> None: + """With the default `purge_on_start`, opening the sub-queues empties what a previous run left in them.""" + manager = await _open_fs_manager(fs_service_locator) + await manager.add_requests([f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2']) + + # Restart under the default configuration, which purges an aliased store as it opens. + 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 + + # 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 + + +async def test_purge_empties_sub_queues_from_a_previous_run(fs_service_locator: ServiceLocator) -> None: + """purge() empties sub-queues left behind by a previous run.""" + manager = await _open_fs_manager(fs_service_locator) + await manager.add_requests([f'https://{THROTTLED_DOMAIN}/page1', f'https://{THROTTLED_DOMAIN}/page2']) + + restarted = await _restart_fs_manager(fs_service_locator) + await restarted.purge() + + sub_queue = await RequestQueue.open( + alias=f'throttled-{THROTTLED_DOMAIN}', + storage_client=fs_service_locator.get_storage_client(), + configuration=fs_service_locator.get_configuration(), + ) + assert await sub_queue.get_total_count() == 0 + + +async def test_drop_removes_sub_queues_from_a_previous_run(fs_service_locator: ServiceLocator) -> None: + """drop() removes the on-disk sub-queues left behind by a previous run.""" + manager = await _open_fs_manager(fs_service_locator) + await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1') + + sub_queue_path = ( + Path(fs_service_locator.get_configuration().storage_dir) / 'request_queues' / f'throttled-{THROTTLED_DOMAIN}' + ) + assert sub_queue_path.exists() + + restarted = await _restart_fs_manager(fs_service_locator) + await restarted.drop() + + assert not sub_queue_path.exists() + + +async def test_reclaim_routes_to_sub_manager_after_restart(fs_service_locator: ServiceLocator) -> None: + """After a restart, a reclaimed request goes back to its sub-manager rather than to inner.""" + manager = await _open_fs_manager(fs_service_locator) + await manager.add_request(f'https://{THROTTLED_DOMAIN}/page1') + + restarted = await _restart_fs_manager(fs_service_locator) + request = await restarted.fetch_next_request() + assert request is not None + + await restarted.reclaim_request(request) + + assert not await restarted._sub_managers[THROTTLED_DOMAIN].is_empty() + assert await restarted.inner.is_empty() + + # ── Utility Tests ──────────────────────────────────────