fix: Add exponential backoff for shared RQ propagation delay in integration tests

In shared request queue mode, the Apify API has an eventual consistency
propagation delay before newly added or reclaimed requests become visible
in fetch_next_request(). Many integration tests were missing the
call_with_exp_backoff() retry pattern, causing sporadic failures.

Also added rq_access_mode parameter to call_with_exp_backoff() so callers
can pass it directly instead of wrapping every call in an if/else block.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: vdusek

tests/integration/_utils.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from typing import TYPE_CHECKING, TypeVar
+from typing import TYPE_CHECKING, Literal, TypeVar
 
 from crawlee._utils.crypto import crypto_random_object_id
 
@@ -16,7 +16,7 @@
 async def call_with_exp_backoff(
     fn: Callable[[], Awaitable[T]],
     *,
-    rq_access_mode: str | None = None,
+    rq_access_mode: Literal['single', 'shared'],
     max_retries: int = 3,
 ) -> T | None:
     """Call an async callable with exponential backoff retries until it returns a truthy value.
@@ -25,24 +25,27 @@ async def call_with_exp_backoff(
     become visible in the API (see https://github.com/apify/apify-sdk-python/issues/808). This helper retries with
     exponential backoff to handle that delay in integration tests.
 
-    When `rq_access_mode` is provided and is not `'shared'`, the function is called once without retries.
+    When `rq_access_mode` is `'single'`, the function is called once without retries.
     """
-    if rq_access_mode is not None and rq_access_mode != 'shared':
+    if rq_access_mode == 'single':
         return await fn()
 
-    result = None
+    if rq_access_mode == 'shared':
+        result = None
 
-    for attempt in range(max_retries):
-        result = await fn()
+        for attempt in range(max_retries):
+            result = await fn()
 
-        if result:
-            return result
+            if result:
+                return result
 
-        delay = 2**attempt
-        Actor.log.info(f'{fn} returned {result!r}, retrying in {delay}s (attempt {attempt + 1}/{max_retries})')
-        await asyncio.sleep(delay)
+            delay = 2**attempt
+            Actor.log.info(f'{fn} returned {result!r}, retrying in {delay}s (attempt {attempt + 1}/{max_retries})')
+            await asyncio.sleep(delay)
 
-    return result
+        return result
+
+    raise ValueError(f'Invalid rq_access_mode: {rq_access_mode}')
 
 
 def generate_unique_resource_name(label: str) -> str:

tests/integration/test_request_queue.py

@@ -25,6 +25,9 @@
 
     from apify.storage_clients._apify._models import ApifyRequestQueueMetadata
 
