From 3f3421b232616345b68d0c04dd99f8228b19d781 Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Mon, 13 Jul 2026 13:03:13 +0000 Subject: [PATCH] feat(client): retry transient GET failures with backoff Idempotent GETs now retry on 500/502/503/504 and transient transport errors (connection reset/refused, timeout, truncated body) with doubling backoff capped at 30s, up to 7 attempts per logical request. Mutating methods are never duplicated and streamed GETs are excluded so stream consumers keep control of the open path. SSL certificate errors raise on the first attempt because they are deterministic and retrying only delays the report. The existing 429 Retry-After handling is unchanged and composes with the new layer. Retry warnings go through the client logger; pipeline-run, artifact, pipeline hydration, published-component, and secret commands thread their --log-type logger into the client so the warnings follow the configured sink, and logger-less programmatic clients stay silent. --- .../src/tangle_cli/artifacts_cli.py | 1 + packages/tangle-cli/src/tangle_cli/client.py | 153 +++++- .../src/tangle_cli/pipeline_runs_cli.py | 16 +- .../src/tangle_cli/pipelines_cli.py | 1 + .../tangle_cli/published_components_cli.py | 9 +- .../tangle-cli/src/tangle_cli/secrets_cli.py | 7 +- tests/test_api_cli.py | 2 + tests/test_artifacts_cli.py | 2 + tests/test_client.py | 439 ++++++++++++++++++ tests/test_components_cli.py | 3 + tests/test_secrets_cli.py | 3 + 11 files changed, 622 insertions(+), 14 deletions(-) diff --git a/packages/tangle-cli/src/tangle_cli/artifacts_cli.py b/packages/tangle-cli/src/tangle_cli/artifacts_cli.py index a6d59e1..bb9f496 100644 --- a/packages/tangle-cli/src/tangle_cli/artifacts_cli.py +++ b/packages/tangle-cli/src/tangle_cli/artifacts_cli.py @@ -80,6 +80,7 @@ def artifacts_get( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="artifact commands", + logger=logger, ) if require_available := getattr(client, "require_available", None): require_available() diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 060c960..9b4c118 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -40,6 +40,29 @@ ) +class _RetryBudget: + """Shared attempt/deadline budget for one logical request. + + The transient-5xx, 429 rate-limit, and 401 auth-refresh retry layers all + draw from a single instance so a composed outage cannot multiply their + per-layer limits into a large physical request count. ``remaining`` counts + the physical requests still permitted; ``deadline`` is a ``time.monotonic`` + value past which no further retry is attempted. + """ + + __slots__ = ("remaining", "deadline") + + def __init__(self, max_attempts: int, deadline: float) -> None: + self.remaining = max_attempts + self.deadline = deadline + + def consume(self) -> None: + self.remaining -= 1 + + def can_retry(self) -> bool: + return self.remaining > 0 and time.monotonic() < self.deadline + + class TangleApiClient(GeneratedTangleApiOperations): """Single public API wrapper for Tangle backends. @@ -50,9 +73,19 @@ class TangleApiClient(GeneratedTangleApiOperations): _REDIRECT_STATUSES = {301, 302, 303, 307, 308} _MAX_REDIRECTS = 5 - _MAX_RATE_LIMIT_RETRIES = 3 _RATE_LIMIT_BACKOFF_SECONDS = 1.0 _MAX_RETRY_AFTER_SECONDS = 60.0 + _RETRYABLE_GET_STATUSES = frozenset({500, 502, 503, 504}) + _MAX_GET_RETRIES = 6 + _GET_RETRY_BACKOFF_SECONDS = 1.0 + _MAX_GET_RETRY_BACKOFF_SECONDS = 30.0 + # A single logical request may issue at most ``_MAX_GET_RETRIES + 1`` + # physical requests total, shared across the transient-5xx, 429 rate-limit, + # and 401 auth-refresh layers, and must not spend more than + # ``_MAX_RETRY_ELAPSED_SECONDS`` retrying. One shared budget prevents the + # layers from multiplying into a large physical request count during an + # outage (e.g. interleaved 503/429 responses, or a 401 mid-sequence). + _MAX_RETRY_ELAPSED_SECONDS = 120.0 def __init__( self, @@ -121,6 +154,11 @@ def _make_request( clean_params = self._clean_mapping(params) request_method = method.upper() + budget = _RetryBudget( + self._MAX_GET_RETRIES + 1, + time.monotonic() + self._MAX_RETRY_ELAPSED_SECONDS, + ) + self._refresh_auth() response = self._request_with_rate_limit_retries( request_method, @@ -130,8 +168,11 @@ def _make_request( extra_headers=extra_headers, timeout=timeout, request_kwargs=kwargs, + budget=budget, ) - if response.status_code == 401: + # The auth-refresh retry draws from the same budget, so a 401 late in a + # transient/rate-limit sequence cannot start a fresh round of retries. + if response.status_code == 401 and budget.can_retry(): self._refresh_auth() response = self._request_with_rate_limit_retries( request_method, @@ -141,6 +182,7 @@ def _make_request( extra_headers=extra_headers, timeout=timeout, request_kwargs=kwargs, + budget=budget, ) return response @@ -154,10 +196,11 @@ def _request_with_rate_limit_retries( extra_headers: Mapping[str, str] | None, timeout: float, request_kwargs: Mapping[str, Any], + budget: _RetryBudget, ) -> requests.Response: - response: requests.Response | None = None - for attempt in range(self._MAX_RATE_LIMIT_RETRIES + 1): - response = self._request_with_same_origin_redirects( + rate_limit_round = 0 + while True: + response = self._request_with_transient_retries( method, url, params=params, @@ -165,11 +208,105 @@ def _request_with_rate_limit_retries( extra_headers=extra_headers, timeout=timeout, request_kwargs=request_kwargs, + budget=budget, ) - if response.status_code != 429 or attempt == self._MAX_RATE_LIMIT_RETRIES: + # A 429 retry re-enters the transient layer, so it must draw from the + # shared budget rather than a per-round allowance. + if response.status_code != 429 or not budget.can_retry(): return response - self._sleep_for_rate_limit(response, attempt) - return response + self._sleep_for_rate_limit(response, rate_limit_round) + rate_limit_round += 1 + + def _request_with_transient_retries( + self, + method: str, + url: str, + *, + params: Mapping[str, Any] | None, + json_data: Any, + extra_headers: Mapping[str, str] | None, + timeout: float, + request_kwargs: Mapping[str, Any], + budget: _RetryBudget, + ) -> requests.Response: + """Retry idempotent GETs on transient 5xx and transport errors. + + Mutating methods are sent once (never duplicated). Streamed GETs bypass + this layer so their consumer owns any stream-open retries. 429s are left + to the rate-limit layer, whose retries re-enter this layer with a fresh + backoff. ``SSLError`` raises immediately: certificate failures are + deterministic, so retrying only delays the report. Every physical send + draws from the shared ``budget`` so the transient, rate-limit, and + auth-refresh layers cannot multiply into a large request count. Each + doubling sleep is capped at ``_MAX_GET_RETRY_BACKOFF_SECONDS`` and + announced through ``self.logger`` (a null logger on non-verbose clients + built without one), so a stalled GET is bounded. + """ + + if method.upper() != "GET" or request_kwargs.get("stream"): + budget.consume() + return self._request_with_same_origin_redirects( + method, + url, + params=params, + json_data=json_data, + extra_headers=extra_headers, + timeout=timeout, + request_kwargs=request_kwargs, + ) + + backoff = self._GET_RETRY_BACKOFF_SECONDS + attempt = 0 + while True: + budget.consume() + attempt += 1 + try: + response = self._request_with_same_origin_redirects( + method, + url, + params=params, + json_data=json_data, + extra_headers=extra_headers, + timeout=timeout, + request_kwargs=request_kwargs, + ) + # Transient transport failures (reset/refused, timeout, truncated or + # corrupt body) can succeed on retry; other request errors surface. + except ( + requests.ConnectionError, + requests.Timeout, + requests.exceptions.ChunkedEncodingError, + requests.exceptions.ContentDecodingError, + ) as exc: + # SSLError subclasses ConnectionError but signals a certificate + # or TLS configuration problem that no retry can fix. + if isinstance(exc, requests.exceptions.SSLError): + raise + # Budget exhausted (attempts or deadline): surface the failure. + if not budget.can_retry(): + raise + self._sleep_for_transient_retry(backoff, attempt, type(exc).__name__) + else: + if response.status_code not in self._RETRYABLE_GET_STATUSES: + return response + # Budget exhausted: return the final 5xx for raise_for_status. + if not budget.can_retry(): + return response + # Release the intermediate response so its connection returns to the pool. + try: + response.close() + except Exception: + pass + self._sleep_for_transient_retry(backoff, attempt, f"HTTP {response.status_code}") + backoff *= 2.0 + + def _sleep_for_transient_retry(self, backoff: float, attempt: int, reason: str) -> None: + delay = min(backoff, self._MAX_GET_RETRY_BACKOFF_SECONDS) + self.logger.warn( + f"transient {reason} on GET; retrying in {delay:.1f}s " + f"(attempt {attempt + 1}/{self._MAX_GET_RETRIES + 1})" + ) + time.sleep(delay) def _sleep_for_rate_limit(self, response: requests.Response, attempt: int) -> None: retry_after = response.headers.get("Retry-After") diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py index 2da0eb5..5394f72 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py @@ -69,7 +69,9 @@ def _allow_all_hydration_for_args(args: ArgsContainer) -> bool: return bool(config.get("allow_all", False)) -def _api_client(args: ArgsContainer, *, cli_base_url: str | None, command_name: str) -> LazyTangleApiClient: +def _api_client( + args: ArgsContainer, *, cli_base_url: str | None, command_name: str, logger: Logger | None = None +) -> LazyTangleApiClient: return LazyTangleApiClient( base_url=args.base_url, token=args.token, @@ -77,12 +79,15 @@ def _api_client(args: ArgsContainer, *, cli_base_url: str | None, command_name: header=args.header, include_env_credentials=include_env_credentials_for_args(args, cli_base_url), command_name=command_name, + logger=logger, ) def _manager(args: ArgsContainer, *, cli_base_url: str | None, logger: Logger) -> PipelineRunManager: return PipelineRunManager( - client=_api_client(args, cli_base_url=cli_base_url, command_name="pipeline-run commands"), + client=_api_client( + args, cli_base_url=cli_base_url, command_name="pipeline-run commands", logger=logger + ), hooks=PipelineRunHooks( logger=logger, trusted_python_sources=_trusted_sources_for_args(args), @@ -117,7 +122,12 @@ def _run_annotation_action(config: str | None, cli_base_url: str | None, specs: raise SystemExit(str(exc)) from exc try: manager = AnnotationManager( - client=_api_client(args, cli_base_url=cli_base_url, command_name="pipeline-run annotation commands"), + client=_api_client( + args, + cli_base_url=cli_base_url, + command_name="pipeline-run annotation commands", + logger=logger, + ), logger=logger, ) print_json(fn(manager, args)) diff --git a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py index 3a30e57..4ce9006 100644 --- a/packages/tangle-cli/src/tangle_cli/pipelines_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipelines_cli.py @@ -232,6 +232,7 @@ def pipelines_hydrate( ), header=_header_entries(header, config_values), include_env_credentials=include_env_credentials, + logger=logger, ), ) except PipelineValidationError as exc: diff --git a/packages/tangle-cli/src/tangle_cli/published_components_cli.py b/packages/tangle-cli/src/tangle_cli/published_components_cli.py index de4f70a..733eb02 100644 --- a/packages/tangle-cli/src/tangle_cli/published_components_cli.py +++ b/packages/tangle-cli/src/tangle_cli/published_components_cli.py @@ -24,7 +24,7 @@ TokenOption, ) from .component_publisher import ComponentPublisher, deprecate_component -from .logger import logger_for_log_type +from .logger import Logger, logger_for_log_type def _client_from_options( @@ -35,6 +35,7 @@ def _client_from_options( header: list[str] | str | None = None, include_env_credentials: bool = True, command_name: str = "published-component commands", + logger: Logger | None = None, ) -> LazyTangleApiClient: """Create a lazy static client proxy for published-component commands. @@ -49,6 +50,7 @@ def _client_from_options( header=header, include_env_credentials=include_env_credentials, command_name=command_name, + logger=logger, ) @@ -97,6 +99,7 @@ def published_components_search( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="published-component commands", + logger=logger, ) if require_available := getattr(client, "require_available", None): require_available() @@ -163,6 +166,7 @@ def published_components_inspect( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="published-component commands", + logger=logger, ) if require_available := getattr(client, "require_available", None): require_available() @@ -219,6 +223,7 @@ def published_components_library( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="published-component commands", + logger=logger, ) if require_available := getattr(client, "require_available", None): require_available() @@ -288,6 +293,7 @@ def published_components_publish( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="published-component commands", + logger=logger, ) publisher = ComponentPublisher( dry_run=bool(args.dry_run), @@ -358,6 +364,7 @@ def published_components_deprecate( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="published-component commands", + logger=logger, ) result = deprecate_component( client, diff --git a/packages/tangle-cli/src/tangle_cli/secrets_cli.py b/packages/tangle-cli/src/tangle_cli/secrets_cli.py index d6c2da1..48ba9ff 100644 --- a/packages/tangle-cli/src/tangle_cli/secrets_cli.py +++ b/packages/tangle-cli/src/tangle_cli/secrets_cli.py @@ -58,7 +58,9 @@ app = App(name="secrets", help="Manage Tangle secrets.") -def _client(args: ArgsContainer, *, cli_base_url: str | None, command_name: str) -> LazyTangleApiClient: +def _client( + args: ArgsContainer, *, cli_base_url: str | None, command_name: str, logger: Logger | None = None +) -> LazyTangleApiClient: return LazyTangleApiClient( base_url=args.base_url, token=args.token, @@ -66,6 +68,7 @@ def _client(args: ArgsContainer, *, cli_base_url: str | None, command_name: str) header=args.header, include_env_credentials=include_env_credentials_for_args(args, cli_base_url), command_name=command_name, + logger=logger, ) @@ -79,7 +82,7 @@ def _run_secret_action( for args in load_args_or_exit(config, **specs): logger, finalize_logs = logger_for_log_type(getattr(args, "log_type", "console")) try: - client = _client(args, cli_base_url=cli_base_url, command_name="secret commands") + client = _client(args, cli_base_url=cli_base_url, command_name="secret commands", logger=logger) try: results.append(fn(client, args, logger)) except SecretValueError as exc: diff --git a/tests/test_api_cli.py b/tests/test_api_cli.py index c93a575..8993cc6 100644 --- a/tests/test_api_cli.py +++ b/tests/test_api_cli.py @@ -1,6 +1,7 @@ import importlib import json import sys +from unittest.mock import ANY import httpx import pytest @@ -460,6 +461,7 @@ def fake_client_from_options(**kwargs): "header": ["X-Config: yes"], "include_env_credentials": False, "command_name": "published-component commands", + "logger": ANY, } diff --git a/tests/test_artifacts_cli.py b/tests/test_artifacts_cli.py index b7b3ec1..f635497 100644 --- a/tests/test_artifacts_cli.py +++ b/tests/test_artifacts_cli.py @@ -5,6 +5,7 @@ import json import sys from typing import Any +from unittest.mock import ANY from tangle_cli import artifacts as artifacts_module from tangle_cli import artifacts_cli, cli @@ -80,6 +81,7 @@ def fake_get_artifacts(self, run_id: str, query: dict[str, Any]) -> dict[str, ob "header": ["X-Config: yes"], "include_env_credentials": False, "command_name": "artifact commands", + "logger": ANY, } ] assert get_calls == [ diff --git a/tests/test_client.py b/tests/test_client.py index 5456ac0..8f21cd9 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -1,13 +1,46 @@ from __future__ import annotations +import json from types import SimpleNamespace +from typing import Any from unittest.mock import MagicMock +import pytest +import requests + import tangle_cli.client as client_module from tangle_cli.client import TangleApiClient +from tangle_cli.logger import CaptureLogger from tangle_cli.models import ComponentInfo +def _response(payload: Any = None, status_code: int = 200) -> requests.Response: + r = requests.Response() + r.status_code = status_code + if payload is None: + r._content = b"" + else: + r._content = json.dumps(payload).encode("utf-8") + r.headers["Content-Type"] = "application/json" + r.request = requests.Request("GET", "https://api.test").prepare() + return r + + +class _FakeSession: + def __init__(self, responses: list[requests.Response | Exception] | None = None) -> None: + self.calls: list[dict[str, Any]] = [] + self.responses = responses or [] + + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + self.calls.append({"method": method, "url": url, **kwargs}) + if self.responses: + next_response = self.responses.pop(0) + if isinstance(next_response, Exception): + raise next_response + return next_response + return _response({}) + + def test_find_existing_components_matches_exact_names_case_insensitively() -> None: client = TangleApiClient("https://api.test") client.list_published_component_infos = MagicMock( @@ -53,3 +86,409 @@ def fail_from_dict(*args, **kwargs): assert client.get_run_pipeline_spec("run-1") is task_spec client.executions_details.assert_called_once_with("root-exec-1") + + +def test_get_retries_transient_5xx_then_succeeds(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + ok = _response({"ok": True}) + session = _FakeSession([_response(status_code=503), _response(status_code=500), ok]) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result is ok + assert len(session.calls) == 3 + assert sleeps == [1.0, 2.0] + + +def test_get_closes_intermediate_5xx_responses_before_retrying(monkeypatch) -> None: + events: list[str] = [] + + def tracking_response(status_code: int, marker: str) -> requests.Response: + r = _response(status_code=status_code) + r.close = lambda: events.append(f"close-{marker}") # type: ignore[method-assign] + return r + + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: events.append("sleep")) + session = _FakeSession( + [tracking_response(503, "1"), tracking_response(500, "2"), tracking_response(200, "3")] + ) + client = TangleApiClient("https://api.test", session=session) + + client._make_request("GET", "/api/test") + + # Intermediate 5xx are closed before each retry; the returned one is left open. + assert events == ["close-1", "sleep", "close-2", "sleep"] + + +def test_get_retries_transport_error_then_succeeds(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + ok = _response({"ok": True}) + session = _FakeSession( + [ + requests.ConnectionError("connection reset"), + requests.Timeout("read timed out"), + requests.exceptions.ChunkedEncodingError("incomplete chunked read"), + requests.exceptions.ContentDecodingError("failed to decode gzip stream"), + ok, + ] + ) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result is ok + assert len(session.calls) == 5 + assert sleeps == [1.0, 2.0, 4.0, 8.0] + + +def test_get_raises_after_exhausting_transport_retries(monkeypatch) -> None: + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + budget = TangleApiClient._MAX_GET_RETRIES + final_attempt_error = requests.exceptions.ChunkedEncodingError("final permitted attempt") + surplus = [requests.ConnectionError("never reached") for _ in range(3)] + queued = [requests.ConnectionError("blip") for _ in range(budget)] + [final_attempt_error] + surplus + assert len(queued) > budget + 1 + session = _FakeSession(queued) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.exceptions.ChunkedEncodingError) as exc_info: + client._make_request("GET", "/api/test") + + assert exc_info.value is final_attempt_error + assert len(session.calls) == budget + 1 + assert len(session.responses) == len(surplus) + + +def test_get_returns_final_5xx_after_exhausting_status_retries(monkeypatch) -> None: + closed: list[str] = [] + + def tracking_5xx(marker: str) -> requests.Response: + r = _response(status_code=503) + r.close = lambda: closed.append(marker) # type: ignore[method-assign] + return r + + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + budget = TangleApiClient._MAX_GET_RETRIES + errors = [tracking_5xx(str(i)) for i in range(budget + 1)] + trailing_ok = _response({"ok": True}) + session = _FakeSession([*errors, trailing_ok]) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result is errors[budget] + assert len(session.calls) == budget + 1 + assert closed == [str(i) for i in range(budget)] + assert session.responses == [trailing_ok] + + +def test_post_is_not_retried_on_transient_5xx(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + server_error = _response(status_code=503) + session = _FakeSession([server_error, _response({"ok": True})]) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("POST", "/api/pipeline_runs/", json_data={"a": 1}) + + assert result is server_error + assert len(session.calls) == 1 + assert sleeps == [] + + +def test_streamed_get_bypasses_transient_retry(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + server_error = _response(status_code=503) + session = _FakeSession([server_error, _response({"ok": True})]) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/logs", stream=True) + + assert result is server_error + assert len(session.calls) == 1 + assert sleeps == [] + + +def test_transient_retry_decision_is_method_case_insensitive(monkeypatch) -> None: + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + + def call(method: str) -> tuple[requests.Response, int]: + ok = _response({"ok": True}) + session = _FakeSession([_response(status_code=503), ok]) + client = TangleApiClient("https://api.test", session=session) + budget = client_module._RetryBudget( + client._MAX_GET_RETRIES + 1, + client_module.time.monotonic() + client._MAX_RETRY_ELAPSED_SECONDS, + ) + result = client._request_with_transient_retries( + method, + "https://api.test/api/test", + params=None, + json_data=None, + extra_headers=None, + timeout=client.timeout, + request_kwargs={}, + budget=budget, + ) + return result, len(session.calls) + + get_result, get_calls = call("get") + assert get_result.status_code == 200 + assert get_calls == 2 + + post_result, post_calls = call("post") + assert post_result.status_code == 503 + assert post_calls == 1 + + +def test_get_retries_proxy_errors(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + ok = _response({"ok": True}) + session = _FakeSession( + [ + requests.exceptions.ProxyError("proxy refused"), + ok, + ] + ) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result is ok + assert len(session.calls) == 2 + assert sleeps == [1.0] + + +def test_get_does_not_retry_ssl_errors(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + error = requests.exceptions.SSLError("certificate verify failed") + session = _FakeSession([error, _response({"ok": True})]) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.exceptions.SSLError) as exc_info: + client._make_request("GET", "/api/test") + + assert exc_info.value is error + assert len(session.calls) == 1 + assert sleeps == [] + + +def test_get_transient_and_rate_limit_retry_layers_compose(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + ok = _response({"ok": True}) + session = _FakeSession( + [ + _response(status_code=503), + _response(status_code=429), + _response(status_code=503), + ok, + ] + ) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result is ok + assert len(session.calls) == 4 + # transient 1.0s, rate-limit 1.0s (no Retry-After), fresh transient 1.0s + assert sleeps == [1.0, 1.0, 1.0] + + +def test_get_retry_sleeps_are_capped_and_announced_without_verbose(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append) + budget = TangleApiClient._MAX_GET_RETRIES + session = _FakeSession([_response(status_code=503) for _ in range(budget + 1)]) + logger = CaptureLogger() + client = TangleApiClient("https://api.test", session=session, logger=logger) + + result = client._make_request("GET", "/api/test") + + assert result.status_code == 503 + assert sleeps == [1.0, 2.0, 4.0, 8.0, 16.0, 30.0] # final sleep capped, not 32.0 + messages = (logger.get_logs() or "").splitlines() + assert len(messages) == budget + assert all(m.startswith("transient HTTP 503 on GET; retrying in ") for m in messages) + assert messages[-1] == "transient HTTP 503 on GET; retrying in 30.0s (attempt 7/7)" + + +def test_get_retries_are_silent_on_default_non_verbose_client(monkeypatch, capsys) -> None: + # A non-verbose client built without a logger stays silent; callers that + # want retry announcements pass a logger (as the CLI command layer does). + monkeypatch.delenv("TANGLE_VERBOSE", raising=False) + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + session = _FakeSession([_response(status_code=503), _response({"ok": True})]) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result.status_code == 200 + captured = capsys.readouterr() + assert captured.out == "" + assert captured.err == "" + + +def test_shared_budget_caps_total_requests_across_transient_and_rate_limit(monkeypatch) -> None: + # Interleaved 503/429 responses must not let the rate-limit layer hand the + # transient layer a fresh budget each round: the total physical request + # count is bounded by the single shared budget, not their product. + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + budget = TangleApiClient._MAX_GET_RETRIES + # Far more responses than the budget allows, alternating retryable states. + session = _FakeSession([_response(status_code=503 if i % 2 == 0 else 429) for i in range(40)]) + client = TangleApiClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + # Exactly the advertised budget of physical requests is spent, then the last + # response surfaces for the caller's raise_for_status (no amplification). + assert len(session.calls) == budget + 1 + assert result.status_code in {503, 429} + + +def test_auth_refresh_shares_transient_retry_budget(monkeypatch) -> None: + # A 401 that triggers an auth refresh must continue on the same budget + # rather than starting a fresh transient-retry round. + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + + class RefreshingClient(TangleApiClient): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.refreshes = 0 + + def _refresh_auth(self) -> None: + self.refreshes += 1 + + budget = TangleApiClient._MAX_GET_RETRIES + # One 401 (consumes a request) followed by an unbroken run of 503s. + session = _FakeSession( + [_response(status_code=401)] + [_response(status_code=503) for _ in range(budget + 5)] + ) + client = RefreshingClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + assert result.status_code == 503 + # Refresh fired once for the 401 (plus the unconditional pre-request refresh). + assert client.refreshes == 2 + # The 401 request plus the post-refresh retries share one budget: the total + # never exceeds the shared cap (a fresh budget would allow budget+1 more). + assert len(session.calls) == budget + 1 + + +def test_shared_budget_caps_total_requests_across_transient_rate_limit_and_auth(monkeypatch) -> None: + # The worst case the reviewer flagged: a 401 auth refresh, 429 rate limits, + # and transient 503s all interleaved for one logical GET. A single shared + # budget must bound the total physical request count instead of letting the + # three layers multiply their per-layer limits together. + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + + class RefreshingClient(TangleApiClient): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.refreshes = 0 + + def _refresh_auth(self) -> None: + self.refreshes += 1 + + budget = TangleApiClient._MAX_GET_RETRIES + # Lead with a 401 (drives one refresh), then alternate 429/503 far past the + # budget so the cap, not the response list, is what stops the retries. + responses = [_response(status_code=401)] + responses += [_response(status_code=429 if i % 2 == 0 else 503) for i in range(40)] + session = _FakeSession(responses) + client = RefreshingClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + # Exactly the advertised shared budget of physical requests is spent. + assert len(session.calls) == budget + 1 + assert result.status_code in {429, 503} + # Pre-request refresh plus exactly one refresh for the single 401; the 401's + # retry continues on the shared budget rather than opening a fresh round. + assert client.refreshes == 2 + + +def test_shared_budget_deadline_halts_composed_retries_without_wallclock(monkeypatch) -> None: + # The shared budget bounds retries by BOTH an attempt count and a + # _MAX_RETRY_ELAPSED_SECONDS wall-time deadline. This pins the deadline + # clause deterministically: a fake monotonic clock advances only when a + # request is sent (never the real wall clock), so the elapsed-time cap, not + # the attempt cap, is what stops the composed auth-refresh / rate-limit / + # transient retry sequence. Without this every other budget test would still + # pass on the attempt cap alone, so the deadline could be removed silently. + clock = SimpleNamespace(now=1_000.0) + step = 50.0 + window = TangleApiClient._MAX_RETRY_ELAPSED_SECONDS + deadline = clock.now + window + monkeypatch.setattr("tangle_cli.client.time.monotonic", lambda: clock.now) + + send_times: list[float] = [] + sleep_times: list[float] = [] + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: sleep_times.append(clock.now)) + + class _ClockAdvancingSession(_FakeSession): + def request(self, method: str, url: str, **kwargs: Any) -> requests.Response: + send_times.append(clock.now) + response = super().request(method, url, **kwargs) + # Time elapses only while a request is in flight; each send consumes + # a fixed slice of the deadline window. + clock.now += step + return response + + class RefreshingClient(TangleApiClient): + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self.refreshes = 0 + + def _refresh_auth(self) -> None: + self.refreshes += 1 + + # A 401 (auth-refresh path), a 429 (rate-limit path), then unbroken 503s + # (transient path): all three layers draw on the one shared budget. Far more + # responses are queued than either cap allows, so the cap that fires first is + # what stops the sequence. + responses = [_response(status_code=401), _response(status_code=429)] + responses += [_response(status_code=503) for _ in range(20)] + session = _ClockAdvancingSession(responses) + client = RefreshingClient("https://api.test", session=session) + + result = client._make_request("GET", "/api/test") + + # The sequence genuinely retried across layers but stopped short of the + # attempt cap with responses still queued: the deadline, not the attempt + # count or the response list, is what ended it. + attempt_cap = TangleApiClient._MAX_GET_RETRIES + 1 + assert 1 < len(session.calls) < attempt_cap + assert session.responses, "unused responses prove the queue did not stop the retries" + # Time actually crossed the deadline, yet no send or sleep happened at/after + # it: can_retry gates every physical send and every sleep across the + # transient and rate-limit layers. + assert clock.now >= deadline + assert all(t < deadline for t in send_times) + assert all(t < deadline for t in sleep_times) + # The composed auth path ran on the shared budget (pre-request refresh plus + # one for the single 401), and the final 5xx surfaces for the caller. + assert client.refreshes == 2 + assert result.status_code == 503 + + +def test_get_5xx_exhaustion_raises_http_error_from_public_operation(monkeypatch) -> None: + monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _delay: None) + budget = TangleApiClient._MAX_GET_RETRIES + session = _FakeSession([_response(status_code=503) for _ in range(budget + 1)]) + client = TangleApiClient("https://api.test", session=session) + + with pytest.raises(requests.HTTPError) as exc_info: + client.pipeline_runs_get("run-1") + + assert exc_info.value.response is not None + assert exc_info.value.response.status_code == 503 + assert len(session.calls) == budget + 1 diff --git a/tests/test_components_cli.py b/tests/test_components_cli.py index 5764695..c52b5d8 100644 --- a/tests/test_components_cli.py +++ b/tests/test_components_cli.py @@ -138,6 +138,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "header": ["X-Config: yes"], "include_env_credentials": False, "command_name": "published-component commands", + "logger": ANY, } ] assert FakePublisher.instances[0].kwargs["client"] is fake_client @@ -271,6 +272,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "header": None, "include_env_credentials": False, "command_name": "published-component commands", + "logger": ANY, } ] assert [publisher.kwargs for publisher in FakePublisher.instances] == [ @@ -341,6 +343,7 @@ def fake_deprecate_component(client: object, digest: str, **kwargs: Any) -> dict "header": ["X-Test: yes"], "include_env_credentials": False, "command_name": "published-component commands", + "logger": ANY, } ] assert deprecate_calls == [ diff --git a/tests/test_secrets_cli.py b/tests/test_secrets_cli.py index 6072131..f61b17f 100644 --- a/tests/test_secrets_cli.py +++ b/tests/test_secrets_cli.py @@ -6,6 +6,7 @@ import sys from types import SimpleNamespace from typing import Any +from unittest.mock import ANY import pytest @@ -341,6 +342,7 @@ def test_sdk_secrets_config_array_and_config_base_url_credential_isolation( "header": ["X-Config: yes"], "include_env_credentials": False, "command_name": "secret commands", + "logger": ANY, }, { "base_url": "https://config.example", @@ -349,6 +351,7 @@ def test_sdk_secrets_config_array_and_config_base_url_credential_isolation( "header": ["X-Config: yes"], "include_env_credentials": False, "command_name": "secret commands", + "logger": ANY, }, ] assert [instance.calls[0]["secret_name"] for instance in FakeLazyTangleApiClient.instances] == [