From c763e7530376e8c22cea6079fd326dd1c55e1615 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 11 Aug 2026 09:55:48 +0200 Subject: [PATCH 1/5] 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/5] 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/5] 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() From e7069798ae3c8d55b7b7b64e19974cff977207d3 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 18 Aug 2026 14:43:51 +0200 Subject: [PATCH 4/5] perf: verify only unconfirmed requests in is_finished, and in parallel batches --- .../_apify/_request_queue_shared_client.py | 54 ++++++++++++++---- .../test_apify_request_queue_client.py | 56 +++++++++++++++++++ 2 files changed, 99 insertions(+), 11 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 17484b67..edb68147 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_shared_client.py @@ -43,6 +43,9 @@ class ApifyRequestQueueSharedClient: _DEFAULT_LOCK_TIME: Final[timedelta] = timedelta(minutes=3) """The default lock time for requests in the queue.""" + _VERIFICATION_BATCH_SIZE: Final[int] = 10 + """How many requests `is_finished` confirms with the platform in parallel.""" + def __init__( self, *, @@ -89,6 +92,13 @@ def __init__( add fails. """ + self._unhandled_request_ids = set[str]() + """Ids of locally known requests not yet confirmed handled, maintained by `_cache_request`. + + An index over `_requests_cache` so that `is_finished` verification does not have to walk the whole cache, + which holds up to a million entries and is consulted on every poll of the crawler's finished check. + """ + self._queue_has_locked_requests: bool | None = None """Whether the queue contains requests currently locked by other clients.""" @@ -355,26 +365,43 @@ async def is_finished(self) -> bool: 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. + 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. Confirmed requests stop being tracked as + unhandled, so each one is verified at most once: a client that handled its own requests has nothing left to + check, and one that is waiting on a straggler only re-fetches that straggler. """ 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 + unhandled = list(self._unhandled_request_ids) + for start in range(0, len(unhandled), self._VERIFICATION_BATCH_SIZE): + if not await self._confirm_requests_handled(unhandled[start : start + self._VERIFICATION_BATCH_SIZE]): + return False - request = await self._get_request_by_id(request_id) + return True + + async def _confirm_requests_handled(self, request_ids: Sequence[str]) -> bool: + """Fetch the given requests in parallel and report whether the platform has all of them handled. + + Requests the platform confirms as handled are dropped from `_unhandled_request_ids` even when another + request in the same batch reports the queue as unfinished, so no confirmation is fetched twice. + """ + requests = await asyncio.gather(*(self._get_request_by_id(request_id) for request_id in request_ids)) + + all_handled = True + for request_id, request in zip(request_ids, requests, strict=True): + # A request that is missing (not yet propagated) or unhandled (pending, or locked by another client) + # means the queue is not finished. if request is None or request.handled_at is None: - return False + all_handled = False + continue - cached_request.was_already_handled = True + self._unhandled_request_ids.discard(request_id) + if cached_request := self._requests_cache.get(request_id): + cached_request.was_already_handled = True - return True + return all_handled async def _is_empty(self) -> bool: """Check whether anything is available to fetch. Lock-free core of `is_empty`, caller must hold the lock.""" @@ -587,3 +614,8 @@ def _cache_request( hydrated=hydrated_request, lock_expires_at=None, ) + + if processed_request.was_already_handled: + self._unhandled_request_ids.discard(cache_key) + else: + self._unhandled_request_ids.add(cache_key) 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 b71e1a4c..0f697a54 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -527,3 +527,59 @@ async def batch_add(*, requests: list, forefront: bool = False) -> BatchAddResul release.set() await add_task + + +async def test_shared_is_finished_confirms_known_requests_in_bounded_parallel_batches() -> None: + """`is_finished` confirms unhandled requests concurrently, in batches bounded by `_VERIFICATION_BATCH_SIZE`.""" + client, api_client = _make_shared_client() + batch_size = ApifyRequestQueueSharedClient._VERIFICATION_BATCH_SIZE + requests = [Request.from_url(f'https://example.com/{i}') for i in range(batch_size * 2 + 5)] + requests_by_id = {unique_key_to_request_id(request.unique_key): request for request in requests} + + api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed(requests)) + await client.add_batch_of_requests(requests) + + in_flight = 0 + peak_in_flight = 0 + + async def get_request(request_id: str) -> ClientRequest: + nonlocal in_flight, peak_in_flight + in_flight += 1 + peak_in_flight = max(peak_in_flight, in_flight) + # Yield to the loop so the whole batch is in flight before any of its calls completes. + await asyncio.sleep(0) + in_flight -= 1 + return _client_request(requests_by_id[request_id], handled_at=datetime.now(tz=UTC)) + + api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head()) + api_client.get_request = AsyncMock(side_effect=get_request) + + assert await client.is_finished() is True + assert api_client.get_request.await_count == len(requests) + assert peak_in_flight == batch_size + + +async def test_shared_is_finished_does_not_refetch_requests_confirmed_in_an_unfinished_batch() -> None: + """Requests confirmed handled alongside an unhandled one are not fetched again by the next `is_finished`.""" + client, api_client = _make_shared_client() + handled, straggler = (Request.from_url(f'https://example.com/{i}') for i in range(2)) + straggler_id = unique_key_to_request_id(straggler.unique_key) + + api_client.batch_add_requests = AsyncMock(return_value=_batch_result_all_processed([handled, straggler])) + await client.add_batch_of_requests([handled, straggler]) + + api_client.list_and_lock_head = AsyncMock(return_value=_empty_locked_head()) + api_client.get_request = AsyncMock( + side_effect=lambda request_id: _client_request( + straggler if request_id == straggler_id else handled, + handled_at=None if request_id == straggler_id else datetime.now(tz=UTC), + ) + ) + + assert await client.is_finished() is False + assert api_client.get_request.await_count == 2 + + api_client.get_request = AsyncMock(return_value=_client_request(straggler, handled_at=datetime.now(tz=UTC))) + + assert await client.is_finished() is True + api_client.get_request.assert_awaited_once_with(straggler_id) From c3414aafc047e263ba6c61e3ce3f6c77e80dd211 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Thu, 20 Aug 2026 10:31:58 +0200 Subject: [PATCH 5/5] docs: clarify is_finished comments and normalize ID spelling in shared RQ client --- .../_apify/_request_queue_shared_client.py | 25 +++++++++++-------- 1 file changed, 14 insertions(+), 11 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 6a191e91..2ffe5706 100644 --- a/src/apify/storage_clients/_apify/_request_queue_shared_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_shared_client.py @@ -106,7 +106,7 @@ def __init__( """ self._unhandled_request_ids = set[str]() - """Ids of locally known requests not yet confirmed handled, maintained by `_cache_request`. + """IDs of locally known requests not yet confirmed handled, maintained by `_cache_request`. An index over `_requests_cache` so that `is_finished` verification does not have to walk the whole cache, which holds up to a million entries and is consulted on every poll of the crawler's finished check. @@ -299,7 +299,7 @@ async def fetch_next_request(self) -> Request | None: return None # `_get_or_hydrate_request` may return a request from the queue-head cache, which is populated by - # `list_and_lock_head` and only holds a partial request (no user data, no headers). Re-fetch it by id to + # `list_and_lock_head` and only holds a partial request (no user data, no headers). Re-fetch it by ID to # guarantee the caller gets the full request object. request = await self._get_request_by_id(next_request_id) if request is None: @@ -402,8 +402,9 @@ async def is_empty(self) -> bool: 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`. - # A locally in-progress request keeps the queue unfinished even when the head lists empty. + # `_is_empty` has to be awaited first: listing the head is what refreshes `_queue_has_locked_requests`, + # which stays `None` until then. A request this client is still processing keeps the queue unfinished even + # when the head lists empty. if not await self._is_empty() or self._queue_has_locked_requests or self._requests_in_progress: return False @@ -415,10 +416,12 @@ async def is_finished(self) -> bool: 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. Confirmed requests stop being tracked as - unhandled, so each one is verified at most once: a client that handled its own requests has nothing left to - check, and one that is waiting on a straggler only re-fetches that straggler. + Each locally known request not yet seen handled is re-read by ID, a signal independent of the head listing + that missed it. The check is one-sided on purpose: only a request the platform reports as handled counts as + done, so a by-ID read that lags behind can delay the finished verdict but never produce it too early. + Confirmed requests stop being tracked as unhandled, so each one is verified at most once: a client that + handled its own requests has nothing left to check, and one waiting on a straggler only re-fetches that + straggler. """ if self._requests_being_added: # An in-flight `add_batch_of_requests` call is about to commit new requests. @@ -503,7 +506,7 @@ async def _ensure_lock_window( window starting at hand-off. Args: - request_id: Id of the request about to be handed to a consumer. + request_id: ID of the request about to be handed to a consumer. lock_expires_at: When the currently held lock on the request expires, if known. now: The current time, as observed when the request was picked from the queue head. @@ -533,10 +536,10 @@ async def _ensure_lock_window( return True async def _get_or_hydrate_request(self, request_id: str) -> Request | None: - """Get a request by id, either from cache or by fetching from API. + """Get a request by ID, either from cache or by fetching from API. Args: - request_id: Id of the request to get. + request_id: ID of the request to get. Returns: The request if found and valid, otherwise None.