From baaa606d0e971531ffc320e259ffa62d9b249c02 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 14:00:23 +0200 Subject: [PATCH 1/5] fix(browsers): Retry the temp directory removal until the browser releases its files --- src/crawlee/browsers/_playwright_browser.py | 34 +++++++++++-- .../unit/browsers/test_playwright_browser.py | 51 ++++++++++++++++++- 2 files changed, 80 insertions(+), 5 deletions(-) diff --git a/src/crawlee/browsers/_playwright_browser.py b/src/crawlee/browsers/_playwright_browser.py index 8ce19bfd26..daf6fbbd96 100644 --- a/src/crawlee/browsers/_playwright_browser.py +++ b/src/crawlee/browsers/_playwright_browser.py @@ -3,6 +3,7 @@ import asyncio import shutil import tempfile +from datetime import timedelta from logging import getLogger from pathlib import Path from typing import TYPE_CHECKING, Any @@ -29,6 +30,12 @@ class PlaywrightPersistentBrowser(Browser): _TMP_DIR_PREFIX = 'apify-playwright-firefox-taac-' + _TMP_DIR_DELETE_ATTEMPTS = 50 + """The number of attempts to remove the temporary user data directory.""" + + _TMP_DIR_DELETE_INTERVAL = timedelta(milliseconds=100) + """The delay between the attempts to remove the temporary user data directory.""" + def __init__( self, browser_type: BrowserType, @@ -39,6 +46,8 @@ def __init__( self._browser_launch_options = browser_launch_options self._user_data_dir = user_data_dir self._temp_dir: Path | None = None + # Both `close` and the context's `close` event trigger the removal, so serialize the two. + self._temp_dir_lock = asyncio.Lock() self._context: BrowserContext | None = None self._is_connected = True @@ -77,9 +86,27 @@ async def new_context(self, **context_options: Any) -> BrowserContext: return self._context async def _delete_temp_dir(self, _: BrowserContext | None) -> None: - if self._temp_dir and self._temp_dir.exists(): - temp_dir = self._temp_dir - await asyncio.to_thread(shutil.rmtree, temp_dir, ignore_errors=True) + """Remove the temporary user data directory, retrying until the browser releases its files. + + The browser process can keep files in the directory open for a while after the context is closed, which makes + the removal fail on Windows. + """ + temp_dir = self._temp_dir + + if not temp_dir: + return + + async with self._temp_dir_lock: + for attempt in range(self._TMP_DIR_DELETE_ATTEMPTS): + if attempt: + await asyncio.sleep(self._TMP_DIR_DELETE_INTERVAL.total_seconds()) + + await asyncio.to_thread(shutil.rmtree, temp_dir, ignore_errors=True) + + if not temp_dir.exists(): + return + + logger.warning(f'Could not remove the temporary user data directory "{temp_dir}".') @override async def close(self, **kwargs: Any) -> None: @@ -88,7 +115,6 @@ async def close(self, **kwargs: Any) -> None: await self._context.close() self._context = None self._is_connected = False - await asyncio.sleep(0.1) await self._delete_temp_dir(self._context) @property diff --git a/tests/unit/browsers/test_playwright_browser.py b/tests/unit/browsers/test_playwright_browser.py index 120b886c59..f73567957d 100644 --- a/tests/unit/browsers/test_playwright_browser.py +++ b/tests/unit/browsers/test_playwright_browser.py @@ -1,7 +1,11 @@ from __future__ import annotations +import logging +import shutil +from datetime import timedelta from pathlib import Path -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any +from unittest.mock import Mock import pytest from playwright.async_api import async_playwright @@ -42,3 +46,48 @@ async def test_delete_temp_folder_with_close_browser(playwright: Playwright) -> assert current_temp_dir.exists() await persist_browser.close() assert not current_temp_dir.exists() + + +async def test_delete_temp_folder_when_files_are_locked( + playwright: Playwright, monkeypatch: pytest.MonkeyPatch +) -> None: + """The temp directory is removed even when the first delete attempts fail, as Windows locks the browser files.""" + real_rmtree = shutil.rmtree + locked_attempts = 3 + rmtree = Mock() + + def rmtree_locked_at_first(path: Any, **kwargs: Any) -> None: + """Model `rmtree(ignore_errors=True)` silently leaving the directory in place while a file is locked.""" + if rmtree.call_count > locked_attempts: + real_rmtree(path, **kwargs) + + rmtree.side_effect = rmtree_locked_at_first + monkeypatch.setattr(shutil, 'rmtree', rmtree) + + persist_browser = PlaywrightPersistentBrowser( + playwright.chromium, user_data_dir=None, browser_launch_options={'headless': True} + ) + await persist_browser.new_context() + assert isinstance(persist_browser._temp_dir, Path) + current_temp_dir = persist_browser._temp_dir + assert current_temp_dir.exists() + await persist_browser.close() + assert rmtree.call_count > locked_attempts + assert not current_temp_dir.exists() + + +async def test_warn_when_temp_folder_cannot_be_deleted( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """A temp directory that stays locked for the whole retry budget is reported with a warning.""" + monkeypatch.setattr(PlaywrightPersistentBrowser, '_TMP_DIR_DELETE_ATTEMPTS', 2) + monkeypatch.setattr(PlaywrightPersistentBrowser, '_TMP_DIR_DELETE_INTERVAL', timedelta(0)) + monkeypatch.setattr(shutil, 'rmtree', Mock()) + + persist_browser = PlaywrightPersistentBrowser(Mock(), user_data_dir=None, browser_launch_options={}) + persist_browser._temp_dir = tmp_path + + with caplog.at_level(logging.WARNING, logger='crawlee.browsers._playwright_browser'): + await persist_browser._delete_temp_dir(None) + + assert 'Could not remove the temporary user data directory' in caplog.text From 30f992d81b39535d7e50c8581feef18cf974c83a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 14:29:06 +0200 Subject: [PATCH 2/5] 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 1f4a215d6de54f34e60ffc607e51ac0427a28408 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 16:00:13 +0200 Subject: [PATCH 3/5] fix(browsers): Spend the temp directory retry budget once per close --- src/crawlee/browsers/_playwright_browser.py | 15 +++++++++------ tests/unit/browsers/test_playwright_browser.py | 7 +++++-- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/src/crawlee/browsers/_playwright_browser.py b/src/crawlee/browsers/_playwright_browser.py index daf6fbbd96..74377ec91f 100644 --- a/src/crawlee/browsers/_playwright_browser.py +++ b/src/crawlee/browsers/_playwright_browser.py @@ -85,18 +85,21 @@ async def new_context(self, **context_options: Any) -> BrowserContext: return self._context - async def _delete_temp_dir(self, _: BrowserContext | None) -> None: + async def _delete_temp_dir(self, _: BrowserContext | None = None) -> None: """Remove the temporary user data directory, retrying until the browser releases its files. The browser process can keep files in the directory open for a while after the context is closed, which makes the removal fail on Windows. """ - temp_dir = self._temp_dir + async with self._temp_dir_lock: + temp_dir = self._temp_dir - if not temp_dir: - return + if not temp_dir: + return + + # One close asks for the removal twice, so the second caller finds nothing left to do. + self._temp_dir = None - async with self._temp_dir_lock: for attempt in range(self._TMP_DIR_DELETE_ATTEMPTS): if attempt: await asyncio.sleep(self._TMP_DIR_DELETE_INTERVAL.total_seconds()) @@ -115,7 +118,7 @@ async def close(self, **kwargs: Any) -> None: await self._context.close() self._context = None self._is_connected = False - await self._delete_temp_dir(self._context) + await self._delete_temp_dir() @property @override diff --git a/tests/unit/browsers/test_playwright_browser.py b/tests/unit/browsers/test_playwright_browser.py index f73567957d..c84534106b 100644 --- a/tests/unit/browsers/test_playwright_browser.py +++ b/tests/unit/browsers/test_playwright_browser.py @@ -52,6 +52,8 @@ async def test_delete_temp_folder_when_files_are_locked( playwright: Playwright, monkeypatch: pytest.MonkeyPatch ) -> None: """The temp directory is removed even when the first delete attempts fail, as Windows locks the browser files.""" + monkeypatch.setattr(PlaywrightPersistentBrowser, '_TMP_DIR_DELETE_INTERVAL', timedelta(0)) + real_rmtree = shutil.rmtree locked_attempts = 3 rmtree = Mock() @@ -72,7 +74,8 @@ def rmtree_locked_at_first(path: Any, **kwargs: Any) -> None: current_temp_dir = persist_browser._temp_dir assert current_temp_dir.exists() await persist_browser.close() - assert rmtree.call_count > locked_attempts + # The context's `close` event and `close` itself both ask for the removal, but only one of them retries. + assert rmtree.call_count == locked_attempts + 1 assert not current_temp_dir.exists() @@ -88,6 +91,6 @@ async def test_warn_when_temp_folder_cannot_be_deleted( persist_browser._temp_dir = tmp_path with caplog.at_level(logging.WARNING, logger='crawlee.browsers._playwright_browser'): - await persist_browser._delete_temp_dir(None) + await persist_browser._delete_temp_dir() assert 'Could not remove the temporary user data directory' in caplog.text From c3191e87eecf899c6ca2a3da2106ca3e29d23d7d Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 16:00:18 +0200 Subject: [PATCH 4/5] 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 ef02218ae2d8bcfc4b9b7436f56ccdf61c04f5fa Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Wed, 16 Sep 2026 16:00:24 +0200 Subject: [PATCH 5/5] 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