From accd89370d643a3c378fbcb9265757f39f8458e5 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 14:29:06 +0200 Subject: [PATCH 1/3] fix(crawlers): Keep the crawler usable and its pending requests after a failed run --- src/crawlee/_autoscaling/autoscaled_pool.py | 4 +- src/crawlee/crawlers/_basic/_basic_crawler.py | 102 ++++++++++-------- .../crawlers/_basic/test_basic_crawler.py | 41 +++++++ 3 files changed, 100 insertions(+), 47 deletions(-) diff --git a/src/crawlee/_autoscaling/autoscaled_pool.py b/src/crawlee/_autoscaling/autoscaled_pool.py index 8e4d54bc0c..fb751289de 100644 --- a/src/crawlee/_autoscaling/autoscaled_pool.py +++ b/src/crawlee/_autoscaling/autoscaled_pool.py @@ -252,7 +252,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None: finally: if finished: logger.debug('`is_finished_function` reports that we are finished') - elif run.result.done() and run.result.exception() is not None: + elif run.result.done() and not run.result.cancelled() and run.result.exception() is not None: logger.debug('Unhandled exception in `run_task_function`') if run.worker_tasks: @@ -269,7 +269,7 @@ async def _worker_task_orchestrator(self, run: _AutoscaledPoolRun) -> None: 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) + logger.error('Unpropagated 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. diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index 39e928e2f9..bdc2cb81e2 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -507,6 +507,7 @@ async def persist_state_factory() -> KeyValueStore: self._keep_alive = keep_alive self._running = False self._has_finished_before = False + self._last_run_failed = False self._failed = False self._unexpected_stop = False self._logger_once = LoggerOnce(self._logger) @@ -706,64 +707,74 @@ async def run( self._running = True - if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager): - self._logger.warning( - 'The `respect_robots_txt_file` option is enabled, but the crawler is not using ' - '`ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To enable ' - 'crawl-delay support, configure the crawler to use `ThrottlingRequestManager` as the request manager.' - ) - - if self._has_finished_before: - await self._statistics.reset() - - if self._use_session_pool: - await self._session_pool.reset_store() - - if purge_request_queue: - request_manager = await self.get_request_manager() - # A `ThrottlingRequestManager` delegates `purge` to the manager it wraps, so inspect the wrapped - # manager when deciding whether the purge would hit a named queue. - inner_manager = ( - request_manager.inner if isinstance(request_manager, ThrottlingRequestManager) else request_manager + try: + if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager): + self._logger.warning( + 'The `respect_robots_txt_file` option is enabled, but the crawler is not using ' + '`ThrottlingRequestManager`. Crawl-delay directives from robots.txt will not be enforced. To ' + 'enable crawl-delay support, configure the crawler to use `ThrottlingRequestManager` as the ' + 'request manager.' ) - # Named storages are persistent and shared across runs, so they are never purged implicitly - # (the same named-storage exemption as in `StorageClient._purge_if_needed`). - is_named_queue = isinstance(inner_manager, RequestQueue) and inner_manager.name is not None - if not is_named_queue: - await request_manager.purge() - if requests is not None: - await self.add_requests(requests) + if self._has_finished_before: + await self._statistics.reset() + + if self._use_session_pool: + await self._session_pool.reset_store() + + # A run that ended with an exception does not count as a previous run, so the requests it left + # pending survive into the retry. + if purge_request_queue and not self._last_run_failed: + request_manager = await self.get_request_manager() + # A `ThrottlingRequestManager` delegates `purge` to the manager it wraps, so inspect the wrapped + # manager when deciding whether the purge would hit a named queue. + inner_manager = ( + request_manager.inner + if isinstance(request_manager, ThrottlingRequestManager) + else request_manager + ) + # Named storages are persistent and shared across runs, so they are never purged implicitly + # (the same named-storage exemption as in `StorageClient._purge_if_needed`). + is_named_queue = isinstance(inner_manager, RequestQueue) and inner_manager.name is not None + if not is_named_queue: + await request_manager.purge() + + if requests is not None: + await self.add_requests(requests) - interrupted = False + interrupted = False - def sigint_handler() -> None: - nonlocal interrupted + def sigint_handler() -> None: + nonlocal interrupted - if not interrupted: - interrupted = True - self._logger.info('Pausing... Press CTRL+C again to force exit.') + if not interrupted: + interrupted = True + self._logger.info('Pausing... Press CTRL+C again to force exit.') - run_task.cancel() + run_task.cancel() - run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task') + run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task') - if threading.current_thread() is threading.main_thread(): # `add_signal_handler` works only in the main thread - with suppress(NotImplementedError): # event loop signal handlers are not supported on Windows - asyncio.get_running_loop().add_signal_handler(signal.SIGINT, sigint_handler) + # `add_signal_handler` works only in the main thread + if threading.current_thread() is threading.main_thread(): + with suppress(NotImplementedError): # event loop signal handlers are not supported on Windows + asyncio.get_running_loop().add_signal_handler(signal.SIGINT, sigint_handler) - try: - await run_task - except CancelledError: - pass + try: + await run_task + except CancelledError: + pass + finally: + if threading.current_thread() is threading.main_thread(): + with suppress(NotImplementedError): + asyncio.get_running_loop().remove_signal_handler(signal.SIGINT) + except BaseException: + self._last_run_failed = True + raise 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) - if self._statistics.error_tracker.total > 0: self._logger.info( 'Error analysis:' @@ -777,6 +788,7 @@ def sigint_handler() -> None: ) self._has_finished_before = True + self._last_run_failed = False await self._save_crawler_state() diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index ce2c2783fb..cc4db930f1 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -97,6 +97,47 @@ async def handler(context: BasicCrawlingContext) -> None: assert await queue.is_finished() +async def test_crawler_is_usable_after_a_failed_run_setup(monkeypatch: pytest.MonkeyPatch) -> None: + """A failure before the crawl starts leaves the instance usable, so the caller can retry after handling it.""" + queue = await RequestQueue.open() + crawler = BasicCrawler(request_manager=queue) + handled_urls = [] + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + handled_urls.append(context.request.url) + + with monkeypatch.context() as monkey: + monkey.setattr(queue, 'add_requests', AsyncMock(side_effect=RuntimeError('Queue unavailable'))) + with pytest.raises(RuntimeError, match='Queue unavailable'): + await crawler.run(['https://a.placeholder.com']) + + await crawler.run(['https://a.placeholder.com']) + assert handled_urls == ['https://a.placeholder.com'] + + +async def test_failed_run_keeps_pending_requests_for_the_retry(monkeypatch: pytest.MonkeyPatch) -> None: + """Requests left pending by a failed run survive into the retry even when an earlier run completed.""" + queue = await RequestQueue.open() + crawler = BasicCrawler(request_manager=queue) + handled_urls = [] + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + handled_urls.append(context.request.url) + + await crawler.run(['https://a.placeholder.com']) + assert handled_urls == ['https://a.placeholder.com'] + + with monkeypatch.context() as monkey: + monkey.setattr(queue, 'is_empty', AsyncMock(side_effect=RuntimeError('Queue status unavailable'))) + with pytest.raises(RuntimeError, match='Queue status unavailable'): + await crawler.run(['https://b.placeholder.com']) + + await crawler.run() + assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com'] + + async def test_processes_requests_from_request_source_tandem() -> None: request_queue = await RequestQueue.open() await request_queue.add_requests( From 7bf7a83a1e6e3510b5af7a5fc17638e1f12c207b Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 16:00:18 +0200 Subject: [PATCH 2/3] fix(crawlers): Mark a run failed when its post-crawl steps raise --- src/crawlee/crawlers/_basic/_basic_crawler.py | 56 +++++++++---------- .../crawlers/_basic/test_basic_crawler.py | 50 +++++++++++++++++ 2 files changed, 78 insertions(+), 28 deletions(-) diff --git a/src/crawlee/crawlers/_basic/_basic_crawler.py b/src/crawlee/crawlers/_basic/_basic_crawler.py index bdc2cb81e2..e79039f2be 100644 --- a/src/crawlee/crawlers/_basic/_basic_crawler.py +++ b/src/crawlee/crawlers/_basic/_basic_crawler.py @@ -697,8 +697,8 @@ async def run( 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. 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. + retry keeps the requests that were still pending even when this is `True`. Named request queues + are considered persistent and are never purged implicitly. """ if self._running: raise RuntimeError( @@ -722,8 +722,7 @@ async def run( if self._use_session_pool: await self._session_pool.reset_store() - # A run that ended with an exception does not count as a previous run, so the requests it left - # pending survive into the retry. + # A failed run does not count as a previous run, so its pending requests survive into the retry. if purge_request_queue and not self._last_run_failed: request_manager = await self.get_request_manager() # A `ThrottlingRequestManager` delegates `purge` to the manager it wraps, so inspect the wrapped @@ -768,36 +767,37 @@ def sigint_handler() -> None: if threading.current_thread() is threading.main_thread(): with suppress(NotImplementedError): asyncio.get_running_loop().remove_signal_handler(signal.SIGINT) - except BaseException: - self._last_run_failed = True - raise - finally: - # A failed run must leave the instance usable, so that the caller can retry after handling the error. - self._running = False - if self._statistics.error_tracker.total > 0: - self._logger.info( - 'Error analysis:' - f' total_errors={self._statistics.error_tracker.total}' - f' unique_errors={self._statistics.error_tracker.unique_error_count}' - ) + if self._statistics.error_tracker.total > 0: + self._logger.info( + 'Error analysis:' + f' total_errors={self._statistics.error_tracker.total}' + f' unique_errors={self._statistics.error_tracker.unique_error_count}' + ) - if interrupted: - self._logger.info( - f'The crawl was interrupted. To resume, do: CRAWLEE_PURGE_ON_START=0 python {sys.argv[0]}' - ) + if interrupted: + self._logger.info( + f'The crawl was interrupted. To resume, do: CRAWLEE_PURGE_ON_START=0 python {sys.argv[0]}' + ) - self._has_finished_before = True - self._last_run_failed = False + self._has_finished_before = True + self._last_run_failed = False - await self._save_crawler_state() + await self._save_crawler_state() - final_statistics = self._statistics.calculate() - if self._statistics_log_format == 'table': - self._logger.info(f'Final request statistics:\n{final_statistics.to_table()}') + final_statistics = self._statistics.calculate() + if self._statistics_log_format == 'table': + self._logger.info(f'Final request statistics:\n{final_statistics.to_table()}') + else: + self._logger.info('Final request statistics:', extra=final_statistics.to_dict()) + except BaseException: + self._last_run_failed = True + raise else: - self._logger.info('Final request statistics:', extra=final_statistics.to_dict()) - return final_statistics + return final_statistics + finally: + # A failed run must leave the instance usable, so that the caller can retry after handling the error. + self._running = False async def _run_crawler(self) -> None: local_event_manager = self._service_locator.get_event_manager() diff --git a/tests/unit/crawlers/_basic/test_basic_crawler.py b/tests/unit/crawlers/_basic/test_basic_crawler.py index cc4db930f1..16b1e9d1e9 100644 --- a/tests/unit/crawlers/_basic/test_basic_crawler.py +++ b/tests/unit/crawlers/_basic/test_basic_crawler.py @@ -138,6 +138,56 @@ async def handler(context: BasicCrawlingContext) -> None: assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com'] +async def test_purge_resumes_once_a_run_succeeds_again(monkeypatch: pytest.MonkeyPatch) -> None: + """The purge exemption lasts only until a run succeeds, so the run after the retry starts from a clean queue.""" + queue = await RequestQueue.open() + crawler = BasicCrawler(request_manager=queue) + handled_urls = [] + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + handled_urls.append(context.request.url) + + await crawler.run(['https://a.placeholder.com']) + + with monkeypatch.context() as monkey: + monkey.setattr(queue, 'is_empty', AsyncMock(side_effect=RuntimeError('Queue status unavailable'))) + with pytest.raises(RuntimeError, match='Queue status unavailable'): + await crawler.run(['https://b.placeholder.com']) + + await crawler.run() + await crawler.run(['https://a.placeholder.com']) + + assert handled_urls == [ + 'https://a.placeholder.com', + 'https://b.placeholder.com', + 'https://a.placeholder.com', + ] + + +async def test_failure_after_the_crawl_marks_the_run_as_failed(monkeypatch: pytest.MonkeyPatch) -> None: + """A run whose post-crawl state save raises counts as failed too, so the retry keeps the queue intact.""" + queue = await RequestQueue.open() + crawler = BasicCrawler(request_manager=queue) + handled_urls = [] + + @crawler.router.default_handler + async def handler(context: BasicCrawlingContext) -> None: + handled_urls.append(context.request.url) + + await crawler.run(['https://a.placeholder.com']) + assert handled_urls == ['https://a.placeholder.com'] + + with monkeypatch.context() as monkey: + monkey.setattr(crawler, '_save_crawler_state', AsyncMock(side_effect=RuntimeError('Key-value store down'))) + with pytest.raises(RuntimeError, match='Key-value store down'): + await crawler.run() + + await queue.add_request('https://b.placeholder.com') + await crawler.run() + assert handled_urls == ['https://a.placeholder.com', 'https://b.placeholder.com'] + + async def test_processes_requests_from_request_source_tandem() -> None: request_queue = await RequestQueue.open() await request_queue.add_requests( From f3c83e5493ba8b87d23c9d39d96a1a839b8bd66e Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 16:00:24 +0200 Subject: [PATCH 3/3] test(autoscaling): Cover the orchestrator error logged on a cancelled run --- .../unit/_autoscaling/test_autoscaled_pool.py | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/unit/_autoscaling/test_autoscaled_pool.py b/tests/unit/_autoscaling/test_autoscaled_pool.py index 6f9763f345..65a9394285 100644 --- a/tests/unit/_autoscaling/test_autoscaled_pool.py +++ b/tests/unit/_autoscaling/test_autoscaled_pool.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import logging from contextlib import suppress from datetime import timedelta from itertools import chain, repeat @@ -164,6 +165,31 @@ async def is_finished() -> bool: await asyncio.gather(pool_run_task, return_exceptions=True) +async def test_orchestrator_error_is_logged_when_the_run_is_cancelled( + system_status: SystemStatus | Mock, caplog: pytest.LogCaptureFixture +) -> None: + """A scheduling error raised while the run is being cancelled is logged, as the cancelled result cannot carry it.""" + + async def is_finished() -> bool: + pool_run_task.cancel() + raise RuntimeError('Queue status unavailable') + + pool = AutoscaledPool( + system_status=system_status, + run_task_function=lambda: future(None), + is_task_ready_function=lambda: future(False), + is_finished_function=is_finished, + ) + + with caplog.at_level(logging.ERROR, logger='crawlee._autoscaling.autoscaled_pool'): + pool_run_task = asyncio.create_task(pool.run()) + with pytest.raises(asyncio.CancelledError): + await pool_run_task + + assert 'Unpropagated exception in worker task orchestrator' in caplog.text + assert 'Queue status unavailable' in caplog.text + + async def test_propagates_exceptions_after_finished(system_status: SystemStatus | Mock) -> None: started_count = 0