fix(autoscaling)!: Finish AutoscaledPool only when idle and scope stop() to the current run

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).

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.
 

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()