feat(low-code): cap the wait WaitUntilTimeFromHeader is willing to return

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -4641,10 +4641,15 @@ definitions:
           - "([-+]?\\d+)"
       max_waiting_time_in_seconds:
         title: Max Waiting Time in Seconds
-        description: Given the value extracted from the header is greater than this value, stop the stream.
-        type: number
+        description: Stop the stream instead of waiting, when the value extracted from the header is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.
+        anyOf:
+          - type: number
+          - type: string
+        interpolation_context:
+          - config
         examples:
           - 3600
+          - "{{ config['max_waiting_time'] * 60 }}"
       $parameters:
         type: object
         additionalProperties: true
@@ -4770,6 +4775,17 @@ definitions:
           - config
         examples:
           - "([-+]?\\d+)"
+      max_waiting_time_in_seconds:
+        title: Max Waiting Time in Seconds
+        description: Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins -- including for the fallback where the header is absent and `min_wait` supplies the wait on its own. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.
+        anyOf:
+          - type: number
+          - type: string
+        interpolation_context:
+          - config
+        examples:
+          - 3600
+          - "{{ config['max_waiting_time'] * 60 }}"
       $parameters:
         type: object
         additionalProperties: true

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -1434,10 +1434,10 @@ class WaitTimeFromHeader(BaseModel):
         examples=["([-+]?\\d+)"],
         title="Extraction Regex",
     )
-    max_waiting_time_in_seconds: Optional[float] = Field(
+    max_waiting_time_in_seconds: Optional[Union[float, str]] = Field(
         None,
-        description="Given the value extracted from the header is greater than this value, stop the stream.",
-        examples=[3600],
+        description="Stop the stream instead of waiting, when the value extracted from the header is greater than or equal to this value. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.",
+        examples=[3600, "{{ config['max_waiting_time'] * 60 }}"],
         title="Max Waiting Time in Seconds",
     )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
@@ -1463,6 +1463,12 @@ class WaitUntilTimeFromHeader(BaseModel):
         examples=["([-+]?\\d+)"],
         title="Extraction Regex",
     )
+    max_waiting_time_in_seconds: Optional[Union[float, str]] = Field(
+        None,
+        description="Stop the stream instead of waiting, when the wait this strategy computes is longer than this value. The comparison is against the computed wait rather than the raw header, since the header holds an absolute timestamp, and it is applied after `min_wait`, so a cap below the floor still wins -- including for the fallback where the header is absent and `min_wait` supplies the wait on its own. Can be a hardcoded number, or a string interpolated from the connector config so that the bound can be changed without a connector release. A value of 0 means never wait.",
+        examples=[3600, "{{ config['max_waiting_time'] * 60 }}"],
+        title="Max Waiting Time in Seconds",
+    )
     parameters: Optional[Dict[str, Any]] = Field(None, alias="$parameters")
 
 

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -4343,9 +4343,7 @@ def create_wait_time_from_header(
             parameters=model.parameters or {},
             config=config,
             regex=model.regex,
-            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds
-            if model.max_waiting_time_in_seconds is not None
-            else None,
+            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
         )
 
     @staticmethod
@@ -4358,6 +4356,7 @@ def create_wait_until_time_from_header(
             config=config,
             min_wait=model.min_wait,
             regex=model.regex,
+            max_waiting_time_in_seconds=model.max_waiting_time_in_seconds,
         )
 
     def get_message_repository(self) -> MessageRepository:

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

