fix: propagate autoscaled pool scheduling errors (#2229)

### Description

When a request manager's `is_empty()` or `is_finished()` raises,
`BasicCrawler.run()` currently logs the orchestrator error but returns
normally with final statistics. A failed queue status check can
therefore look like a successfully completed crawl while requests remain
pending.

Retain scheduling exceptions until the existing worker cleanup finishes,
then fail the pool's result future instead of resolving it successfully.
Preserve a result already set by a worker failure or abort. The
regression tests exercise the crawler with a real filesystem queue,
inject errors only at its status methods, and verify that a new crawler
can process the pending request after recovery. A separate
event-controlled test checks that an active worker finishes before the
scheduling error reaches the caller.

### Issues

Found during source review; no existing issue linked. This concerns
scheduling callbacks, distinct from the worker-task timeout handling in
#2009.

### Testing

- Five new regression cases fail on unmodified
`0853d79e670cba27f9692919688305d1cc565c6c` because no exception reaches
the caller.
- Python 3.10.20: all five regression cases pass.
- Python 3.13.14: `pytest tests/unit/_autoscaling
tests/unit/crawlers/_basic tests/unit/_utils/test_recurring_task.py -k
'not test_send_request_works'` passes 157 tests, with 1 skipped and 2
deselected.
- Full Ruff lint and formatting checks pass (599 files); full `ty check
--python-platform linux` and changed-file native Windows typing pass.
- Native Windows full typing reports two `BaseContext.Process`
diagnostics in unchanged `test_system.py`; the same diagnostics occur on
the unmodified base. The two existing `test_send_request_works` cases
fail with JSON decoding errors against the local test server on both the
patch and the unmodified base; they are excluded from the final
related-suite run. The full repository/browser/service matrix was not
run locally.

---------

Co-authored-by: Vlada Dusek <v.dusek96@gmail.com>

Author: Kuang-xianxin

src/crawlee/_autoscaling/autoscaled_pool.py

@@ -40,8 +40,8 @@ class AutoscaledPool:
     """Manages a pool of asynchronous resource-intensive tasks that are executed in parallel.
 
     The pool keeps `min_concurrency` tasks running even while the system is overloaded, and starts additional tasks
-    only if there is enough free CPU and memory available. If an exception is thrown in any of the tasks, it is
-    propagated and the pool is stopped.
+    only if there is enough free CPU and memory available. If an exception is thrown in any of the tasks or in the
+    pool's scheduling loop, it is propagated and the pool is stopped.
     """
 
     _AUTOSCALE_INTERVAL = timedelta(seconds=10)
@@ -106,7 +106,7 @@ def __init__(
     async def run(self) -> None:
         """Start the autoscaled pool and return when all tasks are completed and `is_finished_function` returns True.
 
-        If there is an exception in one of the tasks, it will be re-raised.
+        If a task or the pool's scheduling loop raises an exception, it will be re-raised.
         """
         if self._current_run is not None:
             raise RuntimeError('The pool is already running')
@@ -217,6 +217,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
         Exits when `is_finished_function` returns True.
         """
         finished = False
+        orchestrator_error: Exception | None = None
 
         try:
             while not (finished := await self._is_finished_function()) and not run.result.done():
@@ -244,6 +245,10 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
 
                 with suppress(asyncio.TimeoutError):
                     await asyncio.wait_for(run.worker_tasks_updated.wait(), timeout=0.5)
+        except Exception as exc:
+            # Surface the error through `run.result` only once the cleanup below has awaited the worker tasks,
+            # so that the caller does not observe the failure while tasks are still in flight.
+            orchestrator_error = exc
         finally:
             if finished:
                 logger.debug('`is_finished_function` reports that we are finished')
@@ -258,7 +263,13 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None:
                 logger.debug('Terminating - no running tasks to wait for')
 
             if not run.result.done():
-                run.result.set_result(object())
+                if orchestrator_error is not None:
+                    run.result.set_exception(orchestrator_error)
+                else:
+                    run.result.set_result(object())
+            elif orchestrator_error is not None:
+                # A worker failure or an abort already decided the run, so this error has no way out.
+                logger.error('Exception in worker task orchestrator', exc_info=orchestrator_error)
 
     def _reap_worker_task(self, task: asyncio.Task, run: _AutoscaledPoolRun) -> None:
         """Handle cleanup and tracking of a completed worker task.

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -695,8 +695,9 @@ async def run(
         Args:
             requests: The requests to be enqueued before the crawler starts.
             purge_request_queue: If this is `True` and the crawler is not being run for the first time, the request
-                queue will be purged. Named request queues are considered persistent and are never purged
-                implicitly.
+                queue will be purged. A run that ended with an exception does not count as a previous run, so a
+                retry keeps the requests that were still pending. Named request queues are considered persistent
+                and are never purged implicitly.
         """
         if self._running:
             raise RuntimeError(
@@ -756,6 +757,9 @@ def sigint_handler() -> None:
         except CancelledError:
             pass
         finally:
+            # A failed run must leave the instance usable, so that the caller can retry after handling the error.
+            self._running = False
+
             if threading.current_thread() is threading.main_thread():
                 with suppress(NotImplementedError):
                     asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
@@ -772,7 +776,6 @@ def sigint_handler() -> None:
                 f'The crawl was interrupted. To resume, do: CRAWLEE_PURGE_ON_START=0 python {sys.argv[0]}'
             )
 
-        self._running = False
         self._has_finished_before = True
 
         await self._save_crawler_state()

tests/unit/_autoscaling/test_autoscaled_pool.py

@@ -123,6 +123,47 @@ async def run() -> None:
     assert done_count < 20
 
 
+async def test_orchestrator_error_waits_for_running_worker(system_status: SystemStatus | Mock) -> None:
+    """A scheduling error reaches the caller only after the worker tasks that were still running have finished."""
+    worker_started = asyncio.Event()
+    worker_finished = asyncio.Event()
+    check_failed = asyncio.Event()
+    release_worker = asyncio.Event()
+    checks = 0
+
+    async def run() -> None:
+        worker_started.set()
+        await release_worker.wait()
+        worker_finished.set()
+
+    async def is_finished() -> bool:
+        nonlocal checks
+        checks += 1
+        if checks > 1:
+            await worker_started.wait()
+            check_failed.set()
+            raise RuntimeError('Queue status unavailable')
+        return False
+
+    pool = AutoscaledPool(
+        system_status=system_status,
+        run_task_function=run,
+        is_task_ready_function=lambda: future(True),
+        is_finished_function=is_finished,
+    )
+    pool_run_task = asyncio.create_task(pool.run())
+    try:
+        await asyncio.wait_for(check_failed.wait(), timeout=5)
+        assert not pool_run_task.done()
+        release_worker.set()
+        with pytest.raises(RuntimeError, match='Queue status unavailable'):
+            await asyncio.wait_for(pool_run_task, timeout=5)
+        assert worker_finished.is_set()
+    finally:
+        release_worker.set()
+        await asyncio.gather(pool_run_task, return_exceptions=True)
+
+
 async def test_propagates_exceptions_after_finished(system_status: SystemStatus | Mock) -> None:
     started_count = 0
 

tests/unit/crawlers/_basic/test_basic_crawler.py

@@ -61,6 +61,42 @@ 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(asyncio.TimeoutError, id='timeout error')],
+)
+async def test_propagates_request_queue_status_errors(
+    monkeypatch: pytest.MonkeyPatch, method: 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()
+    await queue.add_request('https://a.placeholder.com')
+    crawler = BasicCrawler(request_manager=queue)
+    handled_urls = []
+
+    @crawler.router.default_handler
+    async def handler(context: BasicCrawlingContext) -> None:
+        handled_urls.append(context.request.url)
+
+    error = error_type('Queue status unavailable')
+    with monkeypatch.context() as monkey:
+        monkey.setattr(queue, method, 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()
+
+    await crawler.run()
+    assert handled_urls == ['https://a.placeholder.com']
+    assert await queue.is_finished()
+
+
 async def test_processes_requests_from_request_source_tandem() -> None:
     request_queue = await RequestQueue.open()
     await request_queue.add_requests(