From ff9e045407508a14fa1d5e03e6800392ff8654eb Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 16 Oct 2026 12:22:27 +0200 Subject: [PATCH 1/3] fix: skip requests the batch add reports as already handled in single RQ client --- .../_apify/_request_queue_shared_client.py | 5 ++- .../_apify/_request_queue_single_client.py | 9 +++++- .../test_apify_request_queue_client.py | 32 +++++++++++++++++++ 3 files changed, 44 insertions(+), 2 deletions(-) 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..4750b758 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -382,6 +382,38 @@ 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_reports_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=BatchAddResult.model_construct( + processed_requests=[ + AddedRequest.model_construct( + request_id=request_id, + unique_key=request.unique_key, + was_already_present=True, + was_already_handled=True, + ) + ], + unprocessed_requests=[], + ) + ) + + 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 + + @pytest.mark.parametrize( 'platform_request_visible', [ From cb75fee11dfbbb2305c16829df04143a3de91858 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 16 Oct 2026 12:33:33 +0200 Subject: [PATCH 2/3] test: cover shared RQ client skipping requests reported as already handled --- .../test_apify_request_queue_client.py | 42 ++++++++++++------- 1 file changed, 28 insertions(+), 14 deletions(-) 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 4750b758..8606bca5 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( @@ -393,19 +403,7 @@ async def test_single_skips_request_batch_add_reports_already_handled() -> None: ) 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=BatchAddResult.model_construct( - processed_requests=[ - AddedRequest.model_construct( - request_id=request_id, - unique_key=request.unique_key, - was_already_present=True, - was_already_handled=True, - ) - ], - unprocessed_requests=[], - ) - ) + api_client.batch_add_requests = AsyncMock(return_value=_batch_result(already_handled=[request])) await client.add_batch_of_requests([request]) @@ -414,6 +412,22 @@ async def test_single_skips_request_batch_add_reports_already_handled() -> None: 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', [ From 7585abdb6c3d2e90e5d813108a7914583a31ee11 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 16 Oct 2026 12:35:03 +0200 Subject: [PATCH 3/3] test: align single RQ client already-handled test name with its shared twin --- tests/unit/storage_clients/test_apify_request_queue_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 8606bca5..fcd632dd 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -392,7 +392,7 @@ 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_reports_already_handled() -> None: +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=[]))