fix(crawlers)!: Do not purge the request queue between run() calls

docs/guides/request_loaders.mdx

@@ -72,6 +72,7 @@ class RequestManager {
     + add_request()
     + add_requests_batched()
     + reclaim_request()
+    + purge()
     + drop()
 }
 

docs/guides/storages.mdx

@@ -206,15 +206,15 @@ Crawlee provides the following helper function to simplify interactions with the
 
 ## Cleaning up the storages
 
-By default, Crawlee cleans up all unnamed storages (including the default one) at the start of each run, so every crawl begins with a clean state. This behavior is controlled by <ApiLink to="class/Configuration#purge_on_start">`Configuration.purge_on_start`</ApiLink> (default: True). In contrast, named storages are never purged automatically and persist across runs. The exact behavior may vary depending on the storage client implementation.
+By default, Crawlee cleans up all unnamed storages (including the default one) once per process, when each one is first opened, so every process begins with a clean state. This behavior is controlled by <ApiLink to="class/Configuration#purge_on_start">`Configuration.purge_on_start`</ApiLink> (default: True). In contrast, named storages are never purged automatically and persist across runs. The exact behavior may vary depending on the storage client implementation.
 
 ### When purging happens
 
 The cleanup occurs as soon as a storage is accessed:
 
 - When opening a storage explicitly (e.g., <ApiLink to="class/RequestQueue#open">`RequestQueue.open`</ApiLink>, <ApiLink to="class/Dataset#open">`Dataset.open`</ApiLink>, <ApiLink to="class/KeyValueStore#open">`KeyValueStore.open`</ApiLink>).
 - When using helper functions that implicitly open storages (e.g., <ApiLink to="class/PushDataFunction">`push_data`</ApiLink>).
-- Automatically when <ApiLink to="class/BasicCrawler#run">`BasicCrawler.run`</ApiLink> is invoked.
+- Automatically when <ApiLink to="class/BasicCrawler#run">`BasicCrawler.run`</ApiLink> opens the crawler's storages.
 
 ### Disabling automatic purging
 

docs/upgrading/upgrading_to_v2.md

@@ -179,3 +179,39 @@ queue = await RequestQueue.open()
 crawler_1 = BasicCrawler(request_manager=queue)
 crawler_2 = BasicCrawler(request_manager=queue)
 ```
+
+## Repeated `run()` calls no longer purge the request queue
+
+In v1, calling `crawler.run()` again on the same instance purged the request queue, unless the queue was named or the previous run had failed. In v2, nothing is purged between runs, the same as Crawlee JS. A repeated `run()` continues with the same request manager and skips the requests an earlier run handled, failed ones included.
+
+The `purge_request_queue` argument of `run()` is removed. If you passed `purge_request_queue=False`, drop the argument. To crawl the same requests again, purge the request manager yourself:
+
+```python
+# v1
+await crawler.run(urls)
+await crawler.run(urls)
+
+# v2
+await crawler.run(urls)
+request_manager = await crawler.get_request_manager()
+await request_manager.purge()
+await crawler.run(urls)
+```
+
+`RequestManagerTandem.purge()` now purges the manager it wraps, where v1 did nothing. The loader isn't purged, so requests it has already passed to the manager don't come back.
+
+Instead of purging, you can create a new crawler for each crawl. Every crawler after the first opens a request queue of its own, as described in [Only the first crawler uses the default request queue](#only-the-first-crawler-uses-the-default-request-queue). Crawlers created with the same `id` share that queue, so leave `id` unset or give each crawler a different one:
+
+```python
+for urls in batches:
+    crawler = BasicCrawler()
+    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.

src/crawlee/crawlers/_basic/_basic_crawler.py

@@ -530,7 +530,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._handled_request_this_run = False
         self._failed = False
         self._unexpected_stop = False
         self._logger_once = LoggerOnce(self._logger)
@@ -717,20 +717,15 @@ def on_skipped_request(self, callback: SkippedRequestCallback) -> SkippedRequest
         self._on_skipped_request = callback
         return callback
 
