feat(declarative): Add backoff jitter support

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -490,6 +490,19 @@ definitions:
           - 30
           - 30.5
           - "{{ config['backoff_time'] }}"
+      jitter_range_in_seconds:
+        title: Jitter Range
+        description: Optional jitter range in seconds. When set, the backoff time is uniformly distributed between max(0, backoff_time_in_seconds - jitter_range_in_seconds) and backoff_time_in_seconds + jitter_range_in_seconds.
+        anyOf:
+          - type: number
+            title: Number of seconds
+          - type: string
+            title: Interpolated Value
+        interpolation_context:
+          - config
+        examples:
+          - 15
+          - "{{ config['backoff_jitter'] }}"
       $parameters:
         type: object
         additionalProperties: true
@@ -2033,6 +2046,19 @@ definitions:
           - 5
           - 5.5
           - "10"
+      jitter_range_in_seconds:
+        title: Jitter Range
+        description: Optional jitter range in seconds. When set, the backoff time is uniformly distributed between max(0, computed_backoff - jitter_range_in_seconds) and computed_backoff + jitter_range_in_seconds.
+        anyOf:
+          - type: number
+            title: Number of seconds
+          - type: string
+            title: Interpolated Value
+        interpolation_context:
+          - config
+        examples:
+          - 2
+          - "{{ config['backoff_jitter'] }}"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -104,6 +104,12 @@ class ConstantBackoffStrategy(BaseModel):
         examples=[30, 30.5, "{{ config['backoff_time'] }}"],
         title="Backoff Time",
     )
+    jitter_range_in_seconds: Optional[Union[float, str]] = Field(
+        None,
+        description="Optional jitter range in seconds. When set, the backoff time is uniformly distributed between max(0, backoff_time_in_seconds - jitter_range_in_seconds) and backoff_time_in_seconds + jitter_range_in_seconds.",
+        examples=[15, "{{ config['backoff_jitter'] }}"],
+        title="Jitter Range",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -512,6 +518,12 @@ class ExponentialBackoffStrategy(BaseModel):
         examples=[5, 5.5, "10"],
         title="Factor",
     )
