refactor(http-client): extract shared backoff helpers as module-level functions

## Purpose

`_compute_backoff`, `_compute_retry_after_delay`, and `_notify_throttle` were
duplicated identically between `_RetryTransport` and `_AsyncRetryTransport`.
Any future change to the decorrelated-jitter formula or Retry-After parsing
had to be applied in two places, with no enforcement that they stayed in sync.

## Solution

Extracted all three as module-level functions that accept `config: RetryConfig`
as their first argument. Both transport classes now call the shared functions
instead of their own instance methods. Updated existing jitter-distribution tests
in `test_http_client.py` (which were calling the instance methods directly) and
added dedicated unit tests for each helper in `test_retry_transport.py`.

Author: jhamon

pinecone/_internal/http_client.py

@@ -118,6 +118,43 @@ def _log_curl(
     logger.debug("curl equivalent:\n%s", curl_cmd)
 
 
+def _compute_backoff(config: RetryConfig, attempt: int, prev_delay: float | None) -> float:
+    """Decorrelated jitter: uniform(base, prev*3), capped at max_wait."""
+    base_delay = config.backoff_factor
+    if prev_delay is None:
+        prev_delay = base_delay
+    upper = min(config.max_wait, prev_delay * 3.0)
+    return random.uniform(base_delay, max(base_delay, upper))
+
+
+def _compute_retry_after_delay(
+    config: RetryConfig,
+    response: httpx.Response,
+    attempt: int,
+    prev_delay: float | None,
+) -> float:
+    retry_after = response.headers.get("retry-after")
+    if retry_after is not None:
+        try:
+            ra = float(retry_after)
+            if ra >= 0:
+                smear = random.uniform(0.0, ra * 0.5)
+                return ra + smear
+        except (ValueError, TypeError):
+            pass
+    return _compute_backoff(config, attempt, prev_delay)
+
+
+def _notify_throttle(config: RetryConfig, request: httpx.Request) -> None:
+    cb = config.on_throttle
+    if cb is None:
+        return
+    try:
+        cb(request.url.host)
+    except Exception as exc:
+        logger.debug("on_throttle callback raised, ignoring: %s", exc)
+
+
 class _RetryTransport(httpx.BaseTransport):
     """Sync transport wrapper that retries on transient server errors."""
 
@@ -145,17 +182,17 @@ def handle_request(self, request: httpx.Request) -> httpx.Response:
                         self._config.max_retries + 1,
                         exc,
                     )
-                    delay = self._compute_backoff(attempt, prev_delay)
+                    delay = _compute_backoff(self._config, attempt, prev_delay)
                     prev_delay = delay
                     time.sleep(delay)
                 continue
             last_exc = None
             if response.status_code not in self._config.retryable_status_codes:
                 return response
-            self._notify_throttle(request)
+            _notify_throttle(self._config, request)
             if attempt < self._config.max_retries:
                 response.close()
-                delay = self._compute_retry_after_delay(response, attempt, prev_delay)
+                delay = _compute_retry_after_delay(self._config, response, attempt, prev_delay)
                 prev_delay = delay
                 time.sleep(delay)
             else:
@@ -164,40 +201,6 @@ def handle_request(self, request: httpx.Request) -> httpx.Response:
             raise last_exc
         raise RuntimeError("max_retries must be non-negative")
 
-    def _compute_backoff(self, attempt: int, prev_delay: float | None) -> float:
-        """Decorrelated jitter: uniform(base, prev*3), capped at max_wait."""
-        base_delay = self._config.backoff_factor
-        if prev_delay is None:
-            prev_delay = base_delay
-        upper = min(self._config.max_wait, prev_delay * 3.0)
-        return random.uniform(base_delay, max(base_delay, upper))
-
-    def _compute_retry_after_delay(
-        self,
-        response: httpx.Response,
-        attempt: int,
-        prev_delay: float | None,
-    ) -> float:
-        retry_after = response.headers.get("retry-after")
-        if retry_after is not None:
-            try:
-                ra = float(retry_after)
-                if ra >= 0:
-                    smear = random.uniform(0.0, ra * 0.5)
-                    return ra + smear
-            except (ValueError, TypeError):
-                pass
-        return self._compute_backoff(attempt, prev_delay)
-
-    def _notify_throttle(self, request: httpx.Request) -> None:
-        cb = self._config.on_throttle
-        if cb is None:
-            return
-        try:
-            cb(request.url.host)
-        except Exception as exc:
-            logger.debug("on_throttle callback raised, ignoring: %s", exc)
-
     def close(self) -> None:
         self._transport.close()
 