@@ -0,0 +1,83 @@
+#
+# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
+#
+
+import math
+from typing import Any, Mapping, Optional, Union
+
+from airbyte_cdk.models import FailureType
+from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
+from airbyte_cdk.sources.types import Config
+from airbyte_cdk.utils import AirbyteTracedException
+
+MAX_WAITING_TIME_FIELD = "max_waiting_time_in_seconds"
+
+
+def interpolated_max_waiting_time(
+    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]],
+    parameters: Mapping[str, Any],
+) -> Optional[InterpolatedString]:
+    """
+    Cast a `max_waiting_time_in_seconds` field to an InterpolatedString so that a hardcoded number
+    and a value interpolated from the config are resolved through the same path.
+
+    :param max_waiting_time_in_seconds: the value as declared on the backoff strategy
+    :param parameters: parameters to make available to the interpolation
+    :return: the value as an InterpolatedString, or None when no cap is configured
+    """
+    if max_waiting_time_in_seconds is None or isinstance(
+        max_waiting_time_in_seconds, InterpolatedString
+    ):
+        return max_waiting_time_in_seconds
+    return InterpolatedString.create(str(max_waiting_time_in_seconds), parameters=parameters)
+
+
+def evaluate_max_waiting_time(
+    max_waiting_time_in_seconds: Optional[InterpolatedString], config: Config
+) -> Optional[float]:
+    """
+    Resolve a `max_waiting_time_in_seconds` field to a number of seconds.
+
+    The `is None` check is deliberate: 0 is a meaningful cap -- "never wait" -- so a truthiness
+    check would silently disable it. Only an absent field means "no cap"; a field that is present
+    but resolves to nothing raises, rather than quietly leaving the wait unbounded.
+
+    A cap is only read while handling an error the requester is already going to retry, so an
+    interpolation that cannot be resolved would otherwise surface as an unhandled jinja
+    UndefinedError or ValueError in the middle of a sync that has been running fine. It is raised
+    as a system error rather than a config error because the field lives in the manifest: whether
+    it is a bad expression or a config key the manifest reads but the spec does not expose, the
+    connector is at fault and there is nothing for the user to correct.
+
+    :param max_waiting_time_in_seconds: the interpolated field, or None when no cap is configured
+    :param config: the connector config to interpolate against
+    :return: the cap in seconds, or None when no cap is configured
+    """
+    if max_waiting_time_in_seconds is None:
+        return None
+    try:
+        # A cap that resolves to nothing -- an empty or null config value -- is a failure rather
+        # than "no cap": silently dropping the bound restores the unbounded wait the field exists
+        # to prevent, and "no cap" is already spelled by leaving the field out of the manifest.
+        max_waiting_time = float(max_waiting_time_in_seconds.eval(config))
+        if not math.isfinite(max_waiting_time):
+            # NaN would be the one value that disables the cap without saying so: every
+            # comparison against it is False, so the wait this field exists to bound would run
+            # unbounded again. Infinity is rejected alongside it as the same kind of mistake.
+            raise ValueError(f"resolved to {max_waiting_time}, which is not a finite number")
+        return max_waiting_time
+    except AirbyteTracedException:
+        raise
+    except Exception as exception:
+        raise AirbyteTracedException(
+            internal_message=(
+                f"Failed to evaluate {MAX_WAITING_TIME_FIELD} "
+                f"{max_waiting_time_in_seconds.string!r}: {exception}"
+            ),
+            message=(
+                "The connector could not determine how long it is allowed to wait for a rate "
+                "limit to clear. This is a problem with the connector rather than with your "
+                "configuration."
+            ),
+            failure_type=FailureType.system_error,
+        ) from exception

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

@@ -13,6 +13,10 @@
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.header_helper import (
     get_numeric_value_from_header,
 )
+from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import (
+    evaluate_max_waiting_time,
+    interpolated_max_waiting_time,
+)
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategy import (
     BackoffStrategy,
 )
@@ -28,20 +32,24 @@ class WaitTimeFromHeaderBackoffStrategy(BackoffStrategy):
     Attributes:
         header (str): header to read wait time from
         regex (Optional[str]): optional regex to apply on the header to extract its value
-        max_waiting_time_in_seconds: (Optional[float]): given the value extracted from the header is greater than this value, stop the stream
+        max_waiting_time_in_seconds (Optional[Union[float, InterpolatedString, str]]): stop the stream
+            rather than wait longer than this
     """
 
     header: Union[InterpolatedString, str]
     parameters: InitVar[Mapping[str, Any]]
     config: Config
     regex: Optional[Union[InterpolatedString, str]] = None
-    max_waiting_time_in_seconds: Optional[float] = None
+    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.regex = (
             InterpolatedString.create(self.regex, parameters=parameters) if self.regex else None
         )
         self.header = InterpolatedString.create(self.header, parameters=parameters)
+        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
+            self.max_waiting_time_in_seconds, parameters
+        )
 
     def backoff_time(
         self,
@@ -57,14 +65,23 @@ def backoff_time(
         header_value = None
         if isinstance(response_or_exception, requests.Response):
             header_value = get_numeric_value_from_header(response_or_exception, header, regex)
-            if (
-                self.max_waiting_time_in_seconds
-                and header_value
-                and header_value >= self.max_waiting_time_in_seconds
-            ):
+            max_waiting_time = evaluate_max_waiting_time(
+                self._max_waiting_time_in_seconds, self.config
+            )
+            # `max_waiting_time is not None` rather than a truthiness check, so that 0 means
+            # "never wait" instead of silently disabling the cap. The comparison stays `>=`,
+            # which is what this cap has always done; `WaitUntilTimeFromHeader` stops at `>`,
+            # so a wait exactly equal to the cap is allowed there and refused here.
+            # `header_value` is checked for truthiness rather than `is not None` on purpose: a
+            # header of `0` asks for no wait at all, which no cap -- not even 0 -- should refuse.
+            if max_waiting_time is not None and header_value and header_value >= max_waiting_time:
                 raise AirbyteTracedException(
-                    internal_message=f"Rate limit wait time {header_value} is greater than max waiting time of {self.max_waiting_time_in_seconds} seconds. Stopping the stream...",
-                    message="The rate limit is greater than max waiting time has been reached.",
+                    internal_message=(
+                        f"Rate limit wait time {header_value}s is greater than or equal to the "
+                        f"maximum of {max_waiting_time}s this stream is allowed to wait. "
+                        f"Stopping the stream..."
+                    ),
+                    message="The rate limit wait time is longer than the connector is allowed to wait.",
                     failure_type=FailureType.transient_error,
                 )
         return header_value

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

@@ -2,22 +2,27 @@
 # Copyright (c) 2023 Airbyte, Inc., all rights reserved.
 #
 
-import numbers
 import re
 import time
 from dataclasses import InitVar, dataclass
 from typing import Any, Mapping, Optional, Union
 
 import requests
 
+from airbyte_cdk.models import FailureType
 from airbyte_cdk.sources.declarative.interpolation.interpolated_string import InterpolatedString
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.header_helper import (
     get_numeric_value_from_header,
 )
+from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies.max_waiting_time_helper import (
+    evaluate_max_waiting_time,
+    interpolated_max_waiting_time,
+)
 from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategy import (
     BackoffStrategy,
 )
 from airbyte_cdk.sources.types import Config
+from airbyte_cdk.utils import AirbyteTracedException
 
 
 @dataclass
@@ -28,15 +33,18 @@ class WaitUntilTimeFromHeaderBackoffStrategy(BackoffStrategy):
 
     Attributes:
         header (str): header to read wait time from
-        min_wait (Optional[float]): minimum time to wait for safety
+        min_wait (Optional[Union[float, InterpolatedString, str]]): minimum time to wait for safety
         regex (Optional[str]): optional regex to apply on the header to extract its value
+        max_waiting_time_in_seconds (Optional[Union[float, InterpolatedString, str]]): stop the stream
+            rather than wait longer than this
     """
 
     header: Union[InterpolatedString, str]
     parameters: InitVar[Mapping[str, Any]]
     config: Config
     min_wait: Optional[Union[float, InterpolatedString, str]] = None
     regex: Optional[Union[InterpolatedString, str]] = None
