fix: retry a locked shared RQ request that is not found yet

src/apify/storage_clients/_apify/_request_queue_shared_client.py

@@ -293,6 +293,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
 
         # If the request was already handled, skip it
@@ -314,6 +316,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
 
         return request
@@ -465,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.

tests/unit/storage_clients/test_apify_request_queue_client.py

@@ -820,6 +820,77 @@ 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 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)
+
+    # 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()
+    # 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
+
+
 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()