@@ -229,17 +232,17 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
                         self._config.max_retries + 1,
                         exc,
                     )
-                    delay = self._compute_backoff(attempt, prev_delay)
+                    delay = _compute_backoff(self._config, attempt, prev_delay)
                     prev_delay = delay
                     await asyncio.sleep(delay)
                 continue
             last_exc = None
             if response.status_code not in self._config.retryable_status_codes:
                 return response
-            self._notify_throttle(request)
+            _notify_throttle(self._config, request)
             if attempt < self._config.max_retries:
                 await response.aclose()
-                delay = self._compute_retry_after_delay(response, attempt, prev_delay)
+                delay = _compute_retry_after_delay(self._config, response, attempt, prev_delay)
                 prev_delay = delay
                 await asyncio.sleep(delay)
             else:
@@ -248,40 +251,6 @@ async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
             raise last_exc
         raise RuntimeError("max_retries must be non-negative")
 
-    def _compute_backoff(self, attempt: int, prev_delay: float | None) -> float:
-        """Decorrelated jitter: uniform(base, prev*3), capped at max_wait."""
-        base_delay = self._config.backoff_factor
-        if prev_delay is None:
-            prev_delay = base_delay
-        upper = min(self._config.max_wait, prev_delay * 3.0)
-        return random.uniform(base_delay, max(base_delay, upper))
-
-    def _compute_retry_after_delay(
-        self,
-        response: httpx.Response,
-        attempt: int,
-        prev_delay: float | None,
-    ) -> float:
-        retry_after = response.headers.get("retry-after")
-        if retry_after is not None:
-            try:
-                ra = float(retry_after)
-                if ra >= 0:
-                    smear = random.uniform(0.0, ra * 0.5)
-                    return ra + smear
-            except (ValueError, TypeError):
-                pass
-        return self._compute_backoff(attempt, prev_delay)
-
-    def _notify_throttle(self, request: httpx.Request) -> None:
-        cb = self._config.on_throttle
-        if cb is None:
-            return
-        try:
-            cb(request.url.host)
-        except Exception as exc:
-            logger.debug("on_throttle callback raised, ignoring: %s", exc)
-
     async def aclose(self) -> None:
         await self._transport.aclose()
 

tests/unit/_internal/test_retry_transport.py

@@ -8,7 +8,13 @@
 import pytest
 
 from pinecone._internal.config import RetryConfig
-from pinecone._internal.http_client import _AsyncRetryTransport, _RetryTransport
+from pinecone._internal.http_client import (
+    _AsyncRetryTransport,
+    _compute_backoff,
+    _compute_retry_after_delay,
+    _notify_throttle,
+    _RetryTransport,
+)
 
 
 def _transport(max_retries: int = 3) -> tuple[_RetryTransport, MagicMock]:
@@ -218,3 +224,107 @@ async def test_async_post_retried_on_408() -> None:
     result = await rt.handle_async_request(httpx.Request("POST", "https://example.com/query"))
     assert result.status_code == 200
     assert inner.handle_async_request.call_count == 2
