feat(low-code): response-aware ConstantBackoffStrategy with fall-through on falsy renders

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -698,7 +698,7 @@ definitions:
         additionalProperties: true
   ConstantBackoffStrategy:
     title: Constant Backoff
-    description: Backoff strategy with a constant backoff interval.
+    description: Backoff strategy with a constant backoff interval. The backoff time can be computed from the response via interpolation. If the rendered value is empty, None, 0, or non-numeric, this strategy is skipped (no jitter is applied) so the next entry in backoff_strategies is used.
     type: object
     required:
       - type
@@ -709,7 +709,7 @@ definitions:
         enum: [ConstantBackoffStrategy]
       backoff_time_in_seconds:
         title: Backoff Time
-        description: Backoff time in seconds.
+        description: Backoff time in seconds. Can be computed from the response via interpolation; if the rendered value is empty, None, 0, or non-numeric, this strategy is skipped and the next backoff strategy is used.
         anyOf:
           - type: number
             title: Number of seconds
@@ -718,10 +718,16 @@ definitions:
             title: Interpolated Value
         interpolation_context:
           - config
+          - response
+          - headers
+          - attempt_count
         examples:
           - 30
           - 30.5
           - "{{ config['backoff_time'] }}"
+          - "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}"
+          - "{{ (1 / (headers.get('x-amzn-RateLimit-Limit') | float)) if (headers.get('x-amzn-RateLimit-Limit') | float) > 0 else 60 }}"
+          - "{{ (response.get('message', '') | regex_search('(?i)Retry after\\s+(\\d+)\\s+seconds')) or min(2 ** attempt_count, 120) if response is mapping else min(2 ** attempt_count, 120) }}"
       jitter_range_in_seconds:
         title: Jitter Range
         description: Optional additive jitter range in seconds. When set, the backoff time is uniformly distributed between backoff_time_in_seconds and backoff_time_in_seconds + (jitter_range_in_seconds * 2), so jitter only increases the base backoff.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -136,8 +136,15 @@ class ConstantBackoffStrategy(BaseModel):
     type: Literal["ConstantBackoffStrategy"]
     backoff_time_in_seconds: Union[float, str] = Field(
         ...,
-        description="Backoff time in seconds.",
-        examples=[30, 30.5, "{{ config['backoff_time'] }}"],
+        description="Backoff time in seconds. Can be computed from the response via interpolation; if the rendered value is empty, None, 0, or non-numeric, this strategy is skipped and the next backoff strategy is used.",
+        examples=[
+            30,
+            30.5,
+            "{{ config['backoff_time'] }}",
+            "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}",
+            "{{ (1 / (headers.get('x-amzn-RateLimit-Limit') | float)) if (headers.get('x-amzn-RateLimit-Limit') | float) > 0 else 60 }}",
+            "{{ (response.get('message', '') | regex_search('(?i)Retry after\\s+(\\d+)\\s+seconds')) or min(2 ** attempt_count, 120) if response is mapping else min(2 ** attempt_count, 120) }}",
+        ],
         title="Backoff Time",
     )
     jitter_range_in_seconds: Optional[float] = Field(

airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/constant_backoff_strategy.py

@@ -2,21 +2,40 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import logging
 import random
 from dataclasses import InitVar, dataclass
 from typing import Any, Mapping, Optional, Union, cast
 
 import requests
+from jinja2.exceptions import UndefinedError
 
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.streams.http.error_handlers import BackoffStrategy
 from airbyte_cdk.sources.types import Config
 
+logger = logging.getLogger("airbyte")
+
 
 @dataclass
 class ConstantBackoffStrategy(BackoffStrategy):
     """
-    Backoff strategy with a constant backoff interval
+    Backoff strategy with a constant backoff interval.
+
+    The ``backoff_time_in_seconds`` field is evaluated on every call with the
+    following interpolation context, in addition to ``config``:
+
+    * ``response``: the parsed JSON body of the response (``{}`` when the body is
+      not JSON or when the call was made with a request exception or ``None``).
+    * ``headers``: the response headers (``{}`` on a request exception).
+    * ``attempt_count``: the number of the current retry attempt.
+
+    If the expression cannot be evaluated against the response (for example an
+    attribute access on a non-object JSON body), the rendered value is falsy
+    (``None``, ``""``, ``0``), evaluates to ``0`` seconds, or cannot be
+    converted to a number, this strategy returns ``None`` (no jitter is
+    applied) so that the next entry in ``backoff_strategies`` or the default
+    backoff is used.
 
     Attributes:
         backoff_time_in_seconds (float): time to backoff before retrying a retryable request.
@@ -44,10 +63,44 @@ def backoff_time(
         response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
         attempt_count: int,
     ) -> Optional[float]:
-        backoff_time = float(
-            cast(InterpolatedString, self.backoff_time_in_seconds).eval(self.config)
-        )
+        response: Any
+        headers: Mapping[str, Any]
+        if isinstance(response_or_exception, requests.Response):
+            response = self._safe_response_json(response_or_exception)
+            headers = response_or_exception.headers
+        else:
+            response = {}
+            headers = {}
+
+        try:
+            rendered = cast(InterpolatedString, self.backoff_time_in_seconds).eval(
+                self.config,
+                response=response,
+                headers=headers,
+                attempt_count=attempt_count,
+            )
+        except UndefinedError:
+            return None
+        if not rendered:
+            return None
+        try:
+            backoff_time = float(rendered)
+        except (TypeError, ValueError):
+            logger.warning(
+                f"ConstantBackoffStrategy backoff_time_in_seconds rendered to a non-numeric value {rendered!r}; "
+                "skipping this strategy so the next backoff strategy or the default backoff is used."
+            )
+            return None
+        if backoff_time == 0:
+            return None
         if self.jitter_range_in_seconds is None:
             return backoff_time
 
         return random.uniform(backoff_time, backoff_time + (self.jitter_range_in_seconds * 2))
+
+    @staticmethod
+    def _safe_response_json(response: requests.Response) -> Any:
+        try:
+            return response.json()
+        except requests.exceptions.JSONDecodeError:
+            return {}

unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_constant_backoff.py

@@ -2,10 +2,18 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import json
 from unittest.mock import MagicMock
 
 import pytest
+import requests
 
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    ConstantBackoffStrategy as ConstantBackoffStrategyModel,
+)
+from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import (
+    ModelToComponentFactory,
+)
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.constant_backoff_strategy import (
     ConstantBackoffStrategy,
 )
@@ -14,6 +22,34 @@
 PARAMETERS_BACKOFF_TIME = 20
 CONFIG_BACKOFF_TIME = 30
 
+GSC_BACKOFF_EXPRESSION = "{{ 900 if 'load quota exceeded' in ((response.get('error') or {}).get('message') or '') | lower else 0 }}"
+AMAZON_BACKOFF_EXPRESSION = "{{ (1 / (headers.get('x-amzn-RateLimit-Limit') | float)) if (headers.get('x-amzn-RateLimit-Limit') | float) > 0 else 60 }}"
+PINTEREST_BACKOFF_EXPRESSION = "{{ (response.get('message', '') | regex_search('(?i)Retry after\\s+(\\d+)\\s+seconds')) or min(2 ** attempt_count, 120) if response is mapping else min(2 ** attempt_count, 120) }}"
+LINKEDIN_BACKOFF_EXPRESSION = "{{ 330 if 'data request limit has been exceeded' in (response.get('message') or '') | lower and '45 million metric values' in (response.get('message') or '') | lower else 0 }}"
+
+
+def _make_response(status_code=429, json_body=None, raw_body=None, headers=None):
+    response = requests.Response()
+    response.status_code = status_code
+    response.url = "https://airbyte.io"
+    response.headers.update(headers or {})
+    if raw_body is not None:
+        response._content = raw_body
+    elif json_body is not None:
+        response._content = json.dumps(json_body).encode("utf-8")
+    else:
+        response._content = b"{}"
+    return response
+
+
+def _strategy(backoff_time_in_seconds, jitter_range=None, parameters=None, config=None):
+    return ConstantBackoffStrategy(
+        parameters=parameters or {},
+        backoff_time_in_seconds=backoff_time_in_seconds,
+        jitter_range_in_seconds=jitter_range,
+        config=config or {},
+    )
+
 
 @pytest.mark.parametrize(
     "test_name, attempt_count, backofftime, jitter_range, expected_backoff_time",
@@ -60,7 +96,6 @@ def test_constant_backoff(
     [
         pytest.param(60, 15, 60, 90, id="base_backoff_floor"),
         pytest.param(10, 30, 10, 70, id="large_jitter"),
-        pytest.param(0, 5, 0, 10, id="zero_base"),
     ],
 )
 def test_constant_backoff_with_jitter_bounds(
@@ -80,3 +115,127 @@ def test_constant_backoff_with_jitter_bounds(
 
     assert all(expected_lower_bound <= backoff <= expected_upper_bound for backoff in backoff_times)
     assert len(set(backoff_times)) > 1
+
+
+@pytest.mark.parametrize(
+    "json_body, expected_backoff_time",
+    [
+        pytest.param(
+            {"error": {"message": "Search Analytics load quota exceeded."}},
+            900,
+            id="load_quota_exceeded",
+        ),
+        pytest.param(
+            {"error": {"message": "Search Analytics QPS quota exceeded."}},
+            None,
+            id="qps_quota_exceeded",
+        ),
+    ],
+)
+def test_constant_backoff_response_body(json_body, expected_backoff_time):
+    backoff_strategy = _strategy(GSC_BACKOFF_EXPRESSION)
+    response = _make_response(json_body=json_body)
+    assert backoff_strategy.backoff_time(response, attempt_count=1) == expected_backoff_time
+
+
+@pytest.mark.parametrize(
+    "headers, expected_backoff_time",
+    [
+        pytest.param({"x-amzn-RateLimit-Limit": "0.5"}, 2.0, id="low_rate_limit"),
+        pytest.param({"x-amzn-RateLimit-Limit": "2.0"}, 0.5, id="higher_rate_limit"),
+        pytest.param({}, 60, id="missing_rate_limit_header"),
+        pytest.param({"x-amzn-RateLimit-Limit": "abc"}, 60, id="garbage_rate_limit_header"),
+    ],
+)
+def test_constant_backoff_response_headers(headers, expected_backoff_time):
+    backoff_strategy = _strategy(AMAZON_BACKOFF_EXPRESSION)
+    response = _make_response(headers=headers)
+    assert backoff_strategy.backoff_time(response, attempt_count=1) == expected_backoff_time
+
+
+@pytest.mark.parametrize(
+    "json_body, raw_body, attempt_count, expected_backoff_time",
+    [
+        pytest.param(
+            {"message": "Rate limited. Retry after 30 seconds"},
+            None,
+            1,
+            30,
+            id="retry_after_in_body",
+        ),
+        pytest.param({"message": "rate limited"}, None, 2, 4, id="exponential_fallback"),
+        pytest.param({"message": "rate limited"}, None, 10, 120, id="exponential_fallback_capped"),
+        pytest.param([1, 2, 3], None, 2, 4, id="non_dict_json_body"),
+        pytest.param(None, b"not json at all", 3, 8, id="non_json_body"),
+    ],
+)
+def test_constant_backoff_response_body_and_attempt_count(
+    json_body, raw_body, attempt_count, expected_backoff_time
+):
+    backoff_strategy = _strategy(PINTEREST_BACKOFF_EXPRESSION)
+    response = _make_response(json_body=json_body, raw_body=raw_body)
+    assert (
+        backoff_strategy.backoff_time(response, attempt_count=attempt_count)
+        == expected_backoff_time
+    )
+
+
+def test_constant_backoff_unguarded_get_on_list_body_falls_through():
+    backoff_strategy = _strategy("{{ response.get('message') }}")
+    response = _make_response(json_body=[1, 2, 3])
+    assert backoff_strategy.backoff_time(response, attempt_count=1) is None
+
+
+@pytest.mark.parametrize("jitter_range", [None, 15])
+@pytest.mark.parametrize(
+    "backofftime",
+    [
+        pytest.param("{{ '' }}", id="empty_string_render"),
+        pytest.param("{{ None }}", id="none_render"),
+        pytest.param("{{ 0 }}", id="zero_render"),
+        pytest.param(0, id="plain_zero"),
+    ],
+)
+def test_constant_backoff_falsy_render_falls_through(backofftime, jitter_range):
+    backoff_strategy = _strategy(backofftime, jitter_range=jitter_range)
+    response = _make_response()
+    assert backoff_strategy.backoff_time(response, attempt_count=1) is None
+
+
+def test_constant_backoff_non_numeric_render_falls_through(caplog):
+    backoff_strategy = _strategy("{{ 'abc' }}")
+    response = _make_response()
+    assert backoff_strategy.backoff_time(response, attempt_count=1) is None
+
+
+@pytest.mark.parametrize(
+    "response_or_exception",
+    [
+        pytest.param(requests.exceptions.ConnectionError(), id="request_exception"),
+        pytest.param(None, id="no_response"),
+    ],
+)
+def test_constant_backoff_on_exception_or_none(response_or_exception):
+    backoff_strategy = _strategy("{{ headers.get('x') or response.get('y') or 7 }}")
+    assert backoff_strategy.backoff_time(response_or_exception, attempt_count=1) == 7
+
+
+def test_constant_backoff_from_factory():
+    manifest = {
+        "type": "ConstantBackoffStrategy",
+        "backoff_time_in_seconds": LINKEDIN_BACKOFF_EXPRESSION,
+    }
+    backoff_strategy = ModelToComponentFactory().create_component(
+        ConstantBackoffStrategyModel, manifest, config={}
+    )
+    assert isinstance(backoff_strategy, ConstantBackoffStrategy)
+
+    throttled_response = _make_response(
+        json_body={
+            "message": "Data request limit has been exceeded. Limit of 45 million metric values per day."
+        }
+    )
+    assert backoff_strategy.backoff_time(throttled_response, attempt_count=1) == 330
+
+    other_response = _make_response(json_body={"message": "some other error"})
+    assert backoff_strategy.backoff_time(other_response, attempt_count=1) is None