diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 016e78bee6..aa3bbbc5e0 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -4666,7 +4666,7 @@ definitions: - "([-+]?\\d+)" max_waiting_time_in_seconds: title: Max Waiting Time in Seconds - 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. + 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. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.' anyOf: - type: number - type: string @@ -4802,7 +4802,7 @@ definitions: - "([-+]?\\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. + description: 'Stop the stream instead of waiting, when the wait this strategy computes is greater than or equal to 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. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.' anyOf: - type: number - type: string diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index d09848a238..21a92cc3be 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -1449,7 +1449,7 @@ class WaitTimeFromHeader(BaseModel): ) max_waiting_time_in_seconds: Optional[Union[float, str]] = Field( None, - 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.", + 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. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.", examples=[3600, "{{ config['max_waiting_time'] * 60 }}"], title="Max Waiting Time in Seconds", ) @@ -1478,7 +1478,7 @@ class WaitUntilTimeFromHeader(BaseModel): ) 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.", + description="Stop the stream instead of waiting, when the wait this strategy computes is greater than or equal to 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. Not evaluated when a rate-limited retry can rotate to another credential with quota; any other retryable error still consults this strategy.", examples=[3600, "{{ config['max_waiting_time'] * 60 }}"], title="Max Waiting Time in Seconds", ) diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py index 82feb3c976..4ebb74118c 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/max_waiting_time_helper.py @@ -42,12 +42,13 @@ def evaluate_max_waiting_time( 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. + Called once when the strategy is constructed, and again on each cap check. The eager call is + what makes an unresolvable interpolation a startup failure rather than something discovered at + whichever retryable error happens to reach a strategy -- a distinction that matters because + `HttpClient` skips the strategies entirely when a rate-limited retry can rotate credentials. + 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 diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py index b62b013417..0e3ee0a0a6 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_time_from_header_backoff_strategy.py @@ -33,7 +33,11 @@ class WaitTimeFromHeaderBackoffStrategy(BackoffStrategy): 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[Union[float, InterpolatedString, str]]): stop the stream - rather than wait longer than this + rather than wait this long or longer -- the bound is inclusive, so a wait exactly + equal to it is refused. Only governs waits that are actually taken: on a + rate-limited response where the authenticator holds another credential with quota, + `HttpClient` rotates onto it instead of asking this strategy for a wait, and the + bound does not apply. Any other retryable error still consults this strategy. """ header: Union[InterpolatedString, str] @@ -50,6 +54,12 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._max_waiting_time_in_seconds = interpolated_max_waiting_time( self.max_waiting_time_in_seconds, parameters ) + # Resolved here rather than only at the first retryable error. `config` is a field and this + # cap interpolates over `config` alone, so it is fully knowable the moment the component + # exists -- and since `HttpClient` decides token rotation before it asks a strategy for a + # wait, a cap that cannot be evaluated would otherwise stay silent for as long as a spare + # credential keeps the strategies from running. A manifest mistake belongs at startup. + evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) def backoff_time( self, @@ -68,10 +78,14 @@ def backoff_time( max_waiting_time = evaluate_max_waiting_time( self._max_waiting_time_in_seconds, self.config ) + # Not always reached: `HttpClient` decides token rotation before it asks a strategy + # for a wait, so on a rate limit where the authenticator has another credential with + # quota this check does not run. The cap bounds waiting, and that path is not + # waiting. # `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. + # which is what this cap has always done, and `WaitUntilTimeFromHeader` matches it -- + # a wait exactly equal to the cap is refused by both. # `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: diff --git a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py index 8d419060a3..777f53d877 100644 --- a/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py +++ b/airbyte_cdk/sources/declarative/requesters/error_handlers/backoff_strategies/wait_until_time_from_header_backoff_strategy.py @@ -36,7 +36,11 @@ class WaitUntilTimeFromHeaderBackoffStrategy(BackoffStrategy): 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 + rather than wait this long or longer -- the bound is inclusive, so a wait exactly + equal to it is refused. Only governs waits that are actually taken: on a + rate-limited response where the authenticator holds another credential with quota, + `HttpClient` rotates onto it instead of asking this strategy for a wait, and the + bound does not apply. Any other retryable error still consults this strategy. """ header: Union[InterpolatedString, str] @@ -56,6 +60,12 @@ def __post_init__(self, parameters: Mapping[str, Any]) -> None: self._max_waiting_time_in_seconds = interpolated_max_waiting_time( self.max_waiting_time_in_seconds, parameters ) + # Resolved here rather than only at the first retryable error. `config` is a field and this + # cap interpolates over `config` alone, so it is fully knowable the moment the component + # exists -- and since `HttpClient` decides token rotation before it asks a strategy for a + # wait, a cap that cannot be evaluated would otherwise stay silent for as long as a spare + # credential keeps the strategies from running. A manifest mistake belongs at startup. + evaluate_max_waiting_time(self._max_waiting_time_in_seconds, self.config) def backoff_time( self, @@ -84,13 +94,18 @@ def backoff_time( return self._capped(wait_time) def _capped(self, wait_time: float) -> float: - """Raise rather than wait longer than `max_waiting_time_in_seconds`. + """Raise rather than wait `max_waiting_time_in_seconds` or longer. 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. + + Not always reached: `HttpClient` decides token rotation before it asks a strategy for a + wait, so on a rate limit where the authenticator has another credential with quota this + method does not run and the cap does not apply. Waiting is what the cap bounds, and that + path is not waiting. """ 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 @@ -99,8 +114,8 @@ def _capped(self, wait_time: float) -> float: 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..." + f"Rate limit wait time {wait_time}s is greater than or equal to the maximum " + f"of {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, diff --git a/airbyte_cdk/sources/streams/http/error_handlers/backoff_strategy.py b/airbyte_cdk/sources/streams/http/error_handlers/backoff_strategy.py index 6ed821791c..49207b513d 100644 --- a/airbyte_cdk/sources/streams/http/error_handlers/backoff_strategy.py +++ b/airbyte_cdk/sources/streams/http/error_handlers/backoff_strategy.py @@ -18,7 +18,12 @@ def backoff_time( """ Override this method to dynamically determine backoff time e.g: by reading the X-Retry-After header. - This method is called only if should_backoff() returns True for the input request. + Not called for every retryable response. `HttpClient` skips the strategies entirely when a + rate-limited response can be retried on another credential -- the authenticator says so via + `TokenRotatingAuthenticator.has_alternative_token` -- because the wait computed here is + derived from the credential that was rejected, and the retry will not use it. Implementations + must therefore not rely on being called for side effects such as counting attempts or + emitting metrics. :param response_or_exception: The response or exception that caused the backoff. :param attempt_count: The number of attempts already performed for this request. diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 263a4ea1ba..c9008d1e64 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -588,29 +588,48 @@ def _handle_error_resolution( ResponseAction.REFRESH_TOKEN_THEN_RETRY, ): user_defined_backoff_time = None - for backoff_strategy in self._backoff_strategies: - backoff_time = backoff_strategy.backoff_time( - response_or_exception=response if response is not None else exc, - attempt_count=self._request_attempt_count[request], - ) - if backoff_time: - user_defined_backoff_time = backoff_time - break - - if ( - user_defined_backoff_time - and error_resolution.response_action == ResponseAction.RATE_LIMITED + # Asked before the strategies, not after. The backoff they compute describes the + # credential the server just rejected, so when another credential can serve the + # retry that wait is irrelevant -- and a strategy is allowed to refuse a wait by + # raising (`max_waiting_time_in_seconds`), which would otherwise end the stream + # before rotation was ever considered. Rotating is strictly the better outcome + # there: it is the same retry, seconds from now, on a credential with quota. + # + # Two consequences of not calling the strategies, both deliberate. A rate limit + # that yields no backoff at all now rotates too, rather than falling through to + # the default exponential retry -- on a rotating credential that is the better + # behaviour, and `has_alternative_token` only answers True when the retry will + # rotate -- the CDK's own authenticator narrows that further, to a sending credential + # that is tracked and spent. And a `max_waiting_time_in_seconds` the manifest got + # wrong -- one that cannot be evaluated -- is not reported from here, since that + # error is raised from inside the strategy. Both capped strategies therefore resolve + # the field once in `__post_init__` too, so a manifest mistake fails at startup + # rather than waiting for a rate limit that finds no spare credential. + rotate_instead_of_waiting = ( + error_resolution.response_action == ResponseAction.RATE_LIMITED and self._can_retry_on_another_token(request) - ): - # The backoff was derived from this response's headers, which only describe the - # credential that was rejected. Another one has quota, and the retry re-signs the - # request, so waiting out this window would idle for nothing. + ) + + if rotate_instead_of_waiting: + # Says that a wait was skipped without the number, which is no longer computed, + # and names the cap explicitly: a connector that configured one gets no other + # signal that the retry went ahead without consulting it. self._logger.info( - f"Rate limited on the current credential; retrying in " - f"{self.TOKEN_ROTATION_BACKOFF}s with another one instead of waiting " - f"{user_defined_backoff_time:.0f}s for the rate limit to reset." + "Rate limited on the current credential; retrying in " + f"{self.TOKEN_ROTATION_BACKOFF}s with another one instead of waiting for the " + "rate limit to reset. Any configured backoff, including a wait cap, is not " + "evaluated for this retry." ) user_defined_backoff_time = self.TOKEN_ROTATION_BACKOFF + else: + for backoff_strategy in self._backoff_strategies: + backoff_time = backoff_strategy.backoff_time( + response_or_exception=response if response is not None else exc, + attempt_count=self._request_attempt_count[request], + ) + if backoff_time: + user_defined_backoff_time = backoff_time + break error_message = ( error_resolution.error_message diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py index 35c1ece797..4cdee6918c 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_time_from_header.py @@ -137,13 +137,13 @@ def test_max_waiting_time_is_interpolated_from_config(config, expected): def test_given_max_waiting_time_cannot_be_evaluated_then_raise_system_error( max_waiting_time_in_seconds, config ): - """The cap is only read while handling an error that was already going to be retried, so an - unresolvable interpolation must not surface as an unhandled jinja or float error. It is a - system error because the field is declared in the manifest: the user has nothing to fix.""" - strategy = _strategy(max_waiting_time_in_seconds, config=config) - + """Raised when the strategy is built, not when a retry first needs the cap. Waiting for a + retryable error would mean never raising at all on a connector whose rate limits are always + served by rotating to another credential, since `HttpClient` skips the strategies there. It + is a system error because the field is declared in the manifest: the user has nothing to fix, + and an unresolvable interpolation must not surface as a raw jinja or float error either.""" with pytest.raises(AirbyteTracedException) as exc_info: - strategy.backoff_time(_response(120), 1) + _strategy(max_waiting_time_in_seconds, config=config) assert exc_info.value.failure_type == FailureType.system_error assert "max_waiting_time_in_seconds" in exc_info.value.internal_message @@ -169,20 +169,17 @@ def test_given_header_asks_for_no_wait_then_no_cap_refuses_it(max_waiting_time_i ) def test_given_max_waiting_time_resolves_to_nothing_then_raise_rather_than_drop_the_cap(config): """A blank config value must not leave the wait unbounded: "no cap" is spelled by leaving the - field out of the manifest, so a field that is present and resolves to nothing is a failure.""" - strategy = _strategy("{{ config['max_waiting_time'] }}", config=config) - + field out of the manifest, so a field that is present and resolves to nothing is a failure -- + at construction, before any request has been sent.""" with pytest.raises(AirbyteTracedException) as exc_info: - strategy.backoff_time(_response(120), 1) + _strategy("{{ config['max_waiting_time'] }}", config=config) assert exc_info.value.failure_type == FailureType.system_error def test_given_interpolation_raises_a_traced_error_then_keep_its_own_failure_type_and_message(): """`stream_state` interpolation raises an AirbyteTracedException of its own, with a message written for that case. The cap's own error handling must not reclassify or replace it.""" - strategy = _strategy("{{ stream_state['max_waiting_time'] }}") - with pytest.raises(AirbyteTracedException) as exc_info: - strategy.backoff_time(_response(120), 1) + _strategy("{{ stream_state['max_waiting_time'] }}") assert exc_info.value.failure_type == FailureType.config_error assert "`stream_state` is no longer supported for interpolation" in exc_info.value.message diff --git a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py index fdaa0d890e..b6104067e2 100644 --- a/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py +++ b/unit_tests/sources/declarative/requesters/error_handlers/backoff_strategies/test_wait_until_time_from_header.py @@ -233,13 +233,13 @@ def test_cap_is_interpolated_from_config(time_mock, config, expected): def test_given_cap_cannot_be_evaluated_then_raise_system_error( time_mock, max_waiting_time_in_seconds, config ): - """The cap is only read while handling an error that was already going to be retried, so an - unresolvable interpolation must not surface as an unhandled jinja or float error. It is a - system error because the field is declared in the manifest: the user has nothing to fix.""" - strategy = _strategy(max_waiting_time_in_seconds, config=config) - + """Raised when the strategy is built, not when a retry first needs the cap. Waiting for a + retryable error would mean never raising at all on a connector whose rate limits are always + served by rotating to another credential, since `HttpClient` skips the strategies there. It + is a system error because the field is declared in the manifest: the user has nothing to fix, + and an unresolvable interpolation must not surface as a raw jinja or float error either.""" with pytest.raises(AirbyteTracedException) as exc_info: - strategy.backoff_time(_response(), 1) + _strategy(max_waiting_time_in_seconds, config=config) assert exc_info.value.failure_type == FailureType.system_error assert "max_waiting_time_in_seconds" in exc_info.value.internal_message @@ -253,9 +253,8 @@ def test_non_finite_cap_is_rejected(time_mock, cap): """NaN is the one value that would switch the cap off without saying so -- every comparison against it is False, so the wait this field exists to bound would run unbounded again. Infinity is rejected as the same kind of mistake rather than read as "no cap", which is - already spelled by leaving the field out.""" - strategy = _strategy(cap) - + already spelled by leaving the field out. Rejected at construction, like every other cap the + field cannot resolve.""" with pytest.raises(AirbyteTracedException) as exc_info: - strategy.backoff_time(_response(), 1) + _strategy(cap) assert exc_info.value.failure_type == FailureType.system_error diff --git a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py new file mode 100644 index 0000000000..db7d849bba --- /dev/null +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -0,0 +1,93 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +"""The shipped manifest schema has to parse, and only 194 unrelated failures said so. + +`_get_declarative_component_schema()` runs before any low-code source is built, so a malformed +`declarative_component_schema.yaml` fails every declarative connector at startup. It is also easy +to break from a documentation edit alone: the descriptions are plain YAML scalars, so a `": "` in +prose ends the scalar and the file stops being a mapping. That happened in this file's history and +surfaced as 194 unrelated test failures rather than as one obvious error -- the defect was caught, +at the wrong layer and with unreadable output. What these guards add is not coverage but a named +failure in milliseconds. + +Two guards, because the two ways prose breaks a plain scalar look nothing alike. A `": "` is loud: +the file stops parsing, so loading it is enough to catch it. A `" #"` is silent: YAML reads the +rest of the line as a comment, the value keeps parsing as a shorter string, and nothing in the +parsed tree records that a sentence went missing. Only the source text knows, so the second guard +reads that instead of the loaded schema. +""" + +import pkgutil +import re + +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + _get_declarative_component_schema, +) + +# A plain scalar is one not opened with a quote, a block indicator or a flow collection, and it is +# the only style these two characters are dangerous in: inside `'...'` or a `|` block they are +# text like any other. +_QUOTED_BLOCK_OR_FLOW = "'\"|>&*[{" + +# What to read a value out of, and which punctuation can truncate it there. +# +# `key: value` and `- key: value` both hold prose and are checked for both hazards. A sequence +# item that is a bare scalar (`- some text`) is checked for `" #"` only: a `": "` in that +# position does not truncate anything, it makes the item a one-key mapping, which parses. Telling +# that apart from the 125 nested mappings the file legitimately writes that way is not possible +# from the text, so it is left alone. +_SCANS = ( + (re.compile(r"^\s*(?:-\s+)?[\w$\"-]+:\s+(\S.*?)\s*$"), (": ", " #")), + (re.compile(r"^\s*-\s+(\S.*?)\s*$"), (" #",)), +) + + +def test_the_shipped_component_schema_parses(): + schema = _get_declarative_component_schema() + + assert schema["title"] == "DeclarativeSource" + assert "HttpRequester" in schema["definitions"] + + +def test_no_plain_scalar_carries_yaml_punctuation(): + """`": "` ends a plain scalar and `" #"` starts a comment, so either one truncates a value + written as prose -- the first noisily, the second without a trace. Checked against the source + text rather than the parsed schema, which by then has already lost the evidence. + + Every inline value is checked, not only `description`, because the hazard belongs to the + scalar style rather than to the field: a `title`, an `error_message` or a value written + under a sequence item all break the same way. + + Line-anchored, so a plain scalar wrapped onto a continuation line is only checked on its first + line. The file has none today, and quoting a long value is the fix either way. + """ + # The same bytes the loader reads, fetched the same way, so this cannot drift from what ships. + raw_schema = pkgutil.get_data( + "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml" + ) + assert raw_schema is not None, "the manifest schema is missing from the package" + + offenders = [] + + for number, line in enumerate(raw_schema.decode().splitlines(), start=1): + for pattern, hazards in _SCANS: + match = pattern.match(line) + if match is None: + continue + value = match.group(1) + # Both exits break rather than continue: the scans overlap on `- key: value`, and + # whichever reads it first settles the line. Falling through to the next scan would + # re-capture `key: 'text # hash'` as one bare scalar, quotes and all, and report a + # value the author had already quoted correctly. + if value[0] in _QUOTED_BLOCK_OR_FLOW: + break + if any(hazard in value for hazard in hazards): + offenders.append(f"line {number}: {value[:80]}") + break + + assert not offenders, ( + "these are plain YAML scalars containing punctuation that ends them early; " + "wrap each one in single quotes:\n" + "\n".join(offenders) + ) diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index d7e0e51f81..2c642ab9b1 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -1235,7 +1235,7 @@ def has_alternative_token(self, request): return self.has_spare -def _rate_limited_client(authenticator, backoff_seconds=1800): +def _rate_limited_client(authenticator, backoff_seconds=1800, backoff_strategy=None): return HttpClient( name="test", logger=logging.getLogger("test"), @@ -1256,7 +1256,7 @@ def _rate_limited_client(authenticator, backoff_seconds=1800): }, max_retries=1, ), - backoff_strategy=_ConstantBackoffStrategy(backoff_seconds), + backoff_strategy=backoff_strategy or _ConstantBackoffStrategy(backoff_seconds), ) @@ -1295,6 +1295,117 @@ def test_rate_limit_wait_is_paid_when_no_other_credential_is_available(requests_ assert max(sleeps) > 1000, f"the full rate-limit wait should stand, slept {sleeps}" +class _RefusingBackoffStrategy(BackoffStrategy): + """A strategy that refuses to wait, the way `max_waiting_time_in_seconds` does.""" + + def backoff_time(self, *args, **kwargs): + raise AirbyteTracedException( + internal_message="wait longer than allowed", + message="The rate limit wait time is longer than the connector is allowed to wait.", + failure_type=FailureType.transient_error, + ) + + +def test_rotation_is_preferred_over_a_strategy_that_refuses_to_wait(requests_mock): + """A capped strategy raises rather than returning a number. Rotation has to be decided + before it runs, or a bound on waiting silently becomes a bound on the sync: the retry the + spare credential could serve in 0.1s never happens.""" + requests_mock.get("https://example.com/", [{"status_code": 429}, {"status_code": 200}]) + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=True), backoff_strategy=_RefusingBackoffStrategy() + ) + + sleeps = [] + with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)): + _, response = client.send_request( + http_method="GET", url="https://example.com/", request_kwargs={} + ) + + assert response.status_code == 200 + assert max(sleeps) < 5, f"expected a prompt retry on the spare credential, slept {sleeps}" + + +def test_a_refusing_strategy_still_ends_the_stream_without_a_spare_credential(requests_mock): + """The cap must keep working when rotation is not an option — that is what it is for.""" + requests_mock.get("https://example.com/", [{"status_code": 429}, {"status_code": 200}]) + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=False), backoff_strategy=_RefusingBackoffStrategy() + ) + + with pytest.raises(AirbyteTracedException, match="longer than the connector is allowed"): + client.send_request(http_method="GET", url="https://example.com/", request_kwargs={}) + + +def test_rotation_is_preferred_over_the_real_capped_strategy(requests_mock): + """The stub tests above pin the client's contract — a strategy may raise. This one pins the + integration that actually regressed: the real `WaitUntilTimeFromHeaderBackoffStrategy` with a + cap it cannot honour. Without it, a change making the cap return instead of raise would leave + both stub tests green while the bug came back.""" + from airbyte_cdk.sources.declarative.requesters.error_handlers.backoff_strategies import ( + WaitUntilTimeFromHeaderBackoffStrategy, + ) + + reset = int(time.time()) + 3600 + requests_mock.get( + "https://example.com/", + [ + {"status_code": 429, "headers": {"X-RateLimit-Reset": str(reset)}}, + {"status_code": 200}, + ], + ) + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=True), + # A cap far below the hour the response asks for: the strategy would refuse the wait. + backoff_strategy=WaitUntilTimeFromHeaderBackoffStrategy( + header="X-RateLimit-Reset", + parameters={}, + config={}, + max_waiting_time_in_seconds=60, + ), + ) + + sleeps = [] + with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)): + _, response = client.send_request( + http_method="GET", url="https://example.com/", request_kwargs={} + ) + + assert response.status_code == 200 + assert max(sleeps) < 5, f"expected a prompt retry on the spare credential, slept {sleeps}" + + +class _NoWaitBackoffStrategy(BackoffStrategy): + """Returns no wait at all, as a strategy does when the response carries no timing header.""" + + def __init__(self): + self.calls = 0 + + def backoff_time(self, *args, **kwargs): + self.calls += 1 + return None + + +def test_rotation_also_covers_a_rate_limit_with_no_computed_wait(requests_mock): + """Deciding rotation first widens it to rate limits that produce no backoff at all, where the + old order fell through to the default exponential retry. Intended — the retry goes out on a + credential with quota — but it is a behaviour change, so it is pinned rather than implied.""" + requests_mock.get("https://example.com/", [{"status_code": 429}, {"status_code": 200}]) + strategy = _NoWaitBackoffStrategy() + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=True), backoff_strategy=strategy + ) + + sleeps = [] + with patch("time.sleep", side_effect=lambda seconds: sleeps.append(seconds)): + _, response = client.send_request( + http_method="GET", url="https://example.com/", request_kwargs={} + ) + + assert response.status_code == 200 + assert strategy.calls == 0, "the strategies are skipped entirely on the rotation path" + assert max(sleeps) < 5, f"expected the rotation retry, slept {sleeps}" + + def test_non_rate_limit_retry_is_not_shortened_by_a_spare_credential(requests_mock): """A 500 has nothing to do with credentials; its backoff must be left alone.""" requests_mock.get("https://example.com/", [{"status_code": 500}, {"status_code": 200}])