+
+
+# --- module-level helper function tests ---
+
+
+def _cfg(backoff_factor: float = 0.5, max_wait: float = 60.0) -> RetryConfig:
+    return RetryConfig(backoff_factor=backoff_factor, max_wait=max_wait)
+
+
+def test_compute_backoff_first_attempt_stays_at_base() -> None:
+    cfg = _cfg(backoff_factor=0.5, max_wait=60.0)
+    result = _compute_backoff(cfg, 0, None)
+    assert 0.5 <= result <= 1.5  # uniform(base, max(base, min(max_wait, base*3)))
+
+
+def test_compute_backoff_grows_with_prev_delay() -> None:
+    cfg = _cfg(backoff_factor=0.5, max_wait=60.0)
+    result = _compute_backoff(cfg, 1, 2.0)
+    # upper = min(60, 2.0 * 3) = 6.0; result in [0.5, 6.0]
+    assert 0.5 <= result <= 6.0
+
+
+def test_compute_backoff_capped_at_max_wait() -> None:
+    cfg = _cfg(backoff_factor=1.0, max_wait=2.0)
+    result = _compute_backoff(cfg, 5, 100.0)
+    # upper = min(2.0, 300.0) = 2.0; result in [1.0, 2.0]
+    assert 1.0 <= result <= 2.0
+
+
+def test_compute_retry_after_uses_header_value() -> None:
+    cfg = _cfg()
+    response = httpx.Response(429, headers={"Retry-After": "5"})
+    result = _compute_retry_after_delay(cfg, response, 0, None)
+    # Should be between 5 (no smear) and 7.5 (5 + 5*0.5)
+    assert 5.0 <= result <= 7.5
+
+
+def test_compute_retry_after_ignores_invalid_header() -> None:
+    cfg = _cfg(backoff_factor=0.5, max_wait=60.0)
+    response = httpx.Response(429, headers={"Retry-After": "not-a-number"})
+    result = _compute_retry_after_delay(cfg, response, 0, None)
+    # Falls back to _compute_backoff
+    assert result >= 0.5
+
+
+def test_compute_retry_after_no_header_falls_back_to_backoff() -> None:
+    cfg = _cfg(backoff_factor=0.5, max_wait=60.0)
+    response = httpx.Response(503)
+    result = _compute_retry_after_delay(cfg, response, 0, None)
+    assert result >= 0.5
+
+
+def test_notify_throttle_calls_callback() -> None:
+    calls: list[str] = []
+    cfg = RetryConfig(on_throttle=lambda host: calls.append(host))
+    request = httpx.Request("POST", "https://example.com/vectors/upsert")
+    _notify_throttle(cfg, request)
+    assert calls == ["example.com"]
+
+
+def test_notify_throttle_no_callback_is_noop() -> None:
+    cfg = RetryConfig(on_throttle=None)
+    request = httpx.Request("POST", "https://example.com/query")
+    _notify_throttle(cfg, request)  # should not raise
+
+
+def test_notify_throttle_swallows_callback_exception() -> None:
+    def bad_cb(host: str) -> None:
+        raise RuntimeError("oops")
+
+    cfg = RetryConfig(on_throttle=bad_cb)
+    request = httpx.Request("POST", "https://example.com/query")
+    _notify_throttle(cfg, request)  # should not raise
+
+
+def test_sync_transport_calls_module_level_compute_backoff() -> None:
+    """Verify sync transport uses module-level _compute_backoff (not an instance method)."""
+    rt, inner = _transport(max_retries=1)
+    inner.handle_request.side_effect = [
+        httpx.ConnectError("boom"),
+        httpx.Response(200),
+    ]
+    with patch("pinecone._internal.http_client._compute_backoff", return_value=0.001) as mock_cb:
+        rt.handle_request(_req())
+    mock_cb.assert_called_once()
+    args = mock_cb.call_args[0]
+    assert isinstance(args[0], RetryConfig)
+
+
+def test_async_transport_calls_module_level_compute_backoff() -> None:
+    """Verify async transport uses module-level _compute_backoff (not an instance method)."""
+    import asyncio
+
+    rt, inner = _async_transport(max_retries=1)
+    inner.handle_async_request.side_effect = [
+        httpx.ConnectError("boom"),
+        httpx.Response(200),
+    ]
+
+    with patch("pinecone._internal.http_client._compute_backoff", return_value=0.001) as mock_cb:
+        asyncio.get_event_loop().run_until_complete(rt.handle_async_request(_req()))
+    mock_cb.assert_called_once()
+    args = mock_cb.call_args[0]
+    assert isinstance(args[0], RetryConfig)

tests/unit/test_http_client.py

@@ -18,6 +18,8 @@
     HTTPClient,
     _AsyncRetryTransport,
     _build_headers,
+    _compute_backoff,
+    _compute_retry_after_delay,
     _encode_json,
     _log_curl,
     _prepare_json_kwargs,
@@ -895,7 +897,7 @@ def test_backoff_first_attempt_delay_range(self) -> None:
 
         random.seed(42)
         t = _make_sync_retry_transport()
-        delays = [t._compute_backoff(0, None) for _ in range(200)]
+        delays = [_compute_backoff(t._config, 0, None) for _ in range(200)]
         assert all(0.25 <= d <= 0.75 for d in delays), (
             f"out-of-range: {[d for d in delays if not (0.25 <= d <= 0.75)]}"
         )
@@ -908,26 +910,24 @@ def test_backoff_first_attempt_delay_range(self) -> None:
 
     def test_backoff_evolves_with_prev_delay(self) -> None:
         random.seed(42)
-        t = _make_sync_retry_transport()
         cfg = RetryConfig(max_wait=60.0)