+    max_waiting_time_in_seconds: Optional[Union[float, InterpolatedString, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         self.header = InterpolatedString.create(self.header, parameters=parameters)
@@ -45,6 +53,9 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         )
         if not isinstance(self.min_wait, InterpolatedString):
             self.min_wait = InterpolatedString.create(str(self.min_wait), parameters=parameters)
+        self._max_waiting_time_in_seconds = interpolated_max_waiting_time(
+            self.max_waiting_time_in_seconds, parameters
+        )
 
     def backoff_time(
         self,
@@ -60,18 +71,38 @@ def backoff_time(
             regex = None
         wait_until = None
         if isinstance(response_or_exception, requests.Response):
+            # get_numeric_value_from_header returns a float or None, never a string
             wait_until = get_numeric_value_from_header(response_or_exception, header, regex)
         min_wait = self.min_wait.eval(self.config)  # type: ignore # header is always cast to an interpolated string
-        if wait_until is None or not wait_until:
-            return float(min_wait) if min_wait else None
-        if (isinstance(wait_until, str) and wait_until.isnumeric()) or isinstance(
-            wait_until, numbers.Number
-        ):
-            wait_time = float(wait_until) - now
-        else:
-            return float(min_wait)
+        if not wait_until:
+            return self._capped(float(min_wait)) if min_wait else None
+        wait_time = wait_until - now
         if min_wait:
-            return float(max(wait_time, min_wait))
+            return self._capped(float(max(wait_time, min_wait)))
         elif wait_time < 0:
             return None
+        return self._capped(wait_time)
+
+    def _capped(self, wait_time: float) -> float:
+        """Raise rather than wait longer than `max_waiting_time_in_seconds`.
+
+        The cap is compared against the wait this strategy is about to return, not against the
+        raw header: unlike `Retry-After`, the header here is an absolute timestamp, so only the
+        computed difference is a duration. It is also applied after the `min_wait` floor, so a
+        cap below the floor wins -- a caller asking never to wait more than N seconds means it,
+        even when the floor would otherwise round the wait up past N.
+        """
+        max_waiting_time = evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config)
+        # `>=` rather than `>` to match WaitTimeFromHeader, so one field name does not mean two
+        # different things depending on which strategy it is written on. A cap of 0 therefore
+        # refuses every wait, which is what "never wait" has to mean.
+        if max_waiting_time is not None and wait_time >= max_waiting_time:
+            raise AirbyteTracedException(
+                internal_message=(
+                    f"Rate limit wait time {wait_time}s is greater than the maximum of "
+                    f"{max_waiting_time}s this stream is allowed to wait. Stopping the stream..."
+                ),
+                message="The rate limit wait time is longer than the connector is allowed to wait.",
+                failure_type=FailureType.transient_error,
+            )
         return wait_time