fix: correct backoff timing, wire retry_endlessly, add backoff cap

- Fix exponential backoff sequence: 2^(attempt-1) to match old backoff.expo
  (1, 2, 4, 8s) instead of 2^attempt (2, 4, 8, 16s)
- Wire retry_endlessly into the retry loop so rate-limited requests
  without exit_on_rate_limit genuinely retry past the budget
- Guard retry_endlessly with user_defined_backoff_time is None to
  preserve the old mutually-exclusive branching (custom backoff always
  bounded, endless only for rate limits without a strategy)
- Add _MAX_BACKOFF_SECONDS (300s) cap on exponential backoff
- Split test_backoff_strategy_endless into two tests that verify
  bounded vs endless behavior independently

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: sophiecuiy

airbyte_cdk/sources/streams/http/http_client.py

@@ -251,17 +251,20 @@ def _max_time(self) -> int:
             else self._DEFAULT_MAX_TIME
         )
 
+    _MAX_BACKOFF_SECONDS: float = 300  # 5-minute ceiling for exponential backoff
+
     def _compute_backoff(self, exc: RetryRequestException, attempt: int) -> float:
         """Compute the backoff duration in seconds for a retry attempt.
 
-        If the exception carries a user-defined `backoff_time`, that value plus
+        If the exception carries a user-defined ``backoff_time``, that value plus
         one second is returned (preserving the legacy +1 s behaviour).  Otherwise
-        an exponential back-off with base 2 and no jitter is used:
-        ``2 ** attempt`` seconds.
+        an exponential back-off of ``2 ** (attempt - 1)`` seconds is used (matching
+        the previous ``backoff.expo`` with base=2, factor=1), capped at
+        ``_MAX_BACKOFF_SECONDS``.
         """
         if exc.backoff_time is not None:
             return exc.backoff_time + 1  # extra second to cover fractions
-        return float(2**attempt)
+        return min(float(2 ** (attempt - 1)), self._MAX_BACKOFF_SECONDS)
 
     def _send_with_retry(
         self,
@@ -293,12 +296,11 @@ def _send_with_retry(
                 attempt += 1
                 elapsed = time.monotonic() - start_time
 
-                # Determine whether we have exhausted retries.
-                budget_exhausted = False
-                if attempt >= max_tries:
-                    budget_exhausted = True
-                elif elapsed >= max_time:
-                    budget_exhausted = True
+                # Rate-limited requests retry indefinitely unless exit_on_rate_limit was set.
+                # All other retryable errors are bounded by max_tries / max_time.
+                budget_exhausted = not exc.retry_endlessly and (
+                    attempt >= max_tries or elapsed >= max_time
+                )
 
                 if budget_exhausted:
                     self._logger.error("Retries exhausted with backoff exception.", exc_info=True)
@@ -532,9 +534,13 @@ def _handle_error_resolution(
                 or f"Request to {request.url} failed with failure type {error_resolution.failure_type}, response action {error_resolution.response_action}."
             )
 
+            # Only retry endlessly when rate-limited AND no custom backoff strategy matched.
+            # When a strategy provides a specific backoff_time, retries are always bounded
+            # by max_tries/max_time (matching the old mutually-exclusive branching).
             retry_endlessly = (
                 error_resolution.response_action == ResponseAction.RATE_LIMITED
                 and not exit_on_rate_limit
+                and user_defined_backoff_time is None
             )
 
             raise RetryRequestException(

unit_tests/sources/streams/http/test_http_client.py

@@ -689,14 +689,9 @@ def backoff_time(self, *args, **kwargs):
         assert len(trace_messages) == mocked_send.call_count
 
 
-@pytest.mark.parametrize(
-    "exit_on_rate_limit, expected_call_count, expected_error",
-    [[True, 6, RetryRequestException], [False, 6, RetryRequestException]],
-)
 @pytest.mark.usefixtures("mock_sleep")
-def test_backoff_strategy_endless(
-    exit_on_rate_limit: bool, expected_call_count: int, expected_error: Exception
-):
+def test_backoff_strategy_rate_limited_with_exit_on_rate_limit():
+    """When exit_on_rate_limit=True, 429 responses exhaust max_tries then raise."""
     http_client = HttpClient(
         name="test", logger=MagicMock(), error_handler=HttpStatusErrorHandler(logger=MagicMock())
     )
@@ -705,18 +700,47 @@ def test_backoff_strategy_endless(
     mocked_response.status_code = 429
     mocked_response.headers = {}
     mocked_response.ok = False
-    session_send = MagicMock(spec=requests.Session.send)
-    session_send.return_value = mocked_response
 
     with patch.object(requests.Session, "send", return_value=mocked_response) as mocked_send:
-        with pytest.raises(AirbyteTracedException) as e:
+        with pytest.raises(AirbyteTracedException):
             http_client.send_request(
                 http_method="get",
                 url="https://test_base_url.com/v1/endpoint",
                 request_kwargs={},
-                exit_on_rate_limit=exit_on_rate_limit,
+                exit_on_rate_limit=True,
             )
-        assert mocked_send.call_count == expected_call_count
+        assert mocked_send.call_count == 6  # 1 initial + 5 retries
+
+
+@pytest.mark.usefixtures("mock_sleep")
+def test_backoff_strategy_rate_limited_retries_endlessly():
+    """When exit_on_rate_limit=False, 429 responses retry past max_tries until success."""
+    http_client = HttpClient(
+        name="test", logger=MagicMock(), error_handler=HttpStatusErrorHandler(logger=MagicMock())
+    )
+
+    rate_limited_response = MagicMock(spec=requests.Response)
+    rate_limited_response.status_code = 429
+    rate_limited_response.headers = {}
+    rate_limited_response.ok = False
+
+    success_response = MagicMock(spec=requests.Response)
+    success_response.status_code = 200
+    success_response.headers = {}
+    success_response.ok = True
+
+    # Fail 10 times (well past max_tries=6), then succeed
+    side_effects = [rate_limited_response] * 10 + [success_response]
+
+    with patch.object(requests.Session, "send", side_effect=side_effects) as mocked_send:
+        _, response = http_client.send_request(
+            http_method="get",
+            url="https://test_base_url.com/v1/endpoint",
+            request_kwargs={},
+            exit_on_rate_limit=False,
+        )
+        assert response.status_code == 200
+        assert mocked_send.call_count == 11  # 10 rate-limited + 1 success
 
 
 def test_given_different_headers_then_response_is_not_cached(requests_mock):
@@ -834,7 +858,12 @@ def backoff_time(self, response_or_exception, attempt_count):
     )
 
     with pytest.raises(AirbyteTracedException) as e:
-        http_client.send_request(http_method="get", url="https://airbyte.io/", request_kwargs={})
+        http_client.send_request(
+            http_method="get",
+            url="https://airbyte.io/",
+            request_kwargs={},
+            exit_on_rate_limit=True,  # ensure rate-limited retries are bounded so the test terminates
+        )
     assert e.value.failure_type == expected_failure_type