Update to handle parallel requests with same links

Author: Pijukatel

src/apify/storage_clients/_apify/_request_queue_client.py

@@ -239,8 +239,10 @@ async def add_batch_of_requests(
             Response containing information about the added requests.
         """
         # Do not try to add previously added requests to avoid pointless expensive calls to API
+
         new_requests: list[Request] = []
         already_present_requests: list[dict[str, str | bool]] = []
+
         for request in requests:
             if self._requests_cache.get(request.id):
                 # We are no sure if it was already handled at this point, and it is not worth calling API for it.
@@ -254,12 +256,22 @@ async def add_batch_of_requests(
                 )
 
             else:
+                # Add new request to the cache.
+                processed_request = ProcessedRequest.model_validate(
+                    {
+                        'id': request.id,
+                        'uniqueKey': request.unique_key,
+                        'wasAlreadyPresent': True,
+                        'wasAlreadyHandled': request.was_already_handled,
+                    }
+                )
+                self._cache_request(
+                    unique_key_to_request_id(request.unique_key),
+                    processed_request,
+                    forefront=False,
+                )
                 new_requests.append(request)
 
-        logger.debug(
-            f'Adding new requests: {len(new_requests)}, '
-            f'skipping already present requests: {len(already_present_requests)}'
-        )
         if new_requests:
             # Prepare requests for API by converting to dictionaries.
             requests_dict = [
@@ -272,19 +284,16 @@ async def add_batch_of_requests(
 
             # Send requests to API.
             response = await self._api_client.batch_add_requests(requests=requests_dict, forefront=forefront)
-            # Add new requests to the cache.
-            for processed_request_raw in response['processedRequests']:
-                processed_request = ProcessedRequest.model_validate(processed_request_raw)
-                self._cache_request(
-                    unique_key_to_request_id(processed_request.unique_key),
-                    processed_request,
-                    forefront=False,
-                )
             # Add the locally known already present processed requests based on the local cache.
             response['processedRequests'].extend(already_present_requests)
         else:
             response = {'unprocessedRequests': [], 'processedRequests': already_present_requests}
 
+        logger.debug(
+            f'Added new requests: {len(new_requests)}, '
+            f'skipped already present requests: {len(already_present_requests)}'
+        )
+
         # Update assumed total count for newly added requests.
         api_response = AddRequestsResponse.model_validate(response)
         new_request_count = 0

tests/integration/test_actor_request_queue.py

@@ -156,3 +156,51 @@ async def main() -> None:
     run_result = await run_actor(actor)
 
     assert run_result.status == 'SUCCEEDED'
+
+
+async def test_request_queue_parallel_deduplication(
+    make_actor: MakeActorFunction,
+    run_actor: RunActorFunction,
+) -> None:
+    """Test that the deduplication works correctly even with parallel attempts to add same links."""
+
+    async def main() -> None:
+        import asyncio
+        import logging
+
+        from apify import Actor, Request
+
+        async with Actor:
+            logging.getLogger('apify.storage_clients._apify._request_queue_client').setLevel(logging.DEBUG)
+
+            requests = [Request.from_url(f'http://example.com/{i}') for i in range(1000)]
+            rq = await Actor.open_request_queue()
+
+            await asyncio.sleep(10)  # Wait to be sure that metadata are updated
+
+            # Get raw client, because stats are not exposed in `RequestQueue` class, but are available in raw client
+            rq_client = Actor.apify_client.request_queue(request_queue_id=rq.id)
+            _rq = await rq_client.get()
+            assert _rq
+            stats_before = _rq.get('stats', {})
+            Actor.log.info(stats_before)
+
+            # Add same requests in 10 parallel workers
+            async def add_requests_worker() -> None:
+                await rq.add_requests(requests)
+
+            add_requests_workers = [asyncio.create_task(add_requests_worker()) for _ in range(10)]
+            await asyncio.gather(*add_requests_workers)
+
+            await asyncio.sleep(10)  # Wait to be sure that metadata are updated
+            _rq = await rq_client.get()
+            assert _rq
+            stats_after = _rq.get('stats', {})
+            Actor.log.info(stats_after)
+
+            assert (stats_after['writeCount'] - stats_before['writeCount']) == len(requests)
+
+    actor = await make_actor(label='rq-parallel-deduplication', main_func=main)
+    run_result = await run_actor(actor)
+
+    assert run_result.status == 'SUCCEEDED'