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.

Author: arseniy-pplx

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()

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,22 +196,117 @@ 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,
                 json_data=json_data,
                 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")

packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py

@@ -69,20 +69,25 @@ 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,
         auth_header=args.auth_header,
         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))

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:

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,

packages/tangle-cli/src/tangle_cli/secrets_cli.py

@@ -58,14 +58,17 @@
 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,
         auth_header=args.auth_header,
         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:

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,
     }
 
 

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 == [