fix!: Keep the crawler usable after a failed run and retry the browser temp directory removal

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.

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
@@ -76,10 +85,31 @@ 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():
+    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.
+        """
+        async with self._temp_dir_lock:
             temp_dir = self._temp_dir
-            await asyncio.to_thread(shutil.rmtree, temp_dir, ignore_errors=True)
+
+            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
+
+            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,8 +118,7 @@ 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)
+        await self._delete_temp_dir()
 
     @property
     @override

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)
@@ -696,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(
@@ -706,86 +707,97 @@ 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)
-
-        interrupted = False
+            if self._has_finished_before:
+                await self._statistics.reset()
+
+                if self._use_session_pool:
+                    await self._session_pool.reset_store()
+
+                # 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
+                    # 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()
 
-        def sigint_handler() -> None:
-            nonlocal interrupted
+            if requests is not None:
+                await self.add_requests(requests)
 
-            if not interrupted:
-                interrupted = True
-                self._logger.info('Pausing... Press CTRL+C again to force exit.')
+            interrupted = False
 
-            run_task.cancel()
+            def sigint_handler() -> None:
+                nonlocal interrupted
 
-        run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task')
+                if not interrupted:
+                    interrupted = True
+                    self._logger.info('Pausing... Press CTRL+C again to force exit.')
 
-        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)
+                run_task.cancel()
 
-        try:
-            await run_task
-        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
+            run_task = asyncio.create_task(self._run_crawler(), name='run_crawler_task')
 
+            # `add_signal_handler` works only in the main thread
             if threading.current_thread() is threading.main_thread():
-                with suppress(NotImplementedError):
-                    asyncio.get_running_loop().remove_signal_handler(signal.SIGINT)
+                with suppress(NotImplementedError):  # event loop signal handlers are not supported on Windows
+                    asyncio.get_running_loop().add_signal_handler(signal.SIGINT, sigint_handler)
 
-        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}'
-            )
+            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)
+
+            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._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()

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
 

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,51 @@ 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."""
+    monkeypatch.setattr(PlaywrightPersistentBrowser, '_TMP_DIR_DELETE_INTERVAL', timedelta(0))
+
+    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()
+    # 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()
+
+
+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()
+
+    assert 'Could not remove the temporary user data directory' in caplog.text