feat(client): retry transient GET failures with backoff

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