From 00e9031edc502e21f15677ea86ba10ab97df6009 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Mon, 26 Oct 2026 07:36:44 +0100 Subject: [PATCH 1/3] fix: retry a locked shared RQ request that is not found yet --- .../_apify/_request_queue_shared_client.py | 3 + .../test_apify_request_queue_client.py | 68 +++++++++++++++++++ 2 files changed, 71 insertions(+) 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 486bba65..a6815618 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_shared_client.py @@ -293,6 +293,7 @@ async def fetch_next_request(self) -> Request | None: extra={'next_request_id': next_request_id}, ) self._requests_in_progress.discard(next_request_id) + self._queue_head.append(next_request_id) return None # If the request was already handled, skip it @@ -314,6 +315,8 @@ async def fetch_next_request(self) -> Request | None: extra={'next_request_id': next_request_id}, ) self._requests_in_progress.discard(next_request_id) + # This client still holds the lock, so no head listing returns the request; requeue it locally. + self._queue_head.append(next_request_id) return None return request 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 fcd632dd..fbfd0254 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -820,6 +820,74 @@ async def test_fetch_next_request_skips_when_lock_not_reacquired() -> None: assert second.unique_key == request.unique_key +async def test_fetch_next_request_retries_locked_request_not_found_yet() -> None: + """A locked request that `get_request` cannot find yet is retried by the next fetch, not after its lock lapses.""" + client, api_client = _make_shared_client() + request = Request.from_url('https://example.com/1') + future = datetime.now(tz=UTC) + timedelta(seconds=180) + + # Like the platform, the head lists the request once; later listings skip it while this client holds the lock. + api_client.list_and_lock_head = AsyncMock( + side_effect=chain( + [_locked_head([_locked_item(request, lock_expires_at=future)])], + repeat(_empty_locked_head(queue_has_locked_requests=True)), + ) + ) + api_client.get_request = AsyncMock(side_effect=[None, _client_request(request, handled_at=None)]) + + first = await client.fetch_next_request() + second = await client.fetch_next_request() + + assert first is None + assert second is not None + assert second.unique_key == request.unique_key + + +async def test_fetch_next_request_retries_uncached_request_not_found_yet() -> None: + """A locked request evicted from the cache that `get_request` cannot find yet is retried by a later fetch.""" + api_client = AsyncMock() + metadata = _make_metadata() + # A single-entry cache makes listing the second request evict the first. + client = ApifyRequestQueueSharedClient( + api_client=api_client, + metadata=metadata, + cache_size=1, + metadata_getter=AsyncMock(return_value=metadata), + ) + evicted = Request.from_url('https://example.com/1') + cached = Request.from_url('https://example.com/2') + evicted_id = unique_key_to_request_id(evicted.unique_key) + future = datetime.now(tz=UTC) + timedelta(seconds=180) + + api_client.list_and_lock_head = AsyncMock( + side_effect=chain( + [ + _locked_head( + [_locked_item(evicted, lock_expires_at=future), _locked_item(cached, lock_expires_at=future)] + ) + ], + repeat(_empty_locked_head(queue_has_locked_requests=True)), + ) + ) + api_client.prolong_request_lock = AsyncMock(return_value=RequestLockInfo(lock_expires_at=future)) + evicted_lookups = iter([None, _client_request(evicted, handled_at=None), _client_request(evicted, handled_at=None)]) + api_client.get_request = AsyncMock( + side_effect=lambda request_id: ( + next(evicted_lookups) if request_id == evicted_id else _client_request(cached, handled_at=None) + ) + ) + + first = await client.fetch_next_request() + second = await client.fetch_next_request() + third = await client.fetch_next_request() + + assert first is None + assert second is not None + assert second.unique_key == cached.unique_key + assert third is not None + assert third.unique_key == evicted.unique_key + + async def test_is_finished_false_while_request_in_progress() -> None: """`is_finished` stays False while a fetched request is still in progress, even if the head lists empty.""" client, api_client = _make_shared_client() From 4f28f3dc3b656db2d1e0634d8017e1c500dbcb6a Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Mon, 26 Oct 2026 08:17:35 +0100 Subject: [PATCH 2/3] fix: count requeued request IDs in shared RQ is_empty --- .../storage_clients/_apify/_request_queue_shared_client.py | 3 ++- tests/unit/storage_clients/test_apify_request_queue_client.py | 3 +++ 2 files changed, 5 insertions(+), 1 deletion(-) 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 a6815618..ba9dd07c 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_shared_client.py @@ -468,7 +468,8 @@ async def _confirm_requests_handled(self, request_ids: Sequence[str]) -> bool: async def _is_empty(self) -> bool: """Check whether anything is available to fetch. Lock-free core of `is_empty`, caller must hold the lock.""" head = await self._list_head(limit=1) - return len(head.items) == 0 + # The cached head lists only hydrated requests, but `fetch_next_request` also hydrates an evicted ID on demand. + return len(head.items) == 0 and not self._queue_head async def _get_metadata_estimate(self) -> RequestQueueMetadata: """Try to get cached metadata first. If multiple clients, fuse with global metadata. 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 fbfd0254..bbb59fef 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -879,11 +879,14 @@ async def test_fetch_next_request_retries_uncached_request_not_found_yet() -> No first = await client.fetch_next_request() second = await client.fetch_next_request() + # Only the evicted ID is left in the local head; the crawler fetches again only while the RQ reports non-empty. + empty_before_third = await client.is_empty() third = await client.fetch_next_request() assert first is None assert second is not None assert second.unique_key == cached.unique_key + assert empty_before_third is False assert third is not None assert third.unique_key == evicted.unique_key From 1a7987c5ac1ac0070d448d941dda08f3ea42121e Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Mon, 26 Oct 2026 08:17:36 +0100 Subject: [PATCH 3/3] chore: move requeue comment and shorten test docstring --- .../storage_clients/_apify/_request_queue_shared_client.py | 2 +- tests/unit/storage_clients/test_apify_request_queue_client.py | 2 +- 2 files changed, 2 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 ba9dd07c..22cc6f09 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_shared_client.py @@ -293,6 +293,7 @@ async def fetch_next_request(self) -> Request | None: extra={'next_request_id': next_request_id}, ) self._requests_in_progress.discard(next_request_id) + # This client still holds the lock, so no head listing returns the request; requeue it locally. self._queue_head.append(next_request_id) return None @@ -315,7 +316,6 @@ async def fetch_next_request(self) -> Request | None: extra={'next_request_id': next_request_id}, ) self._requests_in_progress.discard(next_request_id) - # This client still holds the lock, so no head listing returns the request; requeue it locally. self._queue_head.append(next_request_id) return None 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 bbb59fef..425e3df6 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -821,7 +821,7 @@ async def test_fetch_next_request_skips_when_lock_not_reacquired() -> None: async def test_fetch_next_request_retries_locked_request_not_found_yet() -> None: - """A locked request that `get_request` cannot find yet is retried by the next fetch, not after its lock lapses.""" + """A locked request that `get_request` cannot find yet is returned by the next fetch.""" client, api_client = _make_shared_client() request = Request.from_url('https://example.com/1') future = datetime.now(tz=UTC) + timedelta(seconds=180)