diff --git a/src/apify/storage_clients/_apify/_request_queue_shared_client.py b/src/apify/storage_clients/_apify/_request_queue_shared_client.py index c9ad6235..486bba65 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/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) diff --git a/src/apify/storage_clients/_apify/_request_queue_single_client.py b/src/apify/storage_clients/_apify/_request_queue_single_client.py index 5d4860de..dfe4632c 100644 --- a/src/apify/storage_clients/_apify/_request_queue_single_client.py +++ b/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) diff --git a/tests/unit/storage_clients/test_apify_request_queue_client.py b/tests/unit/storage_clients/test_apify_request_queue_client.py index 2e1bd7a5..fcd632dd 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/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', [