diff --git a/docs/upgrading/upgrading_to_v2.md b/docs/upgrading/upgrading_to_v2.md index 927116a7e5..4ba5fcfb47 100644 --- a/docs/upgrading/upgrading_to_v2.md +++ b/docs/upgrading/upgrading_to_v2.md @@ -115,6 +115,7 @@ The autoscaling classes now live in the public `crawlee.autoscaling` package, wh - `system_status` and `concurrency_settings` are replaced by `concurrency_system`, which has to be active when the pool runs, and `consumer`, which identifies the pool to the system. - `run_task_function` receives the id the task is booked under. - `current_concurrency` comes from the concurrency system, so a shared system may count the tasks of other pools too. The pool's own tasks are counted by the new `own_concurrency`. +- `is_finished_function` is called only when no new task can start and none of the pool's own tasks are in flight, and never while the pool is paused. In v1, it was called on every loop iteration, so a paused pool could finish. A pool whose `is_task_ready_function` keeps resolving to `True` doesn't finish, so make it resolve to `False` once the work runs out. For example: @@ -154,20 +155,6 @@ async with ConcurrencySystem(settings) as concurrency_system: The value has to be a whole number of at least 1, or infinity, which is the default. A fractional value, such as `0.5` or `7.5`, now raises `ValueError`. -## Autoscaling logs changed - -The periodic log of the autoscaling state and the `Waiting for remaining tasks to finish` message are now logged at DEBUG instead of INFO. The crawler's status message at INFO still reports the desired concurrency. - -The state log also has a new format. Instead of raw ratios for every resource, it names only the overloaded ones, with the share of time each spent over its threshold and its limit: - -```text -current concurrency 3, desired 5, overloaded: cpu 58% of the time (limit 40%) -``` - -To read the concurrency values in code, use `crawler.concurrency_system`. - -The autoscaling loggers moved together with the package. `crawlee._autoscaling.autoscaled_pool`, `crawlee._autoscaling.snapshotter` and `crawlee._autoscaling.system_status` are now `crawlee.autoscaling._autoscaled_pool`, `crawlee.autoscaling._snapshotter` and `crawlee.autoscaling._system_status`. The state log, along with the messages about scaling and free capacity, now comes from the new `crawlee.autoscaling._concurrency_system` logger. To configure all of them at once, use the `crawlee.autoscaling` logger. - ## Only the first crawler uses the default request queue Only the first crawler created in a process opens the default request queue, the same as Crawlee JS. Every later crawler without a `request_manager` opens its own queue under the alias `__default_{id}__`. A crawler without an `id` uses its creation index there, such as `__default_1__`. In v1, all these crawlers shared the default queue, so a URL one crawler handled was skipped by the others. @@ -208,10 +195,46 @@ for urls in batches: await crawler.run(urls) ``` -When a crawl processes no requests because its request manager holds only handled ones, the crawler now logs a warning that explains how to crawl them again. - `Configuration.purge_on_start` isn't affected. Unnamed storages are still purged once per process, when they're first opened. ## `RequestQueue.purge` raises when the storage client can't purge In v1, `RequestQueue.purge()` worked around a storage client that didn't support purging. It dropped an unnamed queue and opened the default queue in its place, and it skipped a named queue with a warning. In v2, it raises the client's `NotImplementedError`, the same as Crawlee JS. The Apify platform's storage client doesn't support purging, so on the platform, call `drop()` or open a new queue instead. + +## Logs changed + +If you configure loggers by name or match log messages, update them for the following changes. + +### Logger names + +The autoscaling loggers moved together with the package. `crawlee._autoscaling.autoscaled_pool`, `crawlee._autoscaling.snapshotter` and `crawlee._autoscaling.system_status` are now `crawlee.autoscaling._autoscaled_pool`, `crawlee.autoscaling._snapshotter` and `crawlee.autoscaling._system_status`. The state log, along with the messages about scaling and free capacity, now comes from the new `crawlee.autoscaling._concurrency_system` logger. To configure all of them at once, use the `crawlee.autoscaling` logger. + +### Log levels + +The periodic log of the autoscaling state and the `Waiting for remaining tasks to finish` message are now logged at DEBUG instead of INFO. The crawler's status message at INFO still reports the desired concurrency. To read the concurrency values in code, use `crawler.concurrency_system`. + +### Log messages + +In v1, the autoscaling state log listed raw ratios for every resource. In v2, it names only the overloaded resources, with the share of time each spent over its threshold and its limit: + +```text +current concurrency 3, desired 5, overloaded: cpu 58% of the time (limit 40%) +``` + +The INFO messages the crawler logs after `stop()` is called or `max_requests_per_crawl` is reached have new wording: + +```text +# v1 +No new requests are allowed because crawler `stop` method was called. Ongoing requests will be allowed to complete. +The crawler will finish any remaining ongoing requests and shut down. + +# v2 +No new requests are allowed because the `stop()` method has been called. Ongoing requests will be allowed to complete. +The crawler has finished all the remaining ongoing requests and will shut down now. +``` + +The crawler logs the first message once per run, the next time it looks for a new request after the stop. It logs the second one once the ongoing requests have finished, where v1 logged it right after the stop. In `Crawler.stop() was called with following reason: ...`, a trailing period or space in the reason is dropped, so the default reason and the `max_requests_per_crawl` one no longer end with a double period. + +When a crawl processes no requests because its request manager holds only handled ones, the crawler logs a new warning. It starts with `This crawl processed no requests` and explains how to crawl the requests again. + +`RequestQueue.purge()` no longer logs the `Storage client "..." does not support purging the request queue.` warnings. It raises `NotImplementedError` instead. For details, see [`RequestQueue.purge` raises when the storage client can't purge](#requestqueuepurge-raises-when-the-storage-client-cant-purge). diff --git a/src/crawlee/autoscaling/_autoscaled_pool.py b/src/crawlee/autoscaling/_autoscaled_pool.py index 0c94f0bc9f..7789a1c412 100644 --- a/src/crawlee/autoscaling/_autoscaled_pool.py +++ b/src/crawlee/autoscaling/_autoscaled_pool.py @@ -66,9 +66,9 @@ def __init__( function is called every time there is free capacity for a new task and it should indicate whether it should start a new task or not by resolving to either `True` or `False`. Besides its obvious use, it is also useful for task throttling to save resources. - is_finished_function: A function that is called only when there are no tasks to be processed. If it - resolves to `True` then the pool's run finishes. Being called only when there are no tasks being - processed means that as long as `is_task_ready_function` keeps resolving to `True`, + is_finished_function: A function that is called only when there are no tasks to be processed and the + pool is not paused. If it resolves to `True` then the pool's run finishes. Being called only when there + are no tasks being processed means that as long as `is_task_ready_function` keeps resolving to `True`, `is_finished_function` will never be called. To abort a run, use the `abort` method. """ self._concurrency_system = concurrency_system @@ -188,15 +188,21 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None: orchestrator_error: Exception | None = None try: - while not (finished := await self._is_finished_function()) and not run.result.done(): + while not run.result.done(): run.worker_tasks_updated.clear() if self._is_paused: logger.debug('Not scheduling new tasks - the autoscaled pool is paused') elif not self._concurrency_system.has_capacity_for_task(self._consumer): logger.debug('Not scheduling new tasks - no free capacity') + # Another pool can keep a shared system saturated indefinitely, so this pool still has to notice + # that it has run out of work. + if finished := await self._maybe_finish(run): + break elif not await self._is_task_ready_function(): logger.debug('Not scheduling new task - no task is ready') + if finished := await self._maybe_finish(run): + break else: task_id = str(self._next_task_id) self._next_task_id += 1 @@ -242,6 +248,13 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None: # A worker failure or an abort already decided the run, so this error has no way out. logger.error('Unpropagated exception in worker task orchestrator', exc_info=orchestrator_error) + async def _maybe_finish(self, run: _AutoscaledPoolRun) -> bool: + """Ask `is_finished_function` whether the run is over, but only while none of this pool's tasks is in flight. + + A running task can still produce new work, and with a shared system only this pool's own tasks count. + """ + return not run.worker_tasks and await self._is_finished_function() + def _reap_worker_task(self, task: asyncio.Task, *, run: _AutoscaledPoolRun, task_id: str) -> None: """Handle cleanup and tracking of a completed worker task. diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index b54e6aa988..c6474c0c34 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -519,7 +519,6 @@ async def persist_state_factory() -> KeyValueStore: # The default system is built per run, so that a run is not scaled on the load and concurrency of the previous # one. A supplied system is known up front. self._concurrency_system: ConcurrencySystemInterface | None = concurrency_system - self._autoscaled_pool: AutoscaledPool | None = None self._crawler_state_rec_task = RecurringTask( func=self._crawler_state_task, delay=status_message_logging_interval, @@ -534,6 +533,7 @@ async def persist_state_factory() -> KeyValueStore: self._failed = False self._unexpected_stop = False self._logger_once = LoggerOnce(self._logger) + self._run_logger_once = LoggerOnce(self._logger) @property def log(self) -> logging.Logger: @@ -577,7 +577,7 @@ def stop(self, reason: str = 'Stop was called externally.') -> None: Args: reason: Reason for stopping that will be used in logs. """ - self._logger.info(f'Crawler.stop() was called with following reason: {reason}.') + self._logger.info(f'Crawler.stop() was called with following reason: {reason.rstrip(" .")}.') self._unexpected_stop = True def _wrap_handler_with_error_context( @@ -600,7 +600,7 @@ async def wrapped_handler(context: TCrawlingContext | BasicCrawlingContext, exce def _stop_if_max_requests_count_exceeded(self) -> None: """Call `stop` when the maximum number of requests to crawl has been reached.""" - if self._max_requests_per_crawl is None: + if self._max_requests_per_crawl is None or self._unexpected_stop: return if self._statistics.state.requests_total >= self._max_requests_per_crawl: @@ -843,9 +843,12 @@ def sigint_handler() -> None: return final_statistics finally: # A failed run must leave the instance usable, so that the caller can retry after handling the error. - self._autoscaled_pool = None self._running = False + self._unexpected_stop = False + self._failed = False + self._run_logger_once = LoggerOnce(self._logger) + async def _run_crawler(self) -> None: local_event_manager = self._service_locator.get_event_manager() global_event_manager = service_locator.get_event_manager() @@ -870,7 +873,7 @@ async def _run_crawler(self) -> None: concurrency_system = owned_concurrency_system self._concurrency_system = concurrency_system - self._autoscaled_pool = autoscaled_pool = AutoscaledPool( + autoscaled_pool = AutoscaledPool( concurrency_system=concurrency_system, consumer=self._consumer, is_finished_function=self.__is_finished_function, @@ -1493,22 +1496,19 @@ async def _commit_key_value_store_changes( async def __is_finished_function(self) -> bool: self._stop_if_max_requests_count_exceeded() if self._unexpected_stop: - self._logger.info('The crawler will finish any remaining ongoing requests and shut down.') + self._logger.info('The crawler has finished all the remaining ongoing requests and will shut down now.') return True if self._abort_on_error and self._failed: - self._failed = False return True if self._keep_alive: return False - # Check for a stall only when the pool is idle. An in-flight request can still enqueue new work or end the - # stall. The pool's own requests are what counts, not those of another crawler sharing the budget. + # The pool asks only while none of this crawler's requests are in flight, so none of them can still enqueue + # new work or end the stall. if ( isinstance(self._request_manager, ThrottlingRequestManager) - and self._autoscaled_pool is not None - and self._autoscaled_pool.own_concurrency == 0 and (reason := await self._request_manager.get_stall_reason()) is not None ): error = PersistentRateLimitError(f'Giving up: {reason}') @@ -1523,12 +1523,17 @@ async def __is_finished_function(self) -> bool: async def __is_task_ready_function(self) -> bool: self._stop_if_max_requests_count_exceeded() if self._unexpected_stop: - self._logger.info( - 'No new requests are allowed because crawler `stop` method was called. ' - 'Ongoing requests will be allowed to complete.' + self._run_logger_once.log( + 'No new requests are allowed because the `stop()` method has been called. ' + 'Ongoing requests will be allowed to complete.', + key='shutting_down', ) return False + # The pool asks whether the run is finished only once nothing is in flight, so stop taking new requests here. + if self._abort_on_error and self._failed: + return False + request_manager = await self.get_request_manager() return not await request_manager.is_empty() diff --git a/tests/unit/autoscaling/test_autoscaled_pool.py b/tests/unit/autoscaling/test_autoscaled_pool.py index 071464a46a..5248aac5c6 100644 --- a/tests/unit/autoscaling/test_autoscaled_pool.py +++ b/tests/unit/autoscaling/test_autoscaled_pool.py @@ -83,7 +83,7 @@ async def test_retries_refused_booking() -> None: concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), + is_task_ready_function=lambda: future(run.await_count < 1), is_finished_function=lambda: future(run.await_count >= 1), ) @@ -112,7 +112,7 @@ async def run(_task_id: str) -> None: concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), + is_task_ready_function=lambda: future(not release.is_set()), is_finished_function=lambda: future(release.is_set()), ) @@ -127,6 +127,87 @@ async def run(_task_id: str) -> None: await asyncio.gather(pool_run_task, return_exceptions=True) +async def test_finishes_only_when_idle(make_concurrency_system: type[ConcurrencySystem]) -> None: + """The pool asks whether it's finished only while none of its tasks are in flight.""" + # The second task outlives the first, so the pool looks for new work while a task is still running. + durations = iter([0.01, 0.1]) + own_concurrency_at_checks = list[int]() + + async def sleep(_task_id: str) -> None: + await asyncio.sleep(next(durations)) + + run = AsyncMock(side_effect=sleep) + + async def is_finished() -> bool: + own_concurrency_at_checks.append(pool.own_concurrency) + return True + + async with make_concurrency_system() as concurrency_system: + pool = AutoscaledPool( + concurrency_system=concurrency_system, + consumer=CONSUMER, + run_task_function=run, + is_task_ready_function=AsyncMock(side_effect=chain([True, True], repeat(False))), + is_finished_function=is_finished, + ) + + await asyncio.wait_for(pool.run(), timeout=5) + + assert run.await_count == 2 + assert own_concurrency_at_checks == [0] + + +async def test_finishes_while_system_saturated(make_concurrency_system: type[ConcurrencySystem]) -> None: + """A pool out of work finishes even while another pool keeps a shared system saturated.""" + settings = ConcurrencySettings(min_concurrency=1, max_concurrency=1, desired_concurrency=1) + is_task_ready = AsyncMock(return_value=False) + + async with make_concurrency_system(settings) as concurrency_system: + # Another pool sharing the system holds the only slot. + assert concurrency_system.try_register_task_start(ConcurrencyConsumer(id='other-pool'), '0') + + pool = AutoscaledPool( + concurrency_system=concurrency_system, + consumer=CONSUMER, + run_task_function=AsyncMock(), + is_task_ready_function=is_task_ready, + is_finished_function=lambda: future(True), + ) + + await asyncio.wait_for(pool.run(), timeout=5) + + # The pool never had capacity to ask for a task, so it finished from the saturated branch. + is_task_ready.assert_not_awaited() + + +async def test_does_not_finish_while_paused(make_concurrency_system: type[ConcurrencySystem]) -> None: + """A paused pool with no work left finishes only once it's resumed.""" + is_finished = AsyncMock(return_value=True) + + async with make_concurrency_system() as concurrency_system: + pool = AutoscaledPool( + concurrency_system=concurrency_system, + consumer=CONSUMER, + run_task_function=AsyncMock(), + is_task_ready_function=lambda: future(False), + is_finished_function=is_finished, + ) + pool.pause() + + pool_run_task = asyncio.create_task(pool.run(), name='pool run task') + try: + await asyncio.sleep(0.1) + is_finished.assert_not_awaited() + + pool.resume() + await asyncio.wait_for(pool_run_task, timeout=5) + finally: + pool_run_task.cancel() + await asyncio.gather(pool_run_task, return_exceptions=True) + + is_finished.assert_awaited_once() + + @pytest.mark.run_alone async def test_runs_concurrently(make_concurrency_system: type[ConcurrencySystem]) -> None: """The pool fills the whole budget the system allows, rather than running the tasks one after another.""" @@ -149,7 +230,7 @@ async def run(_task_id: str) -> None: concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), + is_task_ready_function=lambda: future(done_count < 10), is_finished_function=lambda: future(done_count >= 10), ) @@ -273,7 +354,7 @@ async def run(_task_id: str) -> None: concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), + is_task_ready_function=lambda: future(done_count < 20), is_finished_function=lambda: future(done_count >= 20), ) @@ -333,22 +414,22 @@ async def run(_task_id: str) -> None: await release_worker.wait() worker_finished.set() - async def is_finished() -> bool: + async def is_task_ready() -> bool: nonlocal checks checks += 1 if checks > 1: await worker_started.wait() check_failed.set() raise RuntimeError('Queue status unavailable') - return False + return True async with make_concurrency_system() as concurrency_system: pool = AutoscaledPool( concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), - is_finished_function=is_finished, + is_task_ready_function=is_task_ready, + is_finished_function=lambda: future(False), ) pool_run_task = asyncio.create_task(pool.run()) try: @@ -391,30 +472,27 @@ async def is_finished() -> bool: async def test_propagates_exceptions_after_finished(make_concurrency_system: type[ConcurrencySystem]) -> None: - """A task that fails after the pool is already finished still fails the run.""" - started_count = 0 + """A task that fails once no work is left still fails the run.""" + is_finished = AsyncMock(return_value=True) async def run(_task_id: str) -> None: - nonlocal started_count - started_count += 1 - - await asyncio.sleep(1) - + await asyncio.sleep(0.1) raise RuntimeError('Scheduled crash') - async with make_concurrency_system( - ConcurrencySettings(min_concurrency=1, max_concurrency=1, desired_concurrency=1) - ) as concurrency_system: + async with make_concurrency_system() as concurrency_system: pool = AutoscaledPool( concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), - is_finished_function=lambda: future(started_count > 0), + is_task_ready_function=AsyncMock(side_effect=chain([True], repeat(False))), + is_finished_function=is_finished, ) with pytest.raises(RuntimeError, match=r'Scheduled crash'): - await pool.run() + await asyncio.wait_for(pool.run(), timeout=5) + + # The pool never asked whether it was finished while the failing task was still running. + is_finished.assert_not_awaited() async def test_propagates_task_timeout_error( @@ -435,7 +513,7 @@ async def run(_task_id: str) -> None: concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), + is_task_ready_function=lambda: future(started_count == 0), is_finished_function=lambda: future(started_count > 0), ) diff --git a/tests/unit/autoscaling/test_concurrency_system.py b/tests/unit/autoscaling/test_concurrency_system.py index 8edc4f1c20..59f008d358 100644 --- a/tests/unit/autoscaling/test_concurrency_system.py +++ b/tests/unit/autoscaling/test_concurrency_system.py @@ -345,7 +345,7 @@ async def run(_task_id: str) -> None: concurrency_system=concurrency_system, consumer=CONSUMER, run_task_function=run, - is_task_ready_function=lambda: future(True), + is_task_ready_function=lambda: future(done_count < 5), is_finished_function=lambda: future(done_count >= 5), ) diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index 5b96c9bca6..b9f8a2eb8c 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -26,7 +26,7 @@ from crawlee._request import Request, RequestState from crawlee._types import BasicCrawlingContext, EnqueueLinksKwargs, HttpMethod from crawlee._utils.robots import RobotsTxtFile -from crawlee.autoscaling import AutoscaledPool, ConcurrencyConsumer, ConcurrencySystem +from crawlee.autoscaling import ConcurrencyConsumer, ConcurrencySystem from crawlee.configuration import Configuration from crawlee.crawlers import BasicCrawler from crawlee.errors import ( @@ -72,15 +72,19 @@ async def handler(context: BasicCrawlingContext) -> None: @pytest.mark.parametrize( - 'method', - [pytest.param('is_empty', id='is_empty'), pytest.param('is_finished', id='is_finished')], + ('method', 'handled_before_error'), + [ + pytest.param('is_empty', [], id='is_empty'), + # The pool asks whether the run is finished only once the queue is drained. + pytest.param('is_finished', ['https://a.placeholder.com'], 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, method: str, handled_before_error: list[str], 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() @@ -99,8 +103,7 @@ async def handler(context: BasicCrawlingContext) -> None: await crawler.run() assert exc_info.value is error - assert handled_urls == [] - assert not await queue.is_finished() + assert handled_urls == handled_before_error await crawler.run() assert handled_urls == ['https://a.placeholder.com'] @@ -1429,6 +1432,23 @@ async def handler(context: BasicCrawlingContext) -> None: assert stats.requests_finished == expected_finished +async def test_abort_on_error_applies_to_one_run() -> None: + """An abort on error ends only the run it happened in, so the next run crawls again.""" + # Without retries, the failed request isn't left in the queue for the second run. + crawler = BasicCrawler(abort_on_error=True, max_request_retries=0) + handled_urls = list[str]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + handled_urls.append(context.request.url) + raise ValueError('Error request') + + await crawler.run(['https://a.placeholder.com']) + await crawler.run(['https://b.placeholder.com']) + + assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com'] + + def test_crawler_log() -> None: crawler = BasicCrawler() assert isinstance(crawler.log, logging.Logger) @@ -1830,6 +1850,54 @@ async def handler(context: BasicCrawlingContext) -> None: assert stats.requests_finished == 2 +async def test_stop_logged_once(caplog: pytest.LogCaptureFixture) -> None: + """The stop is logged once, while the requests in flight still finish one by one.""" + start_urls = [f'https://placeholder.com/{i}' for i in range(3)] + crawler = BasicCrawler( + concurrency_settings=ConcurrencySettings(min_concurrency=3, desired_concurrency=3, max_concurrency=3), + max_requests_per_crawl=1, + ) + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + # Finish one by one, so that the crawler is asked for new requests several times after the limit is reached. + await asyncio.sleep(0.05 * start_urls.index(context.request.url)) + + with caplog.at_level(logging.INFO, logger='crawlee'): + await crawler.run(start_urls) + + assert caplog.text.count('Crawler.stop() was called') == 1 + assert caplog.text.count('No new requests are allowed') == 1 + + +async def test_stop_applies_to_one_run(caplog: pytest.LogCaptureFixture) -> None: + """A stop ends only the run it was called in, so the next run crawls and logs its own stop.""" + crawler = BasicCrawler(concurrency_settings=ConcurrencySettings(desired_concurrency=1, max_concurrency=1)) + handled_urls = list[str]() + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + handled_urls.append(context.request.url) + crawler.stop() + + with caplog.at_level(logging.INFO, logger='crawlee'): + await crawler.run(['https://a.placeholder.com']) + await crawler.run(['https://b.placeholder.com']) + + assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com'] + assert caplog.text.count('No new requests are allowed') == 2 + + +async def test_stop_before_run_starts() -> None: + """A stop called right after scheduling `run` as a task applies to that run.""" + crawler = BasicCrawler(keep_alive=True, request_handler=AsyncMock()) + + run_task = asyncio.create_task(crawler.run()) + crawler.stop() + + await asyncio.wait_for(run_task, timeout=10) + + async def test_services_no_side_effect_on_crawler_init() -> None: custom_configuration = Configuration() custom_event_manager = LocalEventManager.from_config(custom_configuration) @@ -3064,12 +3132,12 @@ async def handler(context: BasicCrawlingContext) -> None: if context.request.url == 'https://free.placeholder.com/slow': stall_reasons.append(await poll_until_condition(throttler.get_stall_reason, 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) + checks = is_empty.await_count + await poll_until_condition(lambda: is_empty.await_count >= checks + 2, timeout=10) await context.add_requests(['https://free.placeholder.com/next']) with ( - patch.object(throttler, 'is_finished', wraps=throttler.is_finished) as is_finished, + patch.object(throttler, 'is_empty', wraps=throttler.is_empty) as is_empty, pytest.raises(PersistentRateLimitError), ): await crawler.run([THROTTLED_URL, 'https://free.placeholder.com/slow']) @@ -3311,18 +3379,3 @@ async def test_consumer_carries_crawler_id() -> None: consumers = {booking.args[0] for booking in try_register_task_start.call_args_list} assert consumers == {ConcurrencyConsumer(id='my-crawler')} - - -async def test_stall_ignores_shared_concurrency() -> None: - """A stall is judged by the crawler's own requests, not by those of another crawler sharing the system.""" - throttler = await open_throttler() - crawler = BasicCrawler(request_manager=throttler) - # Another crawler sharing the system has a request in flight, while this one has none. - pool = Mock(spec=AutoscaledPool, own_concurrency=0, current_concurrency=1) - - with ( - patch.object(crawler, '_autoscaled_pool', pool), - patch.object(throttler, 'get_stall_reason', AsyncMock(return_value='Domain stalled')), - pytest.raises(PersistentRateLimitError), - ): - await cast('Any', crawler)._BasicCrawler__is_finished_function()