feat(request-loaders)!: Raise in SitemapRequestLoader.mark_request_as_handled for a request it isn't processing

docs/06_upgrading/upgrading_to_v2.md

@@ -454,6 +454,10 @@ class MyHttpClient(HttpClient):
 
 `RequestManagerTandem.get_total_count` no longer counts a request twice once it moves from the loader to the manager. In v1, it added up the totals of both, so the crawler's status message showed too many pages in total.
 
+## `SitemapRequestLoader.mark_request_as_handled` raises for a request it isn't processing
+
+`SitemapRequestLoader.mark_request_as_handled` now raises a `ValueError` for a request that the loader isn't processing, such as one it never returned or one that's already handled. Crawlee JS raises in the same case. In v1, such a call did nothing.
+
 ## Resetting statistics and the session pool
 
 `Statistics.reset()` is now synchronous and resets only the in-memory statistics, the same as Crawlee JS. In v1, it also cleared the persisted record. To clear the record too, call the new `reset_store()` once the statistics are exited:

src/crawlee/request_loaders/_sitemap_request_loader.py

@@ -62,8 +62,9 @@ class SitemapRequestLoaderState(BaseModel):
     `processed_sitemap_urls`. If `pending_sitemap_urls` is empty, `completed` is set to `True`.
 
     When `fetch_next_request` is called, a URL is extracted from `url_queue`, and the request built from it is added
-    to `request_data` and its URL to `in_progress`. When `mark_request_as_handled` is called for the request, it is
-    removed from both and `handled_count` is incremented by 1.
+    to `request_data` and its URL to `in_progress`. A URL that's already in `in_progress`, because another sitemap
+    lists it too, is skipped and `total_count` is decremented by 1. When `mark_request_as_handled` is called for the
+    request, it is removed from both and `handled_count` is incremented by 1.
 
     During initial startup or restart after persistence, state validation occurs in `_get_state`. If both
     `pending_sitemap_urls` and `in_progress_sitemap_url` are empty and `completed` is False, this indicates a
@@ -261,7 +262,8 @@ async def fetch_next_request(self) -> Request | None:
                     self._queue_has_capacity.set()
 
                 # A request restored as in progress is served again as it was.
-                if (restored_request := state.request_data.get(url)) is not None:
+                restored_request = state.request_data.get(url)
+                if restored_request is not None and url not in state.in_progress:
                     state.in_progress.add(url)
                     return restored_request
 
@@ -274,19 +276,29 @@ async def fetch_next_request(self) -> Request | None:
                         request_option = transform_request_option
 
                 request = Request.from_url(**request_option)
+                # The URL was listed again while its request is still being processed.
+                if request.url in state.in_progress:
+                    state.total_count -= 1
+                    continue
                 state.in_progress.add(request.url)
                 state.request_data[request.url] = request
 
             return request
 
     @override
     async def mark_request_as_handled(self, request: Request) -> ProcessedRequest | None:
-        """Mark a request as successfully handled."""
+        """Mark a request as successfully handled.
+
+        Raises:
+            ValueError: If the request isn't being processed.
+        """
         state = await self._get_state()
-        if request.url in state.in_progress:
-            state.in_progress.remove(request.url)
-            del state.request_data[request.url]
-            state.handled_count += 1
+        if request.url not in state.in_progress:
+            raise ValueError(f'The request is not being processed (url: {request.url}).')
+
+        state.in_progress.remove(request.url)
+        del state.request_data[request.url]
+        state.handled_count += 1
         return None
 
     async def start(self) -> None:

tests/unit/request_loaders/test_sitemap_request_loader.py

@@ -217,6 +217,62 @@ def to_api_url(request_options: RequestOptions) -> RequestOptions:
     assert restored_request.user_data['page'] == 2
 
 
+async def test_mark_handled_twice_raises() -> None:
+    """Marking a request that's already handled raises and leaves the handled count as is."""
+    sitemap_url = f'{DEFAULT_URL}sitemap.xml'
+    client, _ = make_status_stream_client({sitemap_url: [(200, get_basic_sitemap().encode())]})
+    loader = SitemapRequestLoader([sitemap_url], http_client=client)
+    request = await loader.fetch_next_request()
+    assert request is not None
+    await loader.mark_request_as_handled(request)
+
+    with pytest.raises(ValueError, match='not being processed'):
+        await loader.mark_request_as_handled(request)
+
+    assert await loader.get_handled_count() == 1
+
+
+async def test_duplicate_url_skipped_while_in_progress() -> None:
+    """A URL listed again by another sitemap isn't served while its request is still in progress."""
+    sitemap_body = f'<urlset><url><loc>{DEFAULT_URL}page</loc></url></urlset>'.encode()
+    sitemap_urls = [f'{DEFAULT_URL}sitemap-1.xml', f'{DEFAULT_URL}sitemap-2.xml']
+    client, _ = make_status_stream_client({url: [(200, sitemap_body)] for url in sitemap_urls})
+    loader = SitemapRequestLoader(sitemap_urls, http_client=client)
+
+    request = await loader.fetch_next_request()
+    assert request is not None
+    duplicate = await loader.fetch_next_request()
+    await loader.mark_request_as_handled(request)
+
+    assert duplicate is None
+    assert await loader.get_total_count() == 1
+    assert await loader.get_handled_count() == 1
+
+
+async def test_restored_request_served_once_for_duplicate_url() -> None:
+    """A request restored as in progress isn't served again for the same URL listed by another sitemap."""
+    sitemap_body = f'<urlset><url><loc>{DEFAULT_URL}page</loc></url></urlset>'.encode()
+    sitemap_urls = [f'{DEFAULT_URL}sitemap-1.xml', f'{DEFAULT_URL}sitemap-2.xml']
+    client, _ = make_status_stream_client({url: [(200, sitemap_body)] for url in sitemap_urls})
+    loader = SitemapRequestLoader(sitemap_urls, http_client=client, persist_state_key='sitemap-state')
+    request = await loader.fetch_next_request()
+    assert request is not None
+    # The second sitemap queues the URL again while its request is in progress.
+    assert await poll_until_condition(loader.get_pending_count, lambda count: count == 2) == 2
+    await loader.close()
+
+    restored_loader = SitemapRequestLoader(sitemap_urls, http_client=client, persist_state_key='sitemap-state')
+    restored_request = await restored_loader.fetch_next_request()
+    duplicate = await restored_loader.fetch_next_request()
+    total_count = await restored_loader.get_total_count()
+    await restored_loader.close()
+
+    assert restored_request is not None
+    assert restored_request.url == request.url
+    assert duplicate is None
+    assert total_count == 1
+
+
 async def test_close_before_first_use_stops_listening(monkeypatch: pytest.MonkeyPatch) -> None:
     """A loader closed before its first use doesn't persist on PERSIST_STATE events once it's used."""
     sitemap_url = f'{DEFAULT_URL}sitemap.xml'