fix: Restore original for-loop logic in integration tests

Keep the original loop structure in test_metadata_tracking,
test_state_consistency, and test_persistence_across_operations,
wrapping every iteration with call_with_exp_backoff instead of
extracting the first iteration separately.

Also restore is_finished + fetch_next_request pattern in
test_concurrent_processing_simulation.

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

Author: vdusek

tests/integration/test_request_queue.py

@@ -420,18 +420,12 @@ async def test_metadata_tracking(
     assert handled_after_add == 0, f'handled_after_add={handled_after_add}'
 
     # Process some requests.
-    next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
-    handled = 0
-    if next_request:
-        await rq.mark_request_as_handled(next_request)
-        handled += 1
-    for _ in range(2):
-        next_request = await rq.fetch_next_request()
+    for _ in range(3):
+        next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
         if next_request:
             await rq.mark_request_as_handled(next_request)
-            handled += 1
 
-    Actor.log.info(f'Processed {handled} requests')
+    Actor.log.info('Processed 3 requests')
 
     # Check counts after processing
     final_total = await rq.get_total_count()
@@ -504,12 +498,8 @@ async def test_state_consistency(
     processed_requests = []
     reclaimed_requests = []
 
-    next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
-    if next_request:
-        await rq.mark_request_as_handled(next_request)
-        processed_requests.append(next_request)
-    for i in range(1, 5):
-        next_request = await rq.fetch_next_request()
+    for i in range(5):
+        next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
         if next_request:
             if i % 2 == 0:  # Process even indices
                 await rq.mark_request_as_handled(next_request)
@@ -685,12 +675,8 @@ async def test_persistence_across_operations(
 
     # Process some requests.
     processed_count = 0
-    next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
-    if next_request:
-        await rq.mark_request_as_handled(next_request)
-        processed_count += 1
-    for _ in range(4):
-        next_request = await rq.fetch_next_request()
+    for _ in range(5):
+        next_request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode=rq_access_mode)
         if next_request:
             await rq.mark_request_as_handled(next_request)
             processed_count += 1
@@ -1389,9 +1375,13 @@ async def worker() -> int:
             assert total_after_workers == 20
 
             remaining_count = 0
-            while next_request := await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode='shared'):
-                remaining_count += 1
-                await rq.mark_request_as_handled(next_request)
+            while not await call_with_exp_backoff(rq.is_finished, rq_access_mode='shared'):
+                request = await call_with_exp_backoff(rq.fetch_next_request, rq_access_mode='shared')
+                if request:
+                    remaining_count += 1
+                    await rq.mark_request_as_handled(request)
+                else:
+                    break
 
             final_handled = await rq.get_handled_count()
             final_total = await rq.get_total_count()