Adress review comments

Author: Pijukatel

src/apify/storage_clients/_apify/_request_queue_client.py

@@ -249,7 +249,9 @@ async def add_batch_of_requests(
 
         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.
+                # We are not sure if it was already handled at this point, and it is not worth calling API for it.
+                # It could have been handled by another client in the meantime, so cached information about
+                # `request.was_already_handled` is not reliable.
                 already_present_requests.append(
                     {
                         'id': request.id,
@@ -299,7 +301,7 @@ async def add_batch_of_requests(
 
         logger.debug(
             f'Tried to add new requests: {len(new_requests)}, '
-            f'succeeded to add new requests: {len(response["processedRequests"])}, '
+            f'succeeded to add new requests: {len(response["processedRequests"]) - len(already_present_requests)}, '
             f'skipped already present requests: {len(already_present_requests)}'
         )
 

tests/integration/test_actor_request_queue.py

@@ -163,18 +163,28 @@ 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."""
+    """Test that the deduplication works correctly even with parallel attempts to add same links.
+
+    The test is set up in a way for workers to have some requests that were already added to the queue and some new
+    requests. The function must correctly deduplicate the requests and add only new requests. For example:
+    First worker adding 10 new requests,
+    second worker adding 10 new requests and 10 known requests,
+    third worker adding 10 new requests and 20 known requests and so on"""
 
     async def main() -> None:
         import asyncio
         import logging
 
         from apify import Actor, Request
 
+        worker_count = 10
+        max_requests = 100
+        batch_size = iter(range(10, max_requests + 1, int(max_requests / worker_count)))
+
         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(100)]
+            requests = [Request.from_url(f'http://example.com/{i}') for i in range(max_requests)]
             rq = await Actor.open_request_queue()
 
             await asyncio.sleep(10)  # Wait to be sure that metadata are updated
@@ -186,11 +196,12 @@ async def main() -> None:
             stats_before = _rq.get('stats', {})
             Actor.log.info(stats_before)
 
-            # Add same requests in 10 parallel workers
+            # Add batches of some new and some already present requests in workers
             async def add_requests_worker() -> None:
-                await rq.add_requests(requests)
+                await rq.add_requests(requests[: next(batch_size)])
 
-            add_requests_workers = [asyncio.create_task(add_requests_worker()) for _ in range(10)]
+            # Start all workers
+            add_requests_workers = [asyncio.create_task(add_requests_worker()) for _ in range(worker_count)]
             await asyncio.gather(*add_requests_workers)
 
             await asyncio.sleep(10)  # Wait to be sure that metadata are updated