From 78882bc01a55550f3ad6c54df6e8233605b3d681 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Thu, 20 Aug 2026 19:27:03 +0300 Subject: [PATCH 1/8] fix(http): decide token rotation before a backoff strategy can refuse to wait MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_handle_error_resolution` computes a backoff and then, for a RATE_LIMITED resolution, replaces it with `TOKEN_ROTATION_BACKOFF` when the authenticator reports a spare credential (#1117). That second step is written as a modification of the first one's result, so it only runs if a strategy returned a number. `WaitUntilTimeFromHeader.max_waiting_time_in_seconds` (#1123) does not return a number when the wait exceeds the cap — it raises. So whenever the cap is exceeded the function exits before the rotation question is asked, and the stream ends while a fully-quota'd credential sits idle. Measured on source-github: two tokens, a 60-minute reset and a 30-minute budget stops the sync, where the retry would have gone out on the other token in 0.1s. The rotation question is now asked first. When another credential can serve the retry, the strategies are not consulted at all: the wait they compute describes the window of the credential that was rejected, which is not the one the retry will use. When there is no spare credential the strategies run exactly as before, so a cap still ends the stream — which is what it is for. Co-Authored-By: Claude Opus 5 (1M context) --- .../sources/streams/http/http_client.py | 40 ++++++----- .../sources/streams/http/test_http_client.py | 69 +++++++++++++++++++ 2 files changed, 91 insertions(+), 18 deletions(-) diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 263a4ea1b..40ec61ad4 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -588,29 +588,33 @@ 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. + 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: 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." + f"{self.TOKEN_ROTATION_BACKOFF}s with another one instead of waiting for " + f"the rate limit to reset." ) 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/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index d7e0e51f8..80d023f2d 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -1295,6 +1295,75 @@ 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 = HttpClient( + name="test", + logger=logging.getLogger("test"), + authenticator=_SpareTokenAuthenticator(has_spare=True), + error_handler=HttpStatusErrorHandler( + logger=logging.getLogger("test"), + error_mapping={ + 429: ErrorResolution( + response_action=ResponseAction.RATE_LIMITED, + failure_type=FailureType.transient_error, + error_message="rate limited", + ) + }, + max_retries=1, + ), + 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 = HttpClient( + name="test", + logger=logging.getLogger("test"), + authenticator=_SpareTokenAuthenticator(has_spare=False), + error_handler=HttpStatusErrorHandler( + logger=logging.getLogger("test"), + error_mapping={ + 429: ErrorResolution( + response_action=ResponseAction.RATE_LIMITED, + failure_type=FailureType.transient_error, + error_message="rate limited", + ) + }, + max_retries=1, + ), + 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_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}]) From da2e5a2a34e03b75ff9df4bd8fda054add38308a Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Thu, 20 Aug 2026 20:00:01 +0300 Subject: [PATCH 2/8] docs, tests: address review of the rotation reorder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings applied. The schema descriptions for both `max_waiting_time_in_seconds` fields, and `_capped`'s docstring, promised "stop the stream" and "0 means never wait" unconditionally. Neither holds once rotation preempts the strategy, so each now says the bound applies only when waiting is the only way forward. Two tests added. One drives the real WaitUntilTimeFromHeaderBackoffStrategy with a cap below the wait the response asks for, so the integration that regressed is pinned rather than only the client's contract against a stub — the stub tests would survive the cap being changed to return instead of raise. The other covers a rate limit that produces no backoff at all, which now rotates where it previously fell through to exponential retry: intended, but a behaviour change worth pinning. The log line says a wait was skipped without the number it can no longer compute, and names the cap, since nothing else tells an operator their configured bound was not consulted. Not applied: resolving the cap in `__post_init__` so an unreadable one fails at construction. It is the right fix for a real gap — on the rotation path a manifest-level cap error now surfaces only when no spare credential exists — but it moves when `max_waiting_time_in_seconds` raises and breaks 9 tests that assert the current timing, so it changes #1123's contract rather than this PR's ordering. Recorded in the comment; belongs in its own change. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 4 +- .../models/declarative_component_schema.py | 4 +- ...until_time_from_header_backoff_strategy.py | 5 + .../sources/streams/http/http_client.py | 21 +++- .../sources/streams/http/test_http_client.py | 97 +++++++++++++++++++ 5 files changed, 124 insertions(+), 7 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 016e78bee..974c2aa0e 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. 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 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. 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 d09848a23..31674bd95 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.", 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 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.", 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/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 8d419060a..69a45b5de 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 @@ -91,6 +91,11 @@ def _capped(self, wait_time: float) -> float: 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 diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 40ec61ad4..8944f95b2 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -594,16 +594,31 @@ def _handle_error_resolution( # 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 sending + # credential is tracked and spent. And a `max_waiting_time_in_seconds` the manifest + # got wrong -- one that cannot be evaluated -- is no longer reported here, since + # that error is raised from inside the strategy; it still surfaces on the first + # rate limit that finds no spare credential. Resolving the cap once at construction + # would close that gap on every path, but it moves when `max_waiting_time_in_seconds` + # fails, so it belongs in its own change rather than in this reorder. rotate_instead_of_waiting = ( error_resolution.response_action == ResponseAction.RATE_LIMITED and self._can_retry_on_another_token(request) ) 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 for " - f"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: diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 80d023f2d..2a15546aa 100644 --- a/unit_tests/sources/streams/http/test_http_client.py +++ b/unit_tests/sources/streams/http/test_http_client.py @@ -1364,6 +1364,103 @@ def test_a_refusing_strategy_still_ends_the_stream_without_a_spare_credential(re 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 = HttpClient( + name="test", + logger=logging.getLogger("test"), + authenticator=_SpareTokenAuthenticator(has_spare=True), + error_handler=HttpStatusErrorHandler( + logger=logging.getLogger("test"), + error_mapping={ + 429: ErrorResolution( + response_action=ResponseAction.RATE_LIMITED, + failure_type=FailureType.transient_error, + error_message="rate limited", + ) + }, + max_retries=1, + ), + # 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 = HttpClient( + name="test", + logger=logging.getLogger("test"), + authenticator=_SpareTokenAuthenticator(has_spare=True), + error_handler=HttpStatusErrorHandler( + logger=logging.getLogger("test"), + error_mapping={ + 429: ErrorResolution( + response_action=ResponseAction.RATE_LIMITED, + failure_type=FailureType.transient_error, + error_message="rate limited", + ) + }, + max_retries=1, + ), + 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}]) From e09a77f523655634c7695c7afb52bf898b287343 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Thu, 20 Aug 2026 20:46:30 +0300 Subject: [PATCH 3/8] fix: quote the two descriptions that broke the manifest schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sentence added in the previous commit contains ": ", which ends a plain YAML scalar — so declarative_component_schema.yaml stopped parsing and every low-code source failed at startup, before any connector was built. A documentation edit, not the behaviour change, took out four Pytest matrix jobs and two connector checks. Both descriptions are now single-quoted, which is what the file already does for the three other descriptions containing ": ". Quotes are YAML syntax rather than part of the value, so the text still matches models/declarative_component_schema.py verbatim and no regeneration is needed. Nothing asserted that the schema the CDK ships can be read, which is why a prose edit reached CI at all. test_declarative_component_schema_is_loadable loads it through the same helper every declarative source uses and reads every description back, so a truncated scalar fails in milliseconds instead of taking 194 unrelated tests down with it. Verified it fails on the broken file. Also qualifies both class docstrings, and adds the note to WaitTimeFromHeader, whose cap is checked inline and had none. While in that file: its comment claimed WaitUntilTimeFromHeader stops at `>` and this one at `>=`; both use `>=`. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 4 +- .../wait_time_from_header_backoff_strategy.py | 12 ++++-- ...until_time_from_header_backoff_strategy.py | 4 +- ...eclarative_component_schema_is_loadable.py | 40 +++++++++++++++++++ 4 files changed, 54 insertions(+), 6 deletions(-) create mode 100644 unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 974c2aa0e..6071f6abe 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. + 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.' 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. + 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.' anyOf: - type: number - type: string 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 b62b01341..9b5032d01 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,9 @@ 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 longer than this. Only governs waits that are actually taken: when + 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. """ header: Union[InterpolatedString, str] @@ -68,10 +70,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 69a45b5de..24d0e6e5b 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,9 @@ 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 longer than this. Only governs waits that are actually taken: when + 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. """ header: Union[InterpolatedString, str] 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 000000000..405b61247 --- /dev/null +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -0,0 +1,40 @@ +# +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +# + +"""The shipped manifest schema has to parse, and nothing else asserted that it does. + +`_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. +""" + +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + _get_declarative_component_schema, +) + + +def test_the_shipped_component_schema_parses(): + schema = _get_declarative_component_schema() + + assert schema["title"] == "DeclarativeSource" + assert "HttpRequester" in schema["definitions"] + + +def test_every_description_survives_the_yaml_round_trip(): + """A description that ends early still parses — it just silently loses its tail, or turns the + rest of the sentence into a key. Reading them all back catches the truncation too.""" + schema = _get_declarative_component_schema() + + for name, definition in schema["definitions"].items(): + for field_name, field_schema in (definition.get("properties") or {}).items(): + description = ( + field_schema.get("description") if isinstance(field_schema, dict) else None + ) + if description is not None: + assert isinstance(description, str), ( + f"{name}.{field_name} description is not a string" + ) + assert description.strip(), f"{name}.{field_name} has an empty description" From c8a491e4ca6231304539232d5a5721d51d54e5ad Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Fri, 21 Aug 2026 13:03:40 +0300 Subject: [PATCH 4/8] docs, tests: make the cap guard real and its wording exact The schema guard's second test claimed to catch a description truncated by a `" #"` comment marker, but only asserted that each description was a non-empty string -- which a truncated description still is. It now lints the source text, read through the loader's own `pkgutil.get_data`, for any plain scalar carrying `": "` or `" #"`. Widened from `description` to every inline value, since the hazard belongs to the scalar style rather than to the field; list items stay excluded, because `- key: value` is a nested mapping. Two wording fixes to the `max_waiting_time_in_seconds` prose. The rotation qualification omitted that `HttpClient` only skips the strategy on a rate-limited response, so it overstated for every other retryable error. And both strategies compare `>=` while their documentation said "longer than", including the message `WaitUntilTimeFromHeader._capped` raises -- a wait exactly equal to the cap is refused, and now the prose says so. Behaviour is unchanged. Both schema descriptions were mirrored into the generated models and verified verbatim-equal, so no codegen run is needed. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 4 +- .../models/declarative_component_schema.py | 4 +- .../wait_time_from_header_backoff_strategy.py | 8 ++- ...until_time_from_header_backoff_strategy.py | 14 +++-- ...eclarative_component_schema_is_loadable.py | 61 ++++++++++++++----- 5 files changed, 64 insertions(+), 27 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 6071f6abe..7a794dd0c 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.' + 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.' 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.' + 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.' 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 31674bd95..57c1a46a0 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.", + 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.", 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. This bound applies only when waiting is the only way forward: if the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated.", + 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.", 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/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 9b5032d01..4288fbea9 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,9 +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. Only governs waits that are actually taken: when - 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. + 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] 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 24d0e6e5b..05238d31c 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,9 +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. Only governs waits that are actually taken: when - 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. + 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] @@ -86,7 +88,7 @@ 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 @@ -106,8 +108,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/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py index 405b61247..bd8121f64 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -9,12 +9,30 @@ 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. + +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, ) +# `key: value` with the value on the same line. List items are excluded on purpose: `- key: value` +# is a nested mapping, where a colon is structure rather than a truncated sentence. +_INLINE_VALUE = re.compile(r"^\s*[\w$\"-]+:\s+(\S.*?)\s*$") + +# 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 = "'\"|>&*[{" + def test_the_shipped_component_schema_parses(): schema = _get_declarative_component_schema() @@ -23,18 +41,33 @@ def test_the_shipped_component_schema_parses(): assert "HttpRequester" in schema["definitions"] -def test_every_description_survives_the_yaml_round_trip(): - """A description that ends early still parses — it just silently loses its tail, or turns the - rest of the sentence into a key. Reading them all back catches the truncation too.""" - schema = _get_declarative_component_schema() +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` or an `error_message` breaks the same 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): + match = _INLINE_VALUE.match(line) + if match is None: + continue + value = match.group(1) + if value[0] in _QUOTED_BLOCK_OR_FLOW: + continue + if ": " in value or " #" in value: + offenders.append(f"line {number}: {value[:80]}") - for name, definition in schema["definitions"].items(): - for field_name, field_schema in (definition.get("properties") or {}).items(): - description = ( - field_schema.get("description") if isinstance(field_schema, dict) else None - ) - if description is not None: - assert isinstance(description, str), ( - f"{name}.{field_name} description is not a string" - ) - assert description.strip(), f"{name}.{field_name} has an empty description" + assert not offenders, ( + "these are plain YAML scalars containing punctuation that ends them early; " + "wrap each one in single quotes:\n" + "\n".join(offenders) + ) From d7fcbab0393d8f9685a16ef992c6f612970e736f Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:05:13 +0300 Subject: [PATCH 5/8] test: scan sequence items for the same YAML punctuation hazard CodeRabbit was right that excluding list items left a hole, and the comment justifying the exclusion was wrong. The 125 false positives it cited came from reading `- key: value` as one scalar; capturing only the value, as `key: value` already did, reports none. That form is now scanned for both hazards. Bare sequence scalars (`- some text`) are scanned too, but for `" #"` only. A `": "` there does not truncate anything -- it makes the item a one-key mapping, which parses -- and telling that apart from the nested mappings the file legitimately writes that way is not possible from the text. Mutation-tested: `" #"` on a description, a title, an unquoted `- key: value` and a bare sequence item each fail the guard, as does an unquoted `": "` in a description; the untouched file passes. Co-Authored-By: Claude Opus 5 (1M context) --- ...eclarative_component_schema_is_loadable.py | 37 ++++++++++++------- 1 file changed, 24 insertions(+), 13 deletions(-) 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 index bd8121f64..24f2a5648 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -24,15 +24,23 @@ _get_declarative_component_schema, ) -# `key: value` with the value on the same line. List items are excluded on purpose: `- key: value` -# is a nested mapping, where a colon is structure rather than a truncated sentence. -_INLINE_VALUE = re.compile(r"^\s*[\w$\"-]+:\s+(\S.*?)\s*$") - # 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() @@ -47,7 +55,8 @@ def test_no_plain_scalar_carries_yaml_punctuation(): 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` or an `error_message` breaks the same way. + 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. """ # The same bytes the loader reads, fetched the same way, so this cannot drift from what ships. raw_schema = pkgutil.get_data( @@ -58,14 +67,16 @@ def test_no_plain_scalar_carries_yaml_punctuation(): offenders = [] for number, line in enumerate(raw_schema.decode().splitlines(), start=1): - match = _INLINE_VALUE.match(line) - if match is None: - continue - value = match.group(1) - if value[0] in _QUOTED_BLOCK_OR_FLOW: - continue - if ": " in value or " #" in value: - offenders.append(f"line {number}: {value[:80]}") + for pattern, hazards in _SCANS: + match = pattern.match(line) + if match is None: + continue + value = match.group(1) + if value[0] in _QUOTED_BLOCK_OR_FLOW: + continue + if any(hazard in value for hazard in hazards): + offenders.append(f"line {number}: {value[:80]}") + break # one report per line; the scans overlap on `- key: value` assert not offenders, ( "these are plain YAML scalars containing punctuation that ends them early; " From 1635fb86b3e61a68105164e05332c39504b85671 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Fri, 21 Aug 2026 14:35:53 +0300 Subject: [PATCH 6/8] docs, tests: state the skip on the base contract, trim the tooltip, reuse the client helper Four items from an external review of c8a491e4. The abstract `BackoffStrategy.backoff_time` docstring is the one every custom strategy subclasses, and it still promised the method is called for every retryable response -- via a `should_backoff()` that has not existed in this package for some time. It now states that the client skips the strategies when a rate-limited response can be retried on another credential, and that implementations must not rely on being called for side effects. The two concrete carve-outs read as elaborations of it rather than as the only notice. The two schema descriptions render as the field's help text in the Connector Builder, where two sentences about a mechanism no connector uses yet roughly doubled the tooltip. Trimmed to one clause; the full explanation stays in the class docstrings, which have the right audience. The four rotation tests each rebuilt the client that `_rate_limited_client()` already builds forty lines above. It takes a `backoff_strategy` now, which drops 65 lines with the same coverage -- still 3 of them failing against main's `http_client.py`, verified after the refactor. Recorded rather than fixed: the punctuation scan is line-anchored, so a plain scalar wrapped onto a continuation line is only checked on its first line. Co-Authored-By: Claude Opus 5 (1M context) --- .../declarative_component_schema.yaml | 4 +- .../models/declarative_component_schema.py | 4 +- .../http/error_handlers/backoff_strategy.py | 7 +- ...eclarative_component_schema_is_loadable.py | 3 + .../sources/streams/http/test_http_client.py | 75 +++---------------- 5 files changed, 23 insertions(+), 70 deletions(-) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 7a794dd0c..aa3bbbc5e 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.' + 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 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.' + 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 57c1a46a0..21a92cc3b 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.", + 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 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. This bound applies only when waiting is the only way forward: when a rate-limited response arrives and the authenticator holds another credential with quota, the retry rotates to it and neither the wait nor this bound is evaluated. Any other retryable error still consults this strategy, spare credential or not.", + 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/streams/http/error_handlers/backoff_strategy.py b/airbyte_cdk/sources/streams/http/error_handlers/backoff_strategy.py index 6ed821791..49207b513 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/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py index 24f2a5648..91ac4c302 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -57,6 +57,9 @@ def test_no_plain_scalar_carries_yaml_punctuation(): 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( diff --git a/unit_tests/sources/streams/http/test_http_client.py b/unit_tests/sources/streams/http/test_http_client.py index 2a15546aa..2c642ab9b 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), ) @@ -1311,22 +1311,8 @@ def test_rotation_is_preferred_over_a_strategy_that_refuses_to_wait(requests_moc 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 = HttpClient( - name="test", - logger=logging.getLogger("test"), - authenticator=_SpareTokenAuthenticator(has_spare=True), - error_handler=HttpStatusErrorHandler( - logger=logging.getLogger("test"), - error_mapping={ - 429: ErrorResolution( - response_action=ResponseAction.RATE_LIMITED, - failure_type=FailureType.transient_error, - error_message="rate limited", - ) - }, - max_retries=1, - ), - backoff_strategy=_RefusingBackoffStrategy(), + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=True), backoff_strategy=_RefusingBackoffStrategy() ) sleeps = [] @@ -1342,22 +1328,8 @@ def test_rotation_is_preferred_over_a_strategy_that_refuses_to_wait(requests_moc 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 = HttpClient( - name="test", - logger=logging.getLogger("test"), - authenticator=_SpareTokenAuthenticator(has_spare=False), - error_handler=HttpStatusErrorHandler( - logger=logging.getLogger("test"), - error_mapping={ - 429: ErrorResolution( - response_action=ResponseAction.RATE_LIMITED, - failure_type=FailureType.transient_error, - error_message="rate limited", - ) - }, - max_retries=1, - ), - backoff_strategy=_RefusingBackoffStrategy(), + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=False), backoff_strategy=_RefusingBackoffStrategy() ) with pytest.raises(AirbyteTracedException, match="longer than the connector is allowed"): @@ -1381,21 +1353,8 @@ def test_rotation_is_preferred_over_the_real_capped_strategy(requests_mock): {"status_code": 200}, ], ) - client = HttpClient( - name="test", - logger=logging.getLogger("test"), - authenticator=_SpareTokenAuthenticator(has_spare=True), - error_handler=HttpStatusErrorHandler( - logger=logging.getLogger("test"), - error_mapping={ - 429: ErrorResolution( - response_action=ResponseAction.RATE_LIMITED, - failure_type=FailureType.transient_error, - error_message="rate limited", - ) - }, - max_retries=1, - ), + 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", @@ -1432,22 +1391,8 @@ def test_rotation_also_covers_a_rate_limit_with_no_computed_wait(requests_mock): 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 = HttpClient( - name="test", - logger=logging.getLogger("test"), - authenticator=_SpareTokenAuthenticator(has_spare=True), - error_handler=HttpStatusErrorHandler( - logger=logging.getLogger("test"), - error_mapping={ - 429: ErrorResolution( - response_action=ResponseAction.RATE_LIMITED, - failure_type=FailureType.transient_error, - error_message="rate limited", - ) - }, - max_retries=1, - ), - backoff_strategy=strategy, + client = _rate_limited_client( + _SpareTokenAuthenticator(has_spare=True), backoff_strategy=strategy ) sleeps = [] From a44e143ebbc4012f43b6bbd625a296d66bd85360 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:01:52 +0300 Subject: [PATCH 7/8] fix(tests): stop the schema guard flagging correctly quoted sequence values The quoted-value exit used `continue`, which settles the scan but not the line: `- key: 'text # hash'` was skipped by the first scan, then re-captured whole by the bare-sequence-scalar scan, quotes and all, and reported as an offender. The assertion then told an author who had already quoted the value to quote it. `break` instead, since whichever scan reads a line first settles it. Verified against an inserted probe line: both quoted forms under a sequence key now pass where they previously failed, and every true positive still fails the guard -- `" #"` on a description, a title, an unquoted `- key: value` and a bare sequence item, plus the `": "` cases. Also reworded the module docstring, which claimed nothing asserted the schema parses nine lines above noting that 194 failures did. The 194 were an assertion, at the wrong layer with unreadable output; what the guards add is a named failure, not coverage. Reported by @tolik0. Co-Authored-By: Claude Opus 5 (1M context) --- ...est_declarative_component_schema_is_loadable.py | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) 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 index 91ac4c302..db7d849bb 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -2,13 +2,15 @@ # Copyright (c) 2026 Airbyte, Inc., all rights reserved. # -"""The shipped manifest schema has to parse, and nothing else asserted that it does. +"""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. +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 @@ -75,11 +77,15 @@ def test_no_plain_scalar_carries_yaml_punctuation(): 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: - continue + break if any(hazard in value for hazard in hazards): offenders.append(f"line {number}: {value[:80]}") - break # one report per line; the scans overlap on `- key: value` + break assert not offenders, ( "these are plain YAML scalars containing punctuation that ends them early; " From 69441bda00ce719166162c8ddb8505e8defcc8b3 Mon Sep 17 00:00:00 2001 From: Daryna Ishchenko <80129833+darynaishchenko@users.noreply.github.com> Date: Fri, 21 Aug 2026 15:14:16 +0300 Subject: [PATCH 8/8] fix(low-code): resolve max_waiting_time_in_seconds when the strategy is built A cap the manifest got wrong -- an interpolation over a config key the spec does not expose, a value resolving to NaN, a blank string -- was only discovered when a strategy was asked for a wait. Since rotation is now decided before the strategies run, that question may never be asked: while a spare credential has quota the cap is neither applied nor reported, and the broken expression resurfaces at whichever later rate limit finds every credential spent. Both capped strategies now resolve the field in `__post_init__`. `config` is a dataclass field and the cap interpolates over `config` alone, so the value is knowable as soon as the component exists, and the factory already passes the real config. A manifest mistake fails at startup, on every path. The nine tests that pinned the old timing assert construction now, which is the contract change this carries: the error is still an `AirbyteTracedException` with `system_error`, still never a raw jinja or float error, still the connector's fault rather than the user's -- only the moment moves. Fleet impact is nil today. Every user of the field resolves cleanly: source-github guards with `config.get(...) is not none`, source-klaviyo passes a float, source-granola hardcodes 60. The lazy check stays, so a strategy constructed directly in Python is unchanged. Co-Authored-By: Claude Opus 5 (1M context) --- .../max_waiting_time_helper.py | 13 ++++++----- .../wait_time_from_header_backoff_strategy.py | 6 +++++ ...until_time_from_header_backoff_strategy.py | 6 +++++ .../sources/streams/http/http_client.py | 14 +++++------ .../test_wait_time_from_header.py | 23 ++++++++----------- .../test_wait_until_time_from_header.py | 19 ++++++++------- 6 files changed, 45 insertions(+), 36 deletions(-) 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 82feb3c97..4ebb74118 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 4288fbea9..0e3ee0a0a 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 @@ -54,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, 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 05238d31c..777f53d87 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 @@ -60,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, diff --git a/airbyte_cdk/sources/streams/http/http_client.py b/airbyte_cdk/sources/streams/http/http_client.py index 8944f95b2..c9008d1e6 100644 --- a/airbyte_cdk/sources/streams/http/http_client.py +++ b/airbyte_cdk/sources/streams/http/http_client.py @@ -598,13 +598,13 @@ def _handle_error_resolution( # 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 sending - # credential is tracked and spent. And a `max_waiting_time_in_seconds` the manifest - # got wrong -- one that cannot be evaluated -- is no longer reported here, since - # that error is raised from inside the strategy; it still surfaces on the first - # rate limit that finds no spare credential. Resolving the cap once at construction - # would close that gap on every path, but it moves when `max_waiting_time_in_seconds` - # fails, so it belongs in its own change rather than in this reorder. + # 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) 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 35c1ece79..4cdee6918 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 fdaa0d890..b6104067e 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