-        t._config = cfg
-        delays = [t._compute_backoff(2, 5.0) for _ in range(200)]
+        delays = [_compute_backoff(cfg, 2, 5.0) for _ in range(200)]
         # Decorrelated jitter: uniform(0.25, min(60.0, 5.0*3)) = uniform(0.25, 15.0)
         assert all(0.25 <= d <= 15.0 for d in delays)
         assert any(d > 5.0 for d in delays), "upper bound did not expand with prev_delay"
 
     def test_backoff_capped_at_max_wait(self) -> None:
         random.seed(42)
-        t = _make_sync_retry_transport(RetryConfig(max_wait=10.0))
-        delays = [t._compute_backoff(0, 100.0) for _ in range(50)]
+        cfg = RetryConfig(max_wait=10.0)
+        delays = [_compute_backoff(cfg, 0, 100.0) for _ in range(50)]
         # prev_delay=100 would push upper to 300, but max_wait caps at 10.0
         assert all(d <= 10.0 for d in delays)
 
     def test_retry_after_smear_range(self) -> None:
         random.seed(42)
         t = _make_sync_retry_transport()
         response = httpx.Response(429, headers={"retry-after": "60"})
-        delays = [t._compute_retry_after_delay(response, 0, None) for _ in range(200)]
+        delays = [_compute_retry_after_delay(t._config, response, 0, None) for _ in range(200)]
         # Smear: delay in [60, 60 + 0.5*60) = [60, 90)
         assert all(60.0 <= d < 90.0 for d in delays), (
             f"out-of-range delays: {[d for d in delays if not (60.0 <= d < 90.0)]}"
@@ -939,23 +939,23 @@ def test_retry_after_falls_back_to_backoff_when_invalid(self) -> None:
         random.seed(42)
         t = _make_sync_retry_transport()
         response = httpx.Response(429, headers={"retry-after": "Fri, 31 Dec 2026 23:59:59 GMT"})
-        delays = [t._compute_retry_after_delay(response, 0, None) for _ in range(50)]
+        delays = [_compute_retry_after_delay(t._config, response, 0, None) for _ in range(50)]
         # HTTP-date is not parseable as float; falls back to backoff which is uniform(0.25, 0.75)
         assert all(0.25 <= d <= 0.75 for d in delays)
 
     def test_retry_after_falls_back_to_backoff_when_missing(self) -> None:
         random.seed(42)
         t = _make_sync_retry_transport()
         response = httpx.Response(500)
-        delays = [t._compute_retry_after_delay(response, 0, None) for _ in range(50)]
+        delays = [_compute_retry_after_delay(t._config, response, 0, None) for _ in range(50)]
         # Falls back to backoff: uniform(0.25, 0.75)
         assert all(0.25 <= d <= 0.75 for d in delays)
 
     def test_negative_retry_after_falls_back_to_backoff(self) -> None:
         random.seed(42)
         t = _make_sync_retry_transport()
         response = httpx.Response(429, headers={"retry-after": "-1"})
-        delays = [t._compute_retry_after_delay(response, 0, None) for _ in range(50)]
+        delays = [_compute_retry_after_delay(t._config, response, 0, None) for _ in range(50)]
         # Negative values are ignored; falls back to backoff: uniform(0.25, 0.75)
         assert all(0.25 <= d <= 0.75 for d in delays)
 
@@ -971,7 +971,7 @@ def test_backoff_first_attempt_delay_range(self) -> None:
 
         random.seed(42)
         t = _make_async_retry_transport()
-        delays = [t._compute_backoff(0, None) for _ in range(200)]
+        delays = [_compute_backoff(t._config, 0, None) for _ in range(200)]
         assert all(0.25 <= d <= 0.75 for d in delays), (
             f"out-of-range: {[d for d in delays if not (0.25 <= d <= 0.75)]}"
         )
@@ -986,7 +986,7 @@ def test_retry_after_smear_range(self) -> None:
         random.seed(42)
         t = _make_async_retry_transport()
         response = httpx.Response(429, headers={"retry-after": "60"})
-        delays = [t._compute_retry_after_delay(response, 0, None) for _ in range(200)]
+        delays = [_compute_retry_after_delay(t._config, response, 0, None) for _ in range(200)]
         # Smear: delay in [60, 60 + 0.5*60) = [60, 90)
         assert all(60.0 <= d < 90.0 for d in delays), (
             f"out-of-range delays: {[d for d in delays if not (60.0 <= d < 90.0)]}"