feat(adaptive): add _AdaptiveLimiter and _AdaptiveLimiterRegistry (AIMD)

## Purpose

Bulk paths need to self-tune effective concurrency in response to
throttling (429s) without exposing new public config. This implements
Phase C of the retry-resilience plan: per-host AIMD state managed
internally by the SDK.

## Solution

New module `pinecone/_internal/adaptive.py` with two classes:

- `_AdaptiveLimiter`: per-host AIMD state with `threading.Lock`.
  Multiplicative decrease on throttle (`limit = max(1, limit // 2)`),
  additive increase after `current_limit` consecutive successes.
- `_AdaptiveLimiterRegistry`: per-client dict with lazy creation and
  `report_throttled(host)` shortcut for transport callbacks.

Hardcoded AIMD parameters (no public config). Thread-safe via stdlib
lock; critical sections are O(1) so no async lock needed.

## Follow-ups

Wire registry into bulk paths (Phase C2/C3) and transport callbacks
(Phase C4/C5) in subsequent tasks (DX-0157, DX-0158, etc.).

Author: jhamon

pinecone/_internal/adaptive.py

@@ -0,0 +1,117 @@
+"""Per-host adaptive concurrency limiter (AIMD).
+
+Internal to the SDK. The transport calls ``report_throttled(host)`` on every
+retryable response; the bulk paths read ``current_limit(host, ceiling)``
+before dispatching work. The limiter self-tunes effective concurrency
+between ``1`` and the user-provided ``max_concurrency`` ceiling.
+
+Not thread-coordinated across processes. See ``docs/guides/retries.md``
+for multi-process guidance.
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    pass
+
+
+class _AdaptiveLimiter:
+    """AIMD state for a single host.
+
+    Multiplicative decrease on throttle: ``limit = max(1, limit // 2)``.
+    Additive increase on success: ``limit = min(ceiling, limit + 1)``
+    after ``current_limit`` consecutive successes. The success counter
+    resets on each throttle.
+    """
+
+    __slots__ = ("_ceiling", "_limit", "_lock", "_success_streak")
+
+    def __init__(self, ceiling: int) -> None:
+        if ceiling < 1:
+            raise ValueError(f"ceiling must be >= 1, got {ceiling}")
+        self._lock = threading.Lock()
+        self._ceiling = ceiling
+        self._limit = ceiling
+        self._success_streak = 0
+
+    @property
+    def ceiling(self) -> int:
+        return self._ceiling
+
+    def current_limit(self) -> int:
+        """Return the current effective concurrency limit (1 <= limit <= ceiling)."""
+        return self._limit
+
+    def report_throttled(self) -> None:
+        """Halve the limit (floored at 1) and reset the success streak."""
+        with self._lock:
+            self._limit = max(1, self._limit // 2)
+            self._success_streak = 0
+
+    def report_success(self) -> None:
+        """Increment the success streak; bump limit by 1 if streak hits current limit."""
+        with self._lock:
+            self._success_streak += 1
+            if self._success_streak >= self._limit:
+                self._limit = min(self._ceiling, self._limit + 1)
+                self._success_streak = 0
+
+    def update_ceiling(self, ceiling: int) -> None:
+        """Re-anchor the ceiling (e.g., a later bulk call uses a different max_concurrency).
+
+        Clamps the current limit to the new ceiling. Never raises the limit
+        beyond what AIMD has earned — only the ceiling moves.
+        """
+        if ceiling < 1:
+            raise ValueError(f"ceiling must be >= 1, got {ceiling}")
+        with self._lock:
+            self._ceiling = ceiling
+            if self._limit > ceiling:
+                self._limit = ceiling
+
+
+class _AdaptiveLimiterRegistry:
+    """Per-client ``dict[host, _AdaptiveLimiter]`` with on-demand creation.
+
+    One instance lives on each ``Pinecone`` / ``AsyncPinecone`` client. The
+    transport's ``on_throttle`` callback calls ``report_throttled(host)``;
+    the bulk path calls ``get(host, ceiling).current_limit()`` before
+    each batch dispatch.
+    """
+
+    __slots__ = ("_limiters", "_lock")
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._limiters: dict[str, _AdaptiveLimiter] = {}
+
+    def get(self, host: str, ceiling: int) -> _AdaptiveLimiter:
+        """Return the limiter for ``host``, creating one with ``ceiling`` if absent.
+
+        If a limiter already exists with a different ceiling, the existing
+        limiter's ceiling is updated to the new value (current limit
+        stays unchanged unless it exceeds the new ceiling).
+        """
+        with self._lock:
+            limiter = self._limiters.get(host)
+            if limiter is None:
+                limiter = _AdaptiveLimiter(ceiling)
+                self._limiters[host] = limiter
+            elif limiter.ceiling != ceiling:
+                limiter.update_ceiling(ceiling)
+            return limiter
+
+    def report_throttled(self, host: str) -> None:
+        """Convenience: look up the limiter for ``host`` and decrement.
+
+        If no limiter exists for the host yet (e.g., throttle arrived before
+        any bulk call set a ceiling), this is a no-op. The bulk path will
+        create one with the right ceiling on its first call.
+        """
+        with self._lock:
+            limiter = self._limiters.get(host)
+        if limiter is not None:
+            limiter.report_throttled()

tests/unit/_internal/test_adaptive.py

@@ -0,0 +1,145 @@
+from __future__ import annotations
+
+import pytest
+
+from pinecone._internal.adaptive import _AdaptiveLimiter, _AdaptiveLimiterRegistry
+
+
+class TestAdaptiveLimiter:
+    def test_initial_limit_is_ceiling(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=10)
+        assert lim.current_limit() == 10
+        assert lim.ceiling == 10
+
+    def test_throttle_halves_limit(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=10)
+        lim.report_throttled()
+        assert lim.current_limit() == 5
+
+    def test_throttle_floors_at_one(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=2)
+        lim.report_throttled()  # 2 → 1
+        lim.report_throttled()  # 1 → 1 (floor)
+        lim.report_throttled()
+        assert lim.current_limit() == 1
+
+    def test_success_streak_increases_limit(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=10)
+        lim.report_throttled()  # 10 → 5
+        for _ in range(5):
+            lim.report_success()
+        assert lim.current_limit() == 6
+
+    def test_success_streak_resets_on_throttle(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=10)
+        lim.report_throttled()  # 10 → 5
+        for _ in range(3):
+            lim.report_success()
+        lim.report_throttled()  # 5 → 2; streak reset
+        for _ in range(2):
+            lim.report_success()
+        # Streak hit 2 → limit becomes 3 (not 4 — streak was reset)
+        assert lim.current_limit() == 3
+
+    def test_increase_caps_at_ceiling(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=2)
+        # No throttle; just lots of successes
+        for _ in range(100):
+            lim.report_success()
+        assert lim.current_limit() == 2
+
+    def test_update_ceiling_clamps_limit(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=10)
+        assert lim.current_limit() == 10
+        lim.update_ceiling(3)
+        assert lim.ceiling == 3
+        assert lim.current_limit() == 3
+
+    def test_update_ceiling_does_not_raise_limit(self) -> None:
+        lim = _AdaptiveLimiter(ceiling=10)
+        lim.report_throttled()  # 10 → 5
+        lim.update_ceiling(20)  # ceiling moves up
+        assert lim.ceiling == 20
+        assert lim.current_limit() == 5  # AIMD-earned limit unchanged
+
+    def test_invalid_ceiling_raises(self) -> None:
+        with pytest.raises(ValueError):
+            _AdaptiveLimiter(ceiling=0)
+        with pytest.raises(ValueError):
+            _AdaptiveLimiter(ceiling=-1)
+
+
+class TestAdaptiveLimiterRegistry:
+    def test_get_creates_limiter_on_first_call(self) -> None:
+        reg = _AdaptiveLimiterRegistry()
+        lim = reg.get("host-a.pinecone.io", ceiling=8)
+        assert lim.ceiling == 8
+        assert lim.current_limit() == 8
+
+    def test_get_returns_same_instance_for_same_host(self) -> None:
+        reg = _AdaptiveLimiterRegistry()
+        a1 = reg.get("host-a.pinecone.io", ceiling=8)
+        a2 = reg.get("host-a.pinecone.io", ceiling=8)
+        assert a1 is a2
+
+    def test_get_isolates_hosts(self) -> None:
+        reg = _AdaptiveLimiterRegistry()
+        a = reg.get("host-a.pinecone.io", ceiling=8)
+        b = reg.get("host-b.pinecone.io", ceiling=8)
+        assert a is not b
+        a.report_throttled()
+        assert a.current_limit() == 4
+        assert b.current_limit() == 8
+
+    def test_report_throttled_on_unknown_host_is_noop(self) -> None:
+        reg = _AdaptiveLimiterRegistry()
+        # Should not raise
+        reg.report_throttled("unknown-host.pinecone.io")
+
+    def test_get_with_different_ceiling_updates_existing(self) -> None:
+        reg = _AdaptiveLimiterRegistry()
+        a = reg.get("host-a.pinecone.io", ceiling=8)
+        a.report_throttled()  # 8 → 4
+        a2 = reg.get("host-a.pinecone.io", ceiling=16)
+        assert a is a2
+        assert a.ceiling == 16
+        assert a.current_limit() == 4  # AIMD-earned limit unchanged
+
+    def test_concurrent_access_invariants(self) -> None:
+        """Spawn N threads alternately throttling and reporting success;
+        assert no exceptions and 1 <= current_limit() <= ceiling throughout.
+
+        Smoke-level concurrent-access check — not Hypothesis-grade — but it
+        catches obvious failures (race on ``_limit``, missed ``notify``,
+        silent ``AssertionError`` swallowing, deadlocks).
+        """
+        import threading
+        import time
+
+        reg = _AdaptiveLimiterRegistry()
+        lim = reg.get("test-host", ceiling=16)
+        errors: list[BaseException] = []
+        stop = threading.Event()
+
+        def thrash(action: str) -> None:
+            try:
+                while not stop.is_set():
+                    if action == "throttle":
+                        lim.report_throttled()
+                    else:
+                        lim.report_success()
+                    cur = lim.current_limit()
+                    assert 1 <= cur <= 16, f"limit out of bounds: {cur}"
+            except BaseException as e:
+                errors.append(e)
+
+        threads = [threading.Thread(target=thrash, args=("throttle",)) for _ in range(4)]
+        threads += [threading.Thread(target=thrash, args=("success",)) for _ in range(4)]
+        for t in threads:
+            t.start()
+        time.sleep(0.5)
+        stop.set()
+        for t in threads:
+            t.join(timeout=2.0)
+            assert not t.is_alive(), "thread did not exit"
+        assert not errors, f"errors during concurrent access: {errors}"