From c763e7530376e8c22cea6079fd326dd1c55e1615 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 11 Aug 2026 09:55:48 +0200 Subject: [PATCH 1/3] test: deflake is_finished assertions in shared request queue mode --- tests/integration/test_request_queue.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/tests/integration/test_request_queue.py b/tests/integration/test_request_queue.py index f53224b6..8c74a4c3 100644 --- a/tests/integration/test_request_queue.py +++ b/tests/integration/test_request_queue.py @@ -1241,7 +1241,14 @@ async def test_request_queue_is_finished_and_is_empty( ) await request_queue_apify.add_request(Request.from_url('http://example.com')) - assert not await request_queue_apify.is_finished() + # `is_finished` reads the queue head, which reflects the newly added request only after propagation, so poll + # until the queue stops reporting itself as finished. + assert not await poll_until_condition( + request_queue_apify.is_finished, + lambda finished: not finished, + timeout=rq_poll_timeout, + backoff_factor=2, + ), 'RequestQueue should not be finished after a request is added.' fetched = await poll_until_condition( request_queue_apify.fetch_next_request, timeout=rq_poll_timeout, backoff_factor=2 @@ -1251,9 +1258,12 @@ async def test_request_queue_is_finished_and_is_empty( assert await poll_until_condition(request_queue_apify.is_empty, timeout=rq_poll_timeout, backoff_factor=2), ( 'RequestQueue should be empty because queue does not contain any requests for fetching.' ) - assert not await request_queue_apify.is_finished(), ( - 'RequestQueue should not be finished unless the request is marked as handled.' - ) + assert not await poll_until_condition( + request_queue_apify.is_finished, + lambda finished: not finished, + timeout=rq_poll_timeout, + backoff_factor=2, + ), 'RequestQueue should not be finished unless the request is marked as handled.' await request_queue_apify.mark_request_as_handled(fetched) assert await poll_until_condition(request_queue_apify.is_empty, timeout=rq_poll_timeout, backoff_factor=2) From 87af5b8df8660565e0b1c1ae909106dafcff5fb0 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 14 Aug 2026 09:24:00 +0200 Subject: [PATCH 2/3] fix: verify shared request queue is_finished against per-request reads --- .../_apify/_request_queue_shared_client.py | 32 ++++- tests/integration/test_request_queue.py | 18 +-- .../test_apify_request_queue_client.py | 120 +++++++++++++++++- 3 files changed, 154 insertions(+), 16 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 d403d71b..83c297dc 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_shared_client.py @@ -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.""" diff --git a/tests/integration/test_request_queue.py b/tests/integration/test_request_queue.py index 8c74a4c3..f53224b6 100644 --- a/tests/integration/test_request_queue.py +++ b/tests/integration/test_request_queue.py @@ -1241,14 +1241,7 @@ async def test_request_queue_is_finished_and_is_empty( ) await request_queue_apify.add_request(Request.from_url('http://example.com')) - # `is_finished` reads the queue head, which reflects the newly added request only after propagation, so poll - # until the queue stops reporting itself as finished. - assert not await poll_until_condition( - request_queue_apify.is_finished, - lambda finished: not finished, - timeout=rq_poll_timeout, - backoff_factor=2, - ), 'RequestQueue should not be finished after a request is added.' + assert not await request_queue_apify.is_finished() fetched = await poll_until_condition( request_queue_apify.fetch_next_request, timeout=rq_poll_timeout, backoff_factor=2 @@ -1258,12 +1251,9 @@ async def test_request_queue_is_finished_and_is_empty( assert await poll_until_condition(request_queue_apify.is_empty, timeout=rq_poll_timeout, backoff_factor=2), ( 'RequestQueue should be empty because queue does not contain any requests for fetching.' ) - assert not await poll_until_condition( - request_queue_apify.is_finished, - lambda finished: not finished, - timeout=rq_poll_timeout, - backoff_factor=2, - ), 'RequestQueue should not be finished unless the request is marked as handled.' + assert not await request_queue_apify.is_finished(), ( + 'RequestQueue should not be finished unless the request is marked as handled.' + ) await request_queue_apify.mark_request_as_handled(fetched) assert await poll_until_condition(request_queue_apify.is_empty, timeout=rq_poll_timeout, backoff_factor=2) 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 91b117c4..de37b3c5 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -8,7 +8,15 @@ import pytest -from apify_client._models import AddedRequest, BatchAddResult, RequestDraft, RequestQueueHead, RequestQueueStats +from apify_client._models import ( + AddedRequest, + BatchAddResult, + LockedRequestQueueHead, + RequestDraft, + RequestQueueHead, + RequestQueueStats, +) +from apify_client._models import Request as ClientRequest from crawlee.storage_clients.models import AddRequestsResponse, RequestQueueMetadata from apify import Request @@ -92,6 +100,26 @@ 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 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 +366,93 @@ 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_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 From e7cd6b49873de719338bb91d0bafce64d6eb0ea1 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Fri, 14 Aug 2026 10:50:10 +0200 Subject: [PATCH 3/3] fix: cache post-update handled state when marking and reclaiming requests --- .../_apify/_request_queue_shared_client.py | 12 ++-- .../test_apify_request_queue_client.py | 71 +++++++++++++++++++ 2 files changed, 77 insertions(+), 6 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 83c297dc..17484b67 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/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, ) 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 de37b3c5..b71e1a4c 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -15,6 +15,7 @@ RequestDraft, RequestQueueHead, RequestQueueStats, + RequestRegistration, ) from apify_client._models import Request as ClientRequest from crawlee.storage_clients.models import AddRequestsResponse, RequestQueueMetadata @@ -120,6 +121,15 @@ def _client_request(request: Request, *, handled_at: datetime | None) -> ClientR ) +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) @@ -431,6 +441,67 @@ async def test_shared_is_finished_true_on_queue_with_no_known_requests() -> None 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()