fix(http): classify body-coded rate-limit exhaustion as transient_error

airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py

@@ -26,8 +26,9 @@ class HttpResponseFilter:
     """
     Filter to select a response based on its HTTP status code, error message or a predicate.
     If a response matches the filter, the response action, failure_type, and error message are returned as an ErrorResolution object.
-    For http_codes declared in the filter, the failure_type will default to `system_error`.
-    To override default failure_type use configured failure_type with ResponseAction.FAIL.
+    For http_codes declared in the filter, the failure_type will default to `system_error`, except for
+    `ResponseAction.RATE_LIMITED` which defaults to `transient_error`.
+    A configured failure_type always takes precedence over the default one.
 
     Attributes:
         action (Union[ResponseAction, str]): action to execute if a request matches
@@ -95,8 +96,10 @@ def matches(
                 error_message = self._create_error_message(response_or_exception)
             error_message = error_message or default_error_message
 
-            if self.failure_type and filter_action == ResponseAction.FAIL:
+            if self.failure_type:
                 failure_type = self.failure_type
+            elif filter_action == ResponseAction.RATE_LIMITED:
+                failure_type = FailureType.transient_error
             elif default_mapped_error_resolution:
                 failure_type = default_mapped_error_resolution.failure_type
             else:

airbyte_cdk/sources/streams/http/http_client.py

@@ -300,7 +300,9 @@ def _send_with_retry(
         except BaseBackoffException as e:
             self._logger.error("Retries exhausted with backoff exception.", exc_info=True)
 
-            is_rate_limited = (
+            # A rate limit can be signalled either by an HTTP 429 or, for APIs that encode errors in
+            # the response body, by an error resolution with a `RATE_LIMITED` action on any status code.
+            is_rate_limited = isinstance(e, RateLimitBackoffException) or (
                 isinstance(e.response, requests.Response)
                 and e.response.status_code == requests.codes.too_many_requests
             )

unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py

@@ -178,10 +178,10 @@
             {"status_code": 500},
             ErrorResolution(
                 response_action=ResponseAction.RETRY,
-                failure_type=FailureType.transient_error,
+                failure_type=FailureType.config_error,
                 error_message="rate limits",
             ),
-            id="test_http_code_matches_failure_type_config_error_action_retry_uses_default_failure_type",
+            id="test_configured_failure_type_is_honored_for_non_fail_action",
         ),
         pytest.param(
             ResponseAction.RATE_LIMITED,
@@ -198,6 +198,21 @@
             ),
             id="test_http_code_matches_response_action_rate_limited",
         ),
+        pytest.param(
+            ResponseAction.RATE_LIMITED,
+            None,
+            None,
+            "{{ response.code == 40100 }}",
+            "",
+            "vendor rate limit message",
+            {"status_code": 200, "json": {"code": 40100}},
+            ErrorResolution(
+                response_action=ResponseAction.RATE_LIMITED,
+                failure_type=FailureType.transient_error,
+                error_message="vendor rate limit message",
+            ),
+            id="test_body_coded_rate_limit_on_http_200_defaults_to_transient_error",
+        ),
     ],
 )
 def test_matches(

unit_tests/sources/streams/http/test_http_client.py

@@ -11,6 +11,10 @@
 from requests_cache import CachedRequest
 
 from airbyte_cdk.models import FailureType
+from airbyte_cdk.sources.declarative.requesters.error_handlers import (
+    DefaultErrorHandler,
+    HttpResponseFilter,
+)
 from airbyte_cdk.sources.streams.call_rate import CachedLimiterSession, LimiterSession
 from airbyte_cdk.sources.streams.http import HttpClient
 from airbyte_cdk.sources.streams.http.error_handlers import (
@@ -783,21 +787,81 @@ def test_send_request_respects_environment_variables():
 
 @pytest.mark.usefixtures("mock_sleep")
 @pytest.mark.parametrize(
-    "response_code, expected_failure_type, error_message, exception_class",
+    "response_code, resolution_failure_type, expected_failure_type, error_message, exception_class",
     [
-        (400, FailureType.system_error, "test error message", UserDefinedBackoffException),
-        (401, FailureType.config_error, "test error message", UserDefinedBackoffException),
-        (403, FailureType.transient_error, "test error message", UserDefinedBackoffException),
-        (400, FailureType.system_error, "test error message", DefaultBackoffException),
-        (401, FailureType.config_error, "test error message", DefaultBackoffException),
-        (403, FailureType.transient_error, "test error message", DefaultBackoffException),
-        (400, FailureType.system_error, "test error message", RateLimitBackoffException),
-        (401, FailureType.config_error, "test error message", RateLimitBackoffException),
-        (403, FailureType.transient_error, "test error message", RateLimitBackoffException),
+        (
+            400,
+            FailureType.system_error,
+            FailureType.system_error,
+            "test error message",
+            UserDefinedBackoffException,
+        ),
+        (
+            401,
+            FailureType.config_error,
+            FailureType.config_error,
+            "test error message",
+            UserDefinedBackoffException,
+        ),
+        (
+            403,
+            FailureType.transient_error,
+            FailureType.transient_error,
+            "test error message",
+            UserDefinedBackoffException,
+        ),
+        (
+            400,
+            FailureType.system_error,
+            FailureType.system_error,
+            "test error message",
+            DefaultBackoffException,
+        ),
+        (
+            401,
+            FailureType.config_error,
+            FailureType.config_error,
+            "test error message",
+            DefaultBackoffException,
+        ),
+        (
+            403,
+            FailureType.transient_error,
+            FailureType.transient_error,
+            "test error message",
+            DefaultBackoffException,
+        ),
+        # An exhausted rate limit is always transient, whatever failure type the resolution carries.
+        (
+            400,
+            FailureType.system_error,
+            FailureType.transient_error,
+            "test error message",
+            RateLimitBackoffException,
+        ),
+        (
+            401,
+            FailureType.config_error,
+            FailureType.transient_error,
+            "test error message",
+            RateLimitBackoffException,
+        ),
+        (
+            403,
+            FailureType.transient_error,
+            FailureType.transient_error,
+            "test error message",
+            RateLimitBackoffException,
+        ),
     ],
 )
 def test_send_with_retry_raises_airbyte_traced_exception_with_failure_type(
-    response_code, expected_failure_type, error_message, exception_class, requests_mock
+    response_code,
+    resolution_failure_type,
+    expected_failure_type,
+    error_message,
+    exception_class,
+    requests_mock,
 ):
     if exception_class == UserDefinedBackoffException:
 
@@ -815,7 +879,7 @@ def backoff_time(self, response_or_exception, attempt_count):
         response_action = ResponseAction.RETRY
 
     error_mapping = {
-        response_code: ErrorResolution(response_action, expected_failure_type, error_message),
+        response_code: ErrorResolution(response_action, resolution_failure_type, error_message),
     }
 
     http_client = HttpClient(
@@ -840,6 +904,46 @@ def backoff_time(self, response_or_exception, attempt_count):
     assert e.value.failure_type == expected_failure_type
 
 
+@pytest.mark.usefixtures("mock_sleep")
+def test_send_with_retry_body_coded_rate_limit_raises_transient_error(requests_mock):
+    """A vendor signalling a rate limit through the response body must be classified as transient."""
+    vendor_error_message = "App **** reaches the QPS limit 10, current QPS is 11."
+    error_handler = DefaultErrorHandler(
+        config={},
+        parameters={},
+        max_retries=1,
+        response_filters=[
+            HttpResponseFilter(
+                config={},
+                parameters={},
+                action=ResponseAction.RATE_LIMITED,
+                predicate="{{ response.get('code') == 40100 }}",
+                error_message=vendor_error_message,
+            )
+        ],
+    )
+    http_client = HttpClient(
+        name="test",
+        logger=MagicMock(spec=logging.Logger),
+        error_handler=error_handler,
+    )
+
+    requests_mock.register_uri(
+        "GET",
+        "https://airbyte.io/",
+        status_code=200,
+        json={"code": 40100, "message": vendor_error_message},
+        headers={},
+    )
+
+    with pytest.raises(AirbyteTracedException) as exception_info:
+        http_client.send_request(http_method="get", url="https://airbyte.io/", request_kwargs={})
+
+    assert exception_info.value.failure_type == FailureType.transient_error
+    assert exception_info.value.message == "API rate limit exceeded."
+    assert vendor_error_message in exception_info.value.internal_message
+
+
 class MockOAuthAuthenticator:
     def __init__(self):
         self.access_token = "old_token"