test: Add exponential backoff for shared RQ propagation delay in integration tests (#831)

## Summary

- Added `call_with_exp_backoff(rq.fetch_next_request)` for shared mode
in all integration tests that fetch requests right after
adding/reclaiming them
- Restructured while-loops to use backoff on the first fetch, preventing
premature loop exit when the API hasn't propagated yet
- Added `request: pytest.FixtureRequest` parameter to tests that were
missing it for shared mode detection
- All changes reference the known propagation delay issue #808

## Test plan

- [ ] Integration tests pass in CI (both `single` and `shared`
parameterizations)
- [ ] No more sporadic `test_rq_long_url[shared]` failures (and similar
tests)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

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
 
@@ -13,26 +13,39 @@
 T = TypeVar('T')
 
 
-async def call_with_exp_backoff(fn: Callable[[], Awaitable[T]], *, max_retries: int = 3) -> T | None:
+async def call_with_exp_backoff(
+    fn: Callable[[], Awaitable[T]],
+    *,
+    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.
 
     In shared request queue mode, there is a propagation delay before newly added, reclaimed, or handled requests
     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 `'single'`, the function is called once without retries.
     """
-    result = None
+    if rq_access_mode == 'single':
+        return await fn()
+
+    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: