fix: avoid crawling handled requests again after migration in large request queues

src/apify/storage_clients/_apify/_request_queue_shared_client.py

@@ -183,17 +183,20 @@ async def add_batch_of_requests(
                 # the call succeeds (not before) keeps a failed call from poisoning the cache and silently
                 # deduplicating a later retry of the same request.
                 unprocessed_unique_keys = {request.unique_key for request in api_response.unprocessed_requests}
+                processed_by_unique_key = {request.unique_key: request for request in api_response.processed_requests}
                 for request in new_requests:
                     if request.unique_key in unprocessed_unique_keys:
                         continue
                     request_id = unique_key_to_request_id(request.unique_key)
+                    processed = processed_by_unique_key.get(request.unique_key)
+                    was_already_handled_on_platform = processed is not None and processed.was_already_handled
                     self._cache_request(
                         request_id,
                         ProcessedRequest(
                             id=request_id,
                             unique_key=request.unique_key,
                             was_already_present=True,
-                            was_already_handled=request.was_already_handled,
+                            was_already_handled=request.was_already_handled or was_already_handled_on_platform,
                         ),
                     )
                     committed_request_ids.add(request_id)

src/apify/storage_clients/_apify/_request_queue_single_client.py

@@ -190,16 +190,23 @@ async def add_batch_of_requests(
                 # call succeeds (not before) keeps a failed call from poisoning the cache and silently
                 # deduplicating a later retry of the same request.
                 unprocessed_unique_keys = {request.unique_key for request in api_response.unprocessed_requests}
+                processed_by_unique_key = {request.unique_key: request for request in api_response.processed_requests}
                 for request in new_requests:
                     if request.unique_key in unprocessed_unique_keys:
                         continue
                     request_id = unique_key_to_request_id(request.unique_key)
+                    committed_request_ids.add(request_id)
+                    # A request handled on the platform (e.g. by an earlier client of this RQ) stays out of the
+                    # cache and the head, so it is never fetched again.
+                    processed = processed_by_unique_key.get(request.unique_key)
+                    if processed is not None and processed.was_already_handled:
+                        self._requests_already_handled.add(request_id)
+                        continue
                     self._requests_cache[request_id] = request
                     if forefront:
                         self._head_requests.append(request_id)
                     else:
                         self._head_requests.appendleft(request_id)
-                    committed_request_ids.add(request_id)
 
                 # Add the locally known already present processed requests based on the local cache.
                 api_response.processed_requests.extend(already_present_requests)

tests/unit/storage_clients/test_apify_request_queue_client.py

@@ -53,8 +53,9 @@ def _batch_result(
     *,
     processed: Sequence[Request] = (),
     unprocessed: Sequence[Request] = (),
+    already_handled: Sequence[Request] = (),
 ) -> BatchAddResult:
-    """Build a `batch_add_requests` response with explicit processed / unprocessed splits."""
+    """Build a `batch_add_requests` response with explicit processed / unprocessed / already handled splits."""
     return BatchAddResult.model_construct(
         processed_requests=[
             AddedRequest.model_construct(
@@ -64,6 +65,15 @@ def _batch_result(
                 was_already_handled=False,
             )
             for request in processed
+        ]
+        + [
+            AddedRequest.model_construct(
+                request_id=unique_key_to_request_id(request.unique_key),
+                unique_key=request.unique_key,
+                was_already_present=True,
+                was_already_handled=True,
+            )
+            for request in already_handled
         ],
         unprocessed_requests=[
             UnprocessedRequest.model_construct(
@@ -382,6 +392,42 @@ async def test_partial_unprocessed_commits_only_accepted_requests(access: str) -
     assert [request.unique_key for request in resent] == [rejected.unique_key]
 
 
+async def test_single_skips_request_batch_add_reported_already_handled() -> None:
+    """A request the platform reports as already handled is not cached, and `fetch_next_request()` returns `None`."""
+    client, api_client = _make_single_client()
+    api_client.list_requests = AsyncMock(return_value=SimpleNamespace(items=[]))
+    api_client.list_head = AsyncMock(
+        return_value=RequestQueueHead(
+            limit=1, queue_modified_at=datetime.now(tz=UTC), had_multiple_clients=False, items=[]
+        )
+    )
+    request = Request.from_url('https://example.com/handled')
+    request_id = unique_key_to_request_id(request.unique_key)
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result(already_handled=[request]))
+
+    await client.add_batch_of_requests([request])
+
+    assert request_id not in client._requests_cache
+    assert request_id in client._requests_already_handled
+    assert await client.fetch_next_request() is None
+
+
+async def test_shared_batch_add_reported_already_handled_needs_no_confirmation() -> None:
+    """A request the platform reports as already handled is cached as handled; `is_finished()` does not re-fetch it."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/handled')
+    request_id = unique_key_to_request_id(request.unique_key)
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result(already_handled=[request]))
+
+    await client.add_batch_of_requests([request])
+
+    assert request_id not in client._unhandled_request_ids
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head())
+    api_client.get_request = AsyncMock()
+    assert await client.is_finished() is True
+    api_client.get_request.assert_not_awaited()
+
+
 @pytest.mark.parametrize(
     'platform_request_visible',
     [