From 46a8ef97573c476311ee238c1f1e648913371d8f Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 29 Jul 2026 02:16:52 +0000 Subject: [PATCH] fix(http): classify body-coded rate-limit exhaustion as transient_error Co-Authored-By: bot_apk --- .../error_handlers/http_response_filter.py | 9 +- .../sources/streams/http/http_client.py | 4 +- .../test_http_response_filter.py | 19 ++- .../sources/streams/http/test_http_client.py | 128 ++++++++++++++++-- 4 files changed, 142 insertions(+), 18 deletions(-) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py index 866d6b7d25..5802f2ec81 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/http_response_filter.py +++ b/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: diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index c1d0eabd67..625b48db72 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/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 ) diff --git a/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py b/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py index 87e522d4ab..b8b7172c7f 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/test_http_response_filter.py +++ b/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( diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 48d396cb63..3f7474f636 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/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"