+    jitter_range_in_seconds: Optional[Union[float, str]] = Field(
+        None,
+        description="Optional jitter range in seconds. When set, the backoff time is uniformly distributed between max(0, computed_backoff - jitter_range_in_seconds) and computed_backoff + jitter_range_in_seconds.",
+        examples=[2, "{{ config['backoff_jitter'] }}"],
+        title="Jitter Range",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 
@@ -3077,6 +3089,7 @@ class AsyncRetriever(BaseModel):
     failed_retry_wait_time_in_seconds: Optional[Union[int, str]] = Field(
         None,
         description="Time in seconds to wait before retrying a failed async job. Only applies to jobs that ran on the API side and reported a FAILED status (e.g. report generation failed due to a cooldown). Creation failures (HTTP errors when starting a job, such as 429s) and TIMED_OUT jobs are retried immediately and are not affected by this setting. When set, the orchestrator defers retry of real failed jobs until the wait time has elapsed, without blocking other jobs.",
+        ge=1,
     )
     download_target_requester: Optional[Union[HttpRequester, CustomRequester]] = Field(
         None,

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -1754,6 +1754,7 @@ def create_constant_backoff_strategy(
     ) -> ConstantBackoffStrategy:
         return ConstantBackoffStrategy(
             backoff_time_in_seconds=model.backoff_time_in_seconds,
+            jitter_range_in_seconds=model.jitter_range_in_seconds,
             config=config,
             parameters=model.parameters or {},
         )
@@ -2433,7 +2434,10 @@ def create_exponential_backoff_strategy(
         model: ExponentialBackoffStrategyModel, config: Config
     ) -> ExponentialBackoffStrategy:
         return ExponentialBackoffStrategy(
-            factor=model.factor or 5, parameters=model.parameters or {}, config=config
+            factor=model.factor or 5,
+            jitter_range_in_seconds=model.jitter_range_in_seconds,
+            parameters=model.parameters or {},
+            config=config,
         )
 
     @staticmethod

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

@@ -2,8 +2,9 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import random
 from dataclasses import InitVar, dataclass
-from typing import Any, Mapping, Optional, Union
+from typing import Any, Mapping, Optional, Union, cast
 
 import requests
 
@@ -24,22 +25,39 @@ class ConstantBackoffStrategy(BackoffStrategy):
     backoff_time_in_seconds: Union[float, InterpolatedString, str]
     parameters: InitVar[Mapping[str, Any]]
     config: Config
+    jitter_range_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        if not isinstance(self.backoff_time_in_seconds, InterpolatedString):
-            self.backoff_time_in_seconds = str(self.backoff_time_in_seconds)
-        if isinstance(self.backoff_time_in_seconds, float):
-            self.backoff_time_in_seconds = InterpolatedString.create(
-                str(self.backoff_time_in_seconds), parameters=parameters
-            )
-        else:
-            self.backoff_time_in_seconds = InterpolatedString.create(
-                self.backoff_time_in_seconds, parameters=parameters
+        self.backoff_time_in_seconds = self._as_interpolated_string(
+            self.backoff_time_in_seconds, parameters
+        )
+        if self.jitter_range_in_seconds is not None:
+            self.jitter_range_in_seconds = self._as_interpolated_string(
+                self.jitter_range_in_seconds, parameters
             )
 
+    @staticmethod
+    def _as_interpolated_string(
+        value: Union[float, InterpolatedString, str], parameters: Mapping[str, Any]
+    ) -> InterpolatedString:
+        if not isinstance(value, InterpolatedString):
+            value = str(value)
+        return InterpolatedString.create(value, parameters=parameters)
+
     def backoff_time(
         self,
         response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
         attempt_count: int,
     ) -> Optional[float]:
-        return self.backoff_time_in_seconds.eval(self.config)  # type: ignore # backoff_time_in_seconds is always cast to an interpolated string
+        backoff_time = float(
+            cast(InterpolatedString, self.backoff_time_in_seconds).eval(self.config)
+        )
+        jitter_range_in_seconds = self.jitter_range_in_seconds
+        if jitter_range_in_seconds is None:
+            return backoff_time
+
+        jitter_range = float(cast(InterpolatedString, jitter_range_in_seconds).eval(self.config))
+        if jitter_range < 0:
+            raise ValueError("jitter_range_in_seconds must be greater than or equal to 0")
+
+        return random.uniform(max(0, backoff_time - jitter_range), backoff_time + jitter_range)

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

@@ -2,8 +2,9 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
+import random
 from dataclasses import InitVar, dataclass
-from typing import Any, Mapping, Optional, Union
+from typing import Any, Mapping, Optional, Union, cast
 
 import requests
 
@@ -24,22 +25,39 @@ class ExponentialBackoffStrategy(BackoffStrategy):
     parameters: InitVar[Mapping[str, Any]]
     config: Config
     factor: Union[float, InterpolatedString, str] = 5
+    jitter_range_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
-        if not isinstance(self.factor, InterpolatedString):
-            self.factor = str(self.factor)
-        if isinstance(self.factor, float):
-            self._factor = InterpolatedString.create(str(self.factor), parameters=parameters)
-        else:
-            self._factor = InterpolatedString.create(self.factor, parameters=parameters)
+        self._factor = self._as_interpolated_string(self.factor, parameters)
+        if self.jitter_range_in_seconds is not None:
+            self.jitter_range_in_seconds = self._as_interpolated_string(
+                self.jitter_range_in_seconds, parameters
+            )
+
+    @staticmethod
+    def _as_interpolated_string(
+        value: Union[float, InterpolatedString, str], parameters: Mapping[str, Any]
+    ) -> InterpolatedString:
+        if not isinstance(value, InterpolatedString):
+            value = str(value)
+        return InterpolatedString.create(value, parameters=parameters)
 
     @property
     def _retry_factor(self) -> float:
-        return self._factor.eval(self.config)  # type: ignore # factor is always cast to an interpolated string
+        return float(self._factor.eval(self.config))
 
     def backoff_time(
         self,
         response_or_exception: Optional[Union[requests.Response, requests.RequestException]],
         attempt_count: int,
     ) -> Optional[float]:
-        return self._retry_factor * 2**attempt_count  # type: ignore # factor is always cast to an interpolated string
+        backoff_time = float(self._retry_factor * 2**attempt_count)
+        jitter_range_in_seconds = self.jitter_range_in_seconds
+        if jitter_range_in_seconds is None:
+            return backoff_time
+
+        jitter_range = float(cast(InterpolatedString, jitter_range_in_seconds).eval(self.config))
+        if jitter_range < 0:
+            raise ValueError("jitter_range_in_seconds must be greater than or equal to 0")
+
+        return random.uniform(max(0, backoff_time - jitter_range), backoff_time + jitter_range)

unit_tests/sources/declarative/parsers/test_model_to_component_factory.py

@@ -1754,6 +1754,64 @@ def test_create_requester(test_name, error_handler, expected_backoff_strategy_ty
     )
 
 
+@pytest.mark.parametrize(
+    "backoff_strategy_yaml, expected_backoff_strategy_type",
+    [
+        pytest.param(
+            """
+  error_handler:
+    backoff_strategies:
+      - type: "ConstantBackoffStrategy"
+        backoff_time_in_seconds: 60
+        jitter_range_in_seconds: 15
+            """,
+            ConstantBackoffStrategy,
+            id="constant_backoff_strategy",
+        ),
+        pytest.param(
+            """
+  error_handler:
+    backoff_strategies:
+      - type: "ExponentialBackoffStrategy"
+        factor: 5
+        jitter_range_in_seconds: "{{ config['backoff_jitter'] }}"
+            """,
+            ExponentialBackoffStrategy,
+            id="exponential_backoff_strategy",
+        ),
+    ],
+)
+def test_create_requester_with_backoff_jitter(
+    backoff_strategy_yaml, expected_backoff_strategy_type
+):
+    content = f"""
+requester:
+  type: HttpRequester
+  path: "/v3/marketing/lists"
+  url_base: "https://api.sendgrid.com"
+  {backoff_strategy_yaml}
+    """
+    parsed_manifest = YamlDeclarativeSource._parse(content)
+    resolved_manifest = resolver.preprocess_manifest(parsed_manifest)
+    requester_manifest = transformer.propagate_types_and_parameters(
+        "", resolved_manifest["requester"], {}
+    )
+
+    requester = factory.create_component(
+        model_type=HttpRequesterModel,
+        component_definition=requester_manifest,
+        config={**input_config, "backoff_jitter": 15},
+        name="name",
+        decoder=None,
+    )
+
+    assert isinstance(requester.error_handler, DefaultErrorHandler)
+    assert len(requester.error_handler.backoff_strategies) == 1
+    backoff_strategy = requester.error_handler.backoff_strategies[0]
+    assert isinstance(backoff_strategy, expected_backoff_strategy_type)
+    assert backoff_strategy.jitter_range_in_seconds.eval({"backoff_jitter": 15}) == 15
+
+
 def test_create_request_with_legacy_session_authenticator():
     content = """
 requester:

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

@@ -16,29 +16,86 @@
 
 
 @pytest.mark.parametrize(
-    "test_name, attempt_count, backofftime, expected_backoff_time",
+    "test_name, attempt_count, backofftime, jitter_range, expected_backoff_time",
     [
-        ("test_constant_backoff_first_attempt", 1, BACKOFF_TIME, BACKOFF_TIME),
-        ("test_constant_backoff_first_attempt_float", 1, 6.7, 6.7),
-        ("test_constant_backoff_attempt_round_float", 1.0, 6.7, 6.7),
-        ("test_constant_backoff_attempt_round_float", 1.5, 6.7, 6.7),
-        ("test_constant_backoff_first_attempt_round_float", 1, 10.0, BACKOFF_TIME),
-        ("test_constant_backoff_second_attempt_round_float", 2, 10.0, BACKOFF_TIME),
+        ("test_constant_backoff_first_attempt", 1, BACKOFF_TIME, None, BACKOFF_TIME),
+        ("test_constant_backoff_first_attempt_float", 1, 6.7, None, 6.7),
+        ("test_constant_backoff_attempt_round_float", 1.0, 6.7, None, 6.7),
+        ("test_constant_backoff_attempt_round_float", 1.5, 6.7, None, 6.7),
+        ("test_constant_backoff_first_attempt_round_float", 1, 10.0, None, BACKOFF_TIME),
+        ("test_constant_backoff_second_attempt_round_float", 2, 10.0, None, BACKOFF_TIME),
+        ("test_constant_backoff_zero_jitter", 2, BACKOFF_TIME, 0, BACKOFF_TIME),
         (
             "test_constant_backoff_from_parameters",
             1,
             "{{ parameters['backoff'] }}",
+            None,
             PARAMETERS_BACKOFF_TIME,
         ),
-        ("test_constant_backoff_from_config", 1, "{{ config['backoff'] }}", CONFIG_BACKOFF_TIME),
+        (
+            "test_constant_backoff_from_config",
+            1,
+            "{{ config['backoff'] }}",
+            None,
+            CONFIG_BACKOFF_TIME,
+        ),
+        (
+            "test_constant_jitter_from_config",
+            1,
+            BACKOFF_TIME,
+            "{{ config['jitter'] }}",
+            BACKOFF_TIME,
+        ),
     ],
 )
-def test_constant_backoff(test_name, attempt_count, backofftime, expected_backoff_time):
+def test_constant_backoff(
+    test_name, attempt_count, backofftime, jitter_range, expected_backoff_time
+):
     response_mock = MagicMock()
     backoff_strategy = ConstantBackoffStrategy(
         parameters={"backoff": PARAMETERS_BACKOFF_TIME},
         backoff_time_in_seconds=backofftime,
-        config={"backoff": CONFIG_BACKOFF_TIME},
+        jitter_range_in_seconds=jitter_range,
+        config={"backoff": CONFIG_BACKOFF_TIME, "jitter": 0},
     )
     backoff = backoff_strategy.backoff_time(response_mock, attempt_count=attempt_count)
     assert backoff == expected_backoff_time
+
+
+@pytest.mark.parametrize(
+    "backofftime, jitter_range, expected_lower_bound, expected_upper_bound",
+    [
+        pytest.param(60, 15, 45, 75, id="centered_jitter"),
+        pytest.param(10, 30, 0, 40, id="lower_bound_clamped_to_zero"),
+        pytest.param(0, 5, 0, 5, id="zero_base"),
+    ],
+)
+def test_constant_backoff_with_jitter_bounds(
+    backofftime, jitter_range, expected_lower_bound, expected_upper_bound
+):
+    response_mock = MagicMock()
+    backoff_strategy = ConstantBackoffStrategy(
+        parameters={},
+        backoff_time_in_seconds=backofftime,
+        jitter_range_in_seconds=jitter_range,
+        config={},
+    )
+
+    backoff_times = [
+        backoff_strategy.backoff_time(response_mock, attempt_count=1) for _ in range(2000)
+    ]
+
+    assert all(expected_lower_bound <= backoff <= expected_upper_bound for backoff in backoff_times)
+
+
+def test_constant_backoff_with_negative_jitter_raises_error():
+    response_mock = MagicMock()
+    backoff_strategy = ConstantBackoffStrategy(
+        parameters={},
+        backoff_time_in_seconds=BACKOFF_TIME,
+        jitter_range_in_seconds=-5,
+        config={},
+    )
+
+    with pytest.raises(ValueError, match="jitter_range_in_seconds"):
+        backoff_strategy.backoff_time(response_mock, attempt_count=1)