Address PR review feedback

- Extract duplicate exponential backoff calculation into helper function
- Add upper bound (max_total_attempts) to prevent infinite retry loops with Retry-After
- Improves code maintainability and prevents edge case of continuous Retry-After responses

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

Author: MichaelGHSeg

segment/analytics/consumer.py

@@ -140,9 +140,22 @@ def should_use_retry_after(status):
             """Check if status code should respect Retry-After header"""
             return status in (408, 429, 503)
 
+        def calculate_backoff_delay(attempt):
+            """
+            Calculate exponential backoff delay with jitter.
+            First retry is immediate, then 0.5s, 1s, 2s, 4s, etc.
+            """
+            if attempt == 1:
+                return 0  # First retry is immediate
+            base_delay = 0.5 * (2 ** (attempt - 2))
+            jitter = random.uniform(0, 0.1 * base_delay)
+            return min(base_delay + jitter, 60)  # Cap at 60 seconds
+
         total_attempts = 0
         backoff_attempts = 0
         max_backoff_attempts = self.retries + 1
+        # Prevent infinite retry loops even with Retry-After
+        max_total_attempts = max_backoff_attempts * 10
 
         while True:
             try:
@@ -168,6 +181,13 @@ def should_use_retry_after(status):
             except APIError as e:
                 total_attempts += 1
 
+                # Prevent infinite retry loops
+                if total_attempts >= max_total_attempts:
+                    self.log.error(
+                        f"Maximum total attempts ({max_total_attempts}) reached after {total_attempts} attempts. Final error: {e}"
+                    )
+                    raise
+
                 # Check if we should use Retry-After header
                 if should_use_retry_after(e.status) and e.response:
                     retry_after = parse_retry_after(e.response)
@@ -194,9 +214,7 @@ def should_use_retry_after(status):
                     raise
 
                 # Calculate exponential backoff delay with jitter
-                base_delay = 0.5 * (2 ** (backoff_attempts - 1))
-                jitter = random.uniform(0, 0.1 * base_delay)
-                delay = min(base_delay + jitter, 60)  # Cap at 60 seconds
+                delay = calculate_backoff_delay(backoff_attempts)
 
                 self.log.debug(
                     f"Retry attempt {backoff_attempts}/{self.retries} (total attempts: {total_attempts}) "
@@ -209,16 +227,21 @@ def should_use_retry_after(status):
                 total_attempts += 1
                 backoff_attempts += 1
 
+                # Prevent infinite retry loops
+                if total_attempts >= max_total_attempts:
+                    self.log.error(
+                        f"Maximum total attempts ({max_total_attempts}) reached after {total_attempts} attempts. Final error: {e}"
+                    )
+                    raise
+
                 if backoff_attempts >= max_backoff_attempts:
                     self.log.error(
                         f"All {self.retries} retries exhausted after {total_attempts} total attempts. Final error: {e}"
                     )
                     raise
 
                 # Calculate exponential backoff delay with jitter
-                base_delay = 0.5 * (2 ** (backoff_attempts - 1))
-                jitter = random.uniform(0, 0.1 * base_delay)
-                delay = min(base_delay + jitter, 60)  # Cap at 60 seconds
+                delay = calculate_backoff_delay(backoff_attempts)
 
                 self.log.debug(
                     f"Network error retry {backoff_attempts}/{self.retries} (total attempts: {total_attempts}) "

segment/analytics/test/test_consumer.py

@@ -469,13 +469,12 @@ def mock_sleep(duration):
         self.assertEqual(len(sleep_durations), 3)
 
         # Delays should be increasing (exponential)
-        # First: ~0.5s, Second: ~1s, Third: ~2s (with jitter)
-        self.assertGreater(sleep_durations[0], 0.4)
-        self.assertLess(sleep_durations[0], 1.0)
-        self.assertGreater(sleep_durations[1], 0.9)
-        self.assertLess(sleep_durations[1], 2.0)
-        self.assertGreater(sleep_durations[2], 1.8)
-        self.assertLess(sleep_durations[2], 4.0)
+        # First: 0s (immediate), Second: ~0.5s, Third: ~1s (with jitter)
+        self.assertEqual(sleep_durations[0], 0)  # First retry is immediate
+        self.assertGreater(sleep_durations[1], 0.4)
+        self.assertLess(sleep_durations[1], 0.6)
+        self.assertGreater(sleep_durations[2], 0.9)
+        self.assertLess(sleep_durations[2], 1.2)
 
     def test_fatal_error_not_retried(self):
         """Test that FatalError is not retried"""
@@ -572,11 +571,10 @@ def mock_sleep(duration):
         self.assertEqual(call_count, 2)
         self.assertEqual(retry_counts, [0, 1])
 
-        # Should use backoff delay (around 0.5s with jitter)
+        # First retry should be immediate (0s delay)
         self.assertIsNotNone(sleep_duration)
         if sleep_duration is not None:
-            self.assertGreater(sleep_duration, 0.4)
-            self.assertLess(sleep_duration, 1.0)
+            self.assertEqual(sleep_duration, 0)
 
     def test_408_without_retry_after_uses_backoff(self):
         """T10: 408 without Retry-After header uses backoff retry"""
@@ -613,11 +611,10 @@ def mock_sleep(duration):
         self.assertEqual(call_count, 2)
         self.assertEqual(retry_counts, [0, 1])
 
-        # Should use backoff delay
+        # First retry should be immediate (0s delay)
         self.assertIsNotNone(sleep_duration)
         if sleep_duration is not None:
-            self.assertGreater(sleep_duration, 0.4)
-            self.assertLess(sleep_duration, 1.0)
+            self.assertEqual(sleep_duration, 0)
 
     def test_network_error_retried_with_backoff(self):
         """T15: Network/IO error is retried with backoff"""
@@ -651,11 +648,10 @@ def mock_sleep(duration):
         self.assertEqual(call_count, 2)
         self.assertEqual(retry_counts, [0, 1])
 
-        # Should use backoff delay
+        # First retry should be immediate (0s delay)
         self.assertIsNotNone(sleep_duration)
         if sleep_duration is not None:
-            self.assertGreater(sleep_duration, 0.4)
-            self.assertLess(sleep_duration, 1.0)
+            self.assertEqual(sleep_duration, 0)
 
     def test_511_is_retryable(self):
         """T05: 511 status code is retryable (part of 5xx family, not in non-retryable list)"""