+# In shared mode, there is a propagation delay between operations so we use test helper
+# `call_with_exp_backoff` for exponential backoff. See https://github.com/apify/apify-sdk-python/issues/808.
+
 
 async def test_add_and_fetch_requests(
     request_queue_apify: RequestQueue,
@@ -42,8 +45,6 @@ async def test_add_and_fetch_requests(
         Actor.log.info(f'Adding request {i}...')
         await rq.add_request(f'https://example.com/{i}')
 
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     handled_request_count = 0
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         Actor.log.info('Fetching next request...')
@@ -79,8 +80,6 @@ async def test_add_requests_in_batches(
     total_count = await rq.get_total_count()
     Actor.log.info(f'Added {desired_request_count} requests in batch, total in queue: {total_count}')
 
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     handled_request_count = 0
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         if handled_request_count % 20 == 0:
@@ -120,8 +119,6 @@ async def test_add_non_unique_requests_in_batch(
     total_count = await rq.get_total_count()
     Actor.log.info(f'Added {desired_request_count} requests with duplicate unique keys, total in queue: {total_count}')
 
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     handled_request_count = 0
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         if handled_request_count % 20 == 0:
@@ -166,8 +163,6 @@ async def test_forefront_requests_ordering(
     Actor.log.info(f'Added 2 forefront requests, total in queue: {total_count}')
 
     # Fetch requests and verify order.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     fetched_urls = []
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         Actor.log.info(f'Fetched request: {next_request.url}')
@@ -221,8 +216,6 @@ async def test_request_unique_key_behavior(
     assert result3.was_already_present is False, f'result3.was_already_present={result3.was_already_present}'
 
     # Only 2 requests should be fetchable.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     fetched_count = 0
     fetched_requests = []
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
@@ -256,8 +249,6 @@ async def test_request_reclaim_functionality(
     Actor.log.info('Added test request')
 
     # Fetch and reclaim the request.
-    # In shared mode, there is a propagation delay before the newly added request becomes visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     fetched_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     assert fetched_request is not None
     Actor.log.info(f'Fetched request: {fetched_request.url}')
@@ -270,8 +261,6 @@ async def test_request_reclaim_functionality(
     Actor.log.info('Request reclaimed successfully')
 
     # Should be able to fetch the same request again.
-    # In shared mode, there is a propagation delay before the reclaimed request becomes visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     request2 = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
 
     assert request2 is not None
@@ -303,8 +292,6 @@ async def test_request_reclaim_with_forefront(
     Actor.log.info('Added 3 requests')
 
     # Fetch first request.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     first_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     assert first_request is not None
     Actor.log.info(f'Fetched first request: {first_request.url}')
@@ -314,8 +301,6 @@ async def test_request_reclaim_with_forefront(
     Actor.log.info('Request reclaimed to forefront')
 
     # The reclaimed request should be fetched first again.
-    # In shared mode, there is a propagation delay before the reclaimed request becomes visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
 
     assert next_request is not None
@@ -356,8 +341,6 @@ async def test_complex_request_objects(
     Actor.log.info(f'Added complex request: {complex_request.url} with method {complex_request.method}')
 
     # Fetch and verify all properties are preserved.
-    # In shared mode, there is a propagation delay before the newly added request becomes visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     fetched_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     assert fetched_request is not None, f'fetched_request={fetched_request}'
     Actor.log.info(f'Fetched request: {fetched_request.url}')
@@ -437,8 +420,6 @@ async def test_metadata_tracking(
     assert handled_after_add == 0, f'handled_after_add={handled_after_add}'
 
     # Process some requests.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     handled = 0
     if next_request:
@@ -486,8 +467,6 @@ async def test_batch_operations_performance(
     assert handled_count == 0, f'handled_count={handled_count}'
 
     # Process all requests.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     processed_count = 0
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         processed_count += 1
@@ -522,8 +501,6 @@ async def test_state_consistency(
     Actor.log.info(f'Initial total count: {initial_total}')
 
     # Simulate some requests being processed and others being reclaimed.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     processed_requests = []
     reclaimed_requests = []
 
@@ -558,8 +535,6 @@ async def test_state_consistency(
     assert current_total == 10, f'current_total={current_total}'
 
     # Process remaining requests.
-    # In shared mode, there is a propagation delay before the reclaimed requests become visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     remaining_count = 0
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         remaining_count += 1
@@ -624,8 +599,6 @@ async def test_large_batch_operations(
     assert total_count == 500, f'total_count={total_count}'
 
     # Process all in chunks to test performance.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     processed_count = 0
 
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
@@ -672,8 +645,6 @@ async def test_mixed_string_and_request_objects(
     Actor.log.info(f'Total requests in queue: {total_count}')
 
     # Fetch and verify all types work.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     fetched_requests = []
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         fetched_requests.append(next_request)
@@ -713,8 +684,6 @@ async def test_persistence_across_operations(
     Actor.log.info(f'Total count after initial batch: {initial_total}')
 
     # Process some requests.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     processed_count = 0
     next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     if next_request:
@@ -745,8 +714,6 @@ async def test_persistence_across_operations(
     assert handled_after_additional == 5, f'handled_after_additional={handled_after_additional}'
 
     # Process remaining.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     remaining_processed = 0
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         remaining_processed += 1
@@ -836,8 +803,6 @@ async def test_request_ordering_with_mixed_operations(
     Actor.log.info('Added initial requests')
 
     # Fetch one and reclaim to forefront.
-    # In shared mode, there is a propagation delay before the newly added requests become available
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     request1 = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     assert request1 is not None, f'request1={request1}'
     assert request1.url == 'https://example.com/1', f'request1.url={request1.url}'
@@ -851,7 +816,6 @@ async def test_request_ordering_with_mixed_operations(
     Actor.log.info('Added new forefront request')
 
     # Fetch all requests and verify forefront behavior.
-    # In shared mode, there is a propagation delay before the reclaimed/added requests become visible.
     urls_ordered = list[str]()
     while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode):
         urls_ordered.append(next_request.url)
@@ -1118,8 +1082,6 @@ async def test_pre_existing_request_with_user_data(
     await rq_client.add_request(req.model_dump(by_alias=True))
 
     # Fetch the request by the client under test.
-    # In shared mode, there is a propagation delay before the newly added request becomes visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     request_obtained = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
     assert request_obtained is not None
     # Test that custom_data is preserved in user_data (custom_data should be subset of obtained user_data)
@@ -1154,8 +1116,6 @@ async def test_request_queue_is_finished(
     await request_queue_apify.add_request(Request.from_url('http://example.com'))
     assert not await request_queue_apify.is_finished()
 
-    # In shared mode, there is a propagation delay before the newly added request becomes visible
-    # (see https://github.com/apify/apify-sdk-python/issues/808).
     fetched = await call_with_exp_backoff(request_queue_apify.fetch_next_request, rq_access_mode=rq_access_mode)
     assert fetched is not None
     assert not await request_queue_apify.is_finished(), (
@@ -1428,8 +1388,6 @@ async def worker() -> int:
             total_after_workers = await rq.get_total_count()
             assert total_after_workers == 20
 
-            # In shared mode, there is a propagation delay before the reclaimed requests become visible
-            # (see https://github.com/apify/apify-sdk-python/issues/808).
             remaining_count = 0
             while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode='shared'):
                 remaining_count += 1