fix: verify shared request queue is_finished against per-request reads

src/apify/storage_clients/_apify/_request_queue_shared_client.py

@@ -264,8 +264,6 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
         if request.handled_at is None:
             request.handled_at = datetime.now(tz=UTC)
 
-        if cached_request := self._requests_cache.get(request_id):
-            cached_request.was_already_handled = request.was_already_handled
         try:
             # Update the request in the API
             processed_request = await self._update_request(request)
@@ -277,10 +275,11 @@ async def mark_request_as_handled(self, request: Request) -> ProcessedRequest |
                 self.metadata.handled_request_count += 1
                 self.metadata.pending_request_count -= 1
 
-            # Update the cache with the handled request
+            # Cache the request as handled. The platform response's `was_already_handled` reports the state
+            # before this update, so it must not be cached as the request's current state.
             self._cache_request(
                 cache_key=request_id,
-                processed_request=processed_request,
+                processed_request=processed_request.model_copy(update={'was_already_handled': True}),
                 hydrated_request=request,
             )
         except Exception:
@@ -314,11 +313,12 @@ async def reclaim_request(
                     self.metadata.handled_request_count -= 1
                     self.metadata.pending_request_count += 1
 
-                # Update the cache
+                # Cache the request as pending again. The platform response's `was_already_handled` reports the
+                # state before this update, so it must not be cached as the request's current state.
                 request_id = unique_key_to_request_id(request.unique_key)
                 self._cache_request(
                     request_id,
-                    processed_request,
+                    processed_request.model_copy(update={'was_already_handled': False}),
                     hydrated_request=request,
                 )
 
@@ -344,7 +344,37 @@ async def is_finished(self) -> bool:
         """Specific implementation of this method for the RQ shared access mode."""
         async with self._fetch_lock:
             # Order of operations is important here, because affects on `_queue_has_locked_requests`.
-            return await self._is_empty() and not self._queue_has_locked_requests
+            if not await self._is_empty() or self._queue_has_locked_requests:
+                return False
+
+            # The head listing is eventually consistent: it can miss a just-added request (and report no locked
+            # requests) for a short while, so an empty head alone is not proof the queue is finished. Confirm the
+            # verdict against per-request reads before reporting `True`.
+            return await self._all_known_requests_handled()
+
+    async def _all_known_requests_handled(self) -> bool:
+        """Confirm via the API that every request this client knows about was handled. Caller must hold the lock.
+
+        Unlike the head listing, fetching a request by id is strongly consistent, so each locally known request
+        that was not yet seen handled is re-checked against the platform. A request that is missing (not yet
+        propagated) or unhandled (pending, or locked by another client) means the queue is not finished. Requests
+        confirmed as handled are remembered in the cache, so each one is verified at most once.
+        """
+        if self._requests_being_added:
+            # An in-flight `add_batch_of_requests` call is about to commit new requests.
+            return False
+
+        for request_id, cached_request in list(self._requests_cache.items()):
+            if cached_request.was_already_handled:
+                continue
+
+            request = await self._get_request_by_id(request_id)
+            if request is None or request.handled_at is None:
+                return False
+
+            cached_request.was_already_handled = True
+
+        return True
 
     async def _is_empty(self) -> bool:
         """Check whether anything is available to fetch. Lock-free core of `is_empty`, caller must hold the lock."""

tests/unit/storage_clients/test_apify_request_queue_client.py

@@ -8,7 +8,16 @@
 
 import pytest
 
-from apify_client._models import AddedRequest, BatchAddResult, RequestDraft, RequestQueueHead, RequestQueueStats
+from apify_client._models import (
+    AddedRequest,
+    BatchAddResult,
+    LockedRequestQueueHead,
+    RequestDraft,
+    RequestQueueHead,
+    RequestQueueStats,
+    RequestRegistration,
+)
+from apify_client._models import Request as ClientRequest
 from crawlee.storage_clients.models import AddRequestsResponse, RequestQueueMetadata
 
 from apify import Request
@@ -92,6 +101,35 @@ def _make_shared_client(
     return client, api_client
 
 
+def _empty_locked_head(*, queue_has_locked_requests: bool = False) -> LockedRequestQueueHead:
+    """Build an empty `list_and_lock_head` response, optionally reporting locked requests."""
+    return LockedRequestQueueHead(
+        limit=1,
+        queue_modified_at=datetime.now(tz=UTC),
+        queue_has_locked_requests=queue_has_locked_requests,
+        had_multiple_clients=True,
+        lock_secs=60,
+        items=[],
+    )
+
+
+def _client_request(request: Request, *, handled_at: datetime | None) -> ClientRequest:
+    """Build a `get_request` response for the given request in the given handled state."""
+    return ClientRequest.model_validate(
+        request.model_dump(by_alias=True)
+        | {'id': unique_key_to_request_id(request.unique_key), 'handledAt': handled_at}
+    )
+
+
+def _request_registration(request: Request, *, was_already_handled: bool) -> RequestRegistration:
+    """Build an `update_request` response reporting the given pre-update handled state."""
+    return RequestRegistration(
+        request_id=unique_key_to_request_id(request.unique_key),
+        was_already_present=True,
+        was_already_handled=was_already_handled,
+    )
+
+
 def test_unique_key_to_request_id_length() -> None:
     unique_key = 'exampleKey123'
     request_id = unique_key_to_request_id(unique_key, request_id_length=15)
@@ -338,3 +376,154 @@ async def test_partial_unprocessed_commits_only_accepted_requests(access: str) -
     assert api_client.batch_add_requests.await_args is not None
     resent = api_client.batch_add_requests.await_args.kwargs['requests']
     assert [request['uniqueKey'] for request in resent] == [rejected.unique_key]
+
+
+@pytest.mark.parametrize(
+    'platform_request_visible',
+    [
+        pytest.param(True, id='still_pending'),
+        pytest.param(False, id='not_yet_visible'),
+    ],
+)
+async def test_shared_is_finished_false_while_known_request_unhandled(*, platform_request_visible: bool) -> None:
+    """An empty, lock-free head listing does not report the queue finished while a known request is unhandled:
+    the eventually consistent head can miss a just-added request, so its state is confirmed by fetching it."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/1')
+    request_id = unique_key_to_request_id(request.unique_key)
+
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed([request]))
+    await client.add_batch_of_requests([request])
+
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head())
+    api_client.get_request = AsyncMock(
+        return_value=_client_request(request, handled_at=None) if platform_request_visible else None
+    )
+
+    assert await client.is_finished() is False
+    api_client.get_request.assert_awaited_once_with(request_id)
+
+
+async def test_shared_is_finished_true_once_known_requests_confirmed_handled() -> None:
+    """The queue reports finished once every known request is confirmed handled, and the confirmation is cached
+    so repeated `is_finished` calls do not re-fetch the request."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/1')
+
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed([request]))
+    await client.add_batch_of_requests([request])
+
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head())
+    api_client.get_request = AsyncMock(return_value=_client_request(request, handled_at=datetime.now(tz=UTC)))
+
+    assert await client.is_finished() is True
+    assert await client.is_finished() is True
+    assert api_client.get_request.await_count == 1
+
+
+async def test_shared_is_finished_false_when_head_reports_locked_requests() -> None:
+    """Locked requests reported by the head listing mean the queue is not finished, without any per-request reads."""
+    client, api_client = _make_shared_client()
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head(queue_has_locked_requests=True))
+    api_client.get_request = AsyncMock()
+
+    assert await client.is_finished() is False
+    api_client.get_request.assert_not_awaited()
+
+
+async def test_shared_is_finished_true_on_queue_with_no_known_requests() -> None:
+    """An empty, lock-free queue with no locally known requests reports finished without per-request reads."""
+    client, api_client = _make_shared_client()
+    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()
+
+
+async def test_shared_is_finished_true_after_this_client_marked_request_handled() -> None:
+    """A request this client marked handled is trusted from the cache, so `is_finished` needs no per-request read."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/1')
+
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed([request]))
+    await client.add_batch_of_requests([request])
+
+    # The platform reports the pre-update state, so a first-time handle comes back as not yet handled.
+    api_client.update_request = AsyncMock(return_value=_request_registration(request, was_already_handled=False))
+    assert await client.mark_request_as_handled(request) is not None
+
+    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()
+
+
+async def test_shared_is_finished_false_after_failed_mark_request_as_handled() -> None:
+    """A failed `mark_request_as_handled` leaves the request unconfirmed, so `is_finished` re-checks it."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/1')
+    request_id = unique_key_to_request_id(request.unique_key)
+
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed([request]))
+    await client.add_batch_of_requests([request])
+
+    api_client.update_request = AsyncMock(side_effect=RuntimeError('network down'))
+    assert await client.mark_request_as_handled(request) is None
+
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head())
+    api_client.get_request = AsyncMock(return_value=_client_request(request, handled_at=None))
+
+    assert await client.is_finished() is False
+    api_client.get_request.assert_awaited_once_with(request_id)
+
+
+async def test_shared_is_finished_false_after_reclaiming_handled_request() -> None:
+    """A reclaimed previously-handled request is pending again, so `is_finished` re-checks it via the platform."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/1')
+    request_id = unique_key_to_request_id(request.unique_key)
+
+    api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed([request]))
+    await client.add_batch_of_requests([request])
+
+    api_client.update_request = AsyncMock(return_value=_request_registration(request, was_already_handled=False))
+    await client.mark_request_as_handled(request)
+
+    # Reclaim the handled request: the platform reports the pre-update (handled) state.
+    api_client.update_request = AsyncMock(return_value=_request_registration(request, was_already_handled=True))
+    assert await client.reclaim_request(request) is not None
+
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head())
+    api_client.get_request = AsyncMock(return_value=_client_request(request, handled_at=None))
+
+    assert await client.is_finished() is False
+    api_client.get_request.assert_awaited_once_with(request_id)
+
+
+async def test_shared_is_finished_false_while_add_batch_in_flight() -> None:
+    """The queue does not report finished while an `add_batch_of_requests` call is still in flight."""
+    client, api_client = _make_shared_client()
+    request = Request.from_url('https://example.com/1')
+
+    in_flight = asyncio.Event()
+    release = asyncio.Event()
+
+    async def batch_add(*, requests: list, forefront: bool = False) -> BatchAddResult:  # noqa: ARG001
+        in_flight.set()
+        await release.wait()
+        return _batch_result_all_processed([request])
+
+    api_client.batch_add_requests = AsyncMock(side_effect=batch_add)
+    api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head())
+    api_client.get_request = AsyncMock()
+
+    add_task = asyncio.create_task(client.add_batch_of_requests([request]))
+    await in_flight.wait()
+
+    assert await client.is_finished() is False
+    api_client.get_request.assert_not_awaited()
+
+    release.set()
+    await add_task