From 3fbf757affaaf51bf4fa2c49a54e0a264e2938de Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Mon, 20 Apr 2026 17:37:26 +0200 Subject: [PATCH 1/2] fix(request-queue): discard in-progress id when platform returns None MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ApifyRequestQueueSingleClient.fetch_next_request` committed the id to `_requests_in_progress` before awaiting `_get_request_by_id`. When the platform's `get_request` returns `None` (e.g. propagation delay after a fresh `_list_head`), the method returned `None` to the caller but the id stayed in `_requests_in_progress` forever — permanently poisoning `is_empty()` and excluding the id from future `_list_head` reconciliation, with no public recovery path (the caller had no `Request` to reclaim). Discard the id on the `None` path and continue to the next head entry. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../_apify/_request_queue_single_client.py | 8 +++++++- .../test_apify_request_queue_client.py | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/src/apify/storage_clients/_apify/_request_queue_single_client.py b/src/apify/storage_clients/_apify/_request_queue_single_client.py index 05614cd8f..d1f320918 100644 --- a/src/apify/storage_clients/_apify/_request_queue_single_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_single_client.py @@ -192,7 +192,13 @@ async def fetch_next_request(self) -> Request | None: request_id = self._head_requests.pop() if request_id not in self._requests_in_progress and request_id not in self._requests_already_handled: self._requests_in_progress.add(request_id) - return await self._get_request_by_id(request_id) + request = await self._get_request_by_id(request_id) + if request is None: + # Platform could not return the request (e.g. data propagation delay); do not + # leak the id into `_requests_in_progress`, or `is_empty()` would never be True. + self._requests_in_progress.discard(request_id) + continue + return request # No request locally and the ones returned from the platform are already in progress. 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 6117afd6f..65351ff2d 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -84,3 +84,20 @@ async def test_list_head_limit(in_progress_count: int, expected_limit: int) -> N await client._list_head() api_client.list_head.assert_awaited_once_with(limit=expected_limit) + + +async def test_fetch_next_request_does_not_leak_in_progress_on_none_response() -> None: + """Regression test: when ``get_request`` returns ``None`` for a head id (e.g. due to + platform data propagation delay), the id must not remain in ``_requests_in_progress``, + otherwise ``is_empty()`` can never become ``True`` again. + """ + client, api_client = _make_single_client() + api_client.list_head = AsyncMock(return_value={'items': [], 'hadMultipleClients': False}) + api_client.get_request = AsyncMock(return_value=None) + client._head_requests.append('missing-id') + + result = await client.fetch_next_request() + + assert result is None + assert 'missing-id' not in client._requests_in_progress + assert await client.is_empty() From afde14a10d28181f430c1fd78753f57e5b0f27c2 Mon Sep 17 00:00:00 2001 From: Vlada Dusek Date: Tue, 21 Apr 2026 12:04:45 +0200 Subject: [PATCH 2/2] rm test, just a defensive guard --- .../_apify/_request_queue_single_client.py | 5 +++-- .../test_apify_request_queue_client.py | 17 ----------------- 2 files changed, 3 insertions(+), 19 deletions(-) diff --git a/src/apify/storage_clients/_apify/_request_queue_single_client.py b/src/apify/storage_clients/_apify/_request_queue_single_client.py index d1f320918..9bf9e58ec 100644 --- a/src/apify/storage_clients/_apify/_request_queue_single_client.py +++ b/src/apify/storage_clients/_apify/_request_queue_single_client.py @@ -194,8 +194,9 @@ async def fetch_next_request(self) -> Request | None: self._requests_in_progress.add(request_id) request = await self._get_request_by_id(request_id) if request is None: - # Platform could not return the request (e.g. data propagation delay); do not - # leak the id into `_requests_in_progress`, or `is_empty()` would never be True. + # Defensive guard against an unexpected `None` from the platform: leaving the id in + # `_requests_in_progress` would make `is_empty()` never settle and filter the id out of future + # head reconciliations, with no recovery path for the caller. self._requests_in_progress.discard(request_id) continue 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 65351ff2d..6117afd6f 100644 --- a/tests/unit/storage_clients/test_apify_request_queue_client.py +++ b/tests/unit/storage_clients/test_apify_request_queue_client.py @@ -84,20 +84,3 @@ async def test_list_head_limit(in_progress_count: int, expected_limit: int) -> N await client._list_head() api_client.list_head.assert_awaited_once_with(limit=expected_limit) - - -async def test_fetch_next_request_does_not_leak_in_progress_on_none_response() -> None: - """Regression test: when ``get_request`` returns ``None`` for a head id (e.g. due to - platform data propagation delay), the id must not remain in ``_requests_in_progress``, - otherwise ``is_empty()`` can never become ``True`` again. - """ - client, api_client = _make_single_client() - api_client.list_head = AsyncMock(return_value={'items': [], 'hadMultipleClients': False}) - api_client.get_request = AsyncMock(return_value=None) - client._head_requests.append('missing-id') - - result = await client.fetch_next_request() - - assert result is None - assert 'missing-id' not in client._requests_in_progress - assert await client.is_empty()