-    async def run(
-        self,
-        requests: Sequence[str | Request] | None = None,
-        *,
-        purge_request_queue: bool = True,
-    ) -> FinalStatistics:
+    async def run(self, requests: Sequence[str | Request] | None = None) -> FinalStatistics:
         """Run the crawler until all requests are processed.
 
+        Calling `run()` again on the same instance keeps crawling the same request manager. Requests that a previous
+        run handled, failed ones included, aren't processed again. To crawl them again, purge the request manager, or
+        create a new crawler with a freshly opened request queue.
+
         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. A run that ended with an exception does not count as a previous run, so a
-                retry keeps the requests that were still pending even when this is `True`. Named request queues
-                are considered persistent and are never purged implicitly.
 
         Raises:
             PersistentRateLimitError: If a domain throttled by a `ThrottlingRequestManager` has rate-limited every
@@ -742,6 +737,7 @@ async def run(
             )
 
         self._running = True
+        self._handled_request_this_run = False
 
         try:
             if self._respect_robots_txt_file and not isinstance(self._request_manager, ThrottlingRequestManager):
@@ -758,22 +754,6 @@ async def run(
                 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()
-
             if requests is not None:
                 await self.add_requests(requests)
 
@@ -827,7 +807,6 @@ def sigint_handler() -> None:
                 )
 
             self._has_finished_before = True
-            self._last_run_failed = False
 
             await self._save_crawler_state()
 
@@ -836,10 +815,31 @@ def sigint_handler() -> None:
                 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:
+
+            # A crawl that did nothing while the manager holds only handled requests is a mistake, whoever handled them.
+            # Starting against handled requests is not, since that is what resuming a crawl looks like. The statistics
+            # can't tell, as they don't count skipped requests and may be restored from a previous process.
+            if not self._handled_request_this_run:
+                request_manager = await self.get_request_manager()
+                # The diagnostic must never fail the run.
+                try:
+                    is_finished = await request_manager.is_finished()
+                    handled_count = await request_manager.get_handled_count()
+                except Exception:
+                    is_finished = False
+                    handled_count = 0
+
+                # Pending requests mean the crawl ended early, such as after an abort, not that everything was handled.
+                if is_finished and handled_count > 0:
+                    noun = 'request' if handled_count == 1 else 'requests'
+                    message = (
+                        f'This crawl processed no requests - the request manager holds {handled_count} {noun}, all of '
+                        'them already handled, and a failed request counts as handled too. Nothing empties a queue '
+                        'between runs, so to crawl them again, purge it (`await queue.purge()`) or use a fresh one '
+                        "(e.g. `RequestQueue.open(alias='second-run')`) with a freshly created crawler instance."
+                    )
+                    self._logger_once.log(message, key=message, level=logging.WARNING)
+
             return final_statistics
         finally:
             # A failed run must leave the instance usable, so that the caller can retry after handling the error.
@@ -1953,3 +1953,4 @@ async def _mark_request_as_handled(self, request: Request) -> None:
             logger=self._logger,
             max_retries=3,
         )
+        self._handled_request_this_run = True

src/crawlee/request_loaders/_request_manager_tandem.py

@@ -109,3 +109,12 @@ async def mark_request_as_handled(self, request: Request) -> None:
     @override
     async def drop(self) -> None:
         await self._read_write_manager.drop()
+
+    @override
+    async def purge(self) -> None:
+        """Purge the wrapped request manager.
+
+        The loader can't be purged. Requests it already passed to the manager don't come back, and the ones it still
+        holds are fetched as usual.
+        """
+        await self._read_write_manager.purge()

src/crawlee/storages/_request_queue.py

@@ -151,28 +151,7 @@ async def drop(self) -> None:
 
     @override
     async def purge(self) -> None:
-        try:
-            await self._client.purge()
-        except NotImplementedError:
-            client_name = type(self._client).__name__
-            if self._name is not None:
-                logger.warning(
-                    f'Storage client "{client_name}" does not support purging the request queue. '
-                    f'Skipping purge for named queue "{self._name}" to avoid destroying persistent data; '
-                    'the queue contents are left intact.'
-                )
-                return
-            logger.warning(
-                f'Storage client "{client_name}" does not support purging the request queue. '
-                'Falling back to dropping and recreating the unnamed queue; the request queue ID may change.'
-            )
-            await self.drop()
-            # Override `purge_on_start` so the storage client does not try to purge the freshly recreated
-            # (and necessarily empty) queue and re-raise the same `NotImplementedError`.
-            recreate_config = service_locator.get_configuration().model_copy(update={'purge_on_start': False})
-            new_rq = await RequestQueue.open(configuration=recreate_config)
-            self._client = new_rq._client  # noqa: SLF001
-            self._id = new_rq._id  # noqa: SLF001
+        await self._client.purge()
 
     @override
     async def add_request(