feat(grpc): wire throttle callback from _AdaptiveLimiterRegistry to GrpcChannel

- Add `on_throttle: Callable[[str], None] | None = None` parameter to `GrpcIndex.__init__`
  and forward it to the Rust-backed `GrpcChannel` constructor.
- Update `Pinecone.index(grpc=True)` factory to pass `self._limiter_registry.report_throttled`
  as the `on_throttle` callback, completing the loop started by DX-0159 (REST->registry).
- Add `tests/unit/grpc/test_grpc_limiter_wiring.py` with 4 tests covering:
  - Default on_throttle=None for direct GrpcIndex construction
  - Callable forwarding when on_throttle is provided
  - Pinecone client wires registry callback (verifies __self__ identity)
  - Direct construction without client does not raise
- Update TestIndexFactoryGrpc tests to assert per-kwarg (bound-method identity safe).

DX-0160: gRPC throttle events (RESOURCE_EXHAUSTED) now flow into the same per-host
_AdaptiveLimiter as REST 429s. Bulk paths read limiter.current_limit() to adjust concurrency.

Author: jhamon

pinecone/_client.py

@@ -394,6 +394,7 @@ def index(
                 host=resolved_host,
                 api_key=self._config.api_key,
                 source_tag=self._config.source_tag or None,
+                on_throttle=self._limiter_registry.report_throttled,
             )
 
         from pinecone.index import Index as _Index

pinecone/grpc/__init__.py

@@ -5,7 +5,7 @@
 import builtins
 import logging
 import os
-from collections.abc import Iterator, Mapping, Sequence
+from collections.abc import Callable, Iterator, Mapping, Sequence
 from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, Any
 
@@ -185,6 +185,7 @@ def __init__(
         secure: bool = True,
         timeout: float = 20.0,
         connect_timeout: float = 1.0,
+        on_throttle: Callable[[str], None] | None = None,
     ) -> None:
         # Resolve API key: explicit arg > env var
         resolved_key = api_key or os.environ.get("PINECONE_API_KEY", "")
@@ -213,6 +214,7 @@ def __init__(
             timeout,
             connect_timeout,
             source_tag=source_tag,
+            on_throttle=on_throttle,
         )
 
         self._executor = ThreadPoolExecutor()

tests/unit/grpc/test_grpc_limiter_wiring.py

@@ -0,0 +1,85 @@
+"""Tests that GrpcIndex wires the throttle callback to the _AdaptiveLimiterRegistry."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from unittest.mock import MagicMock, patch
+
+_MOCK_GRPC_MODULE_PATH = "pinecone._grpc"
+
+
+def _make_grpc_index_with_mock(
+    on_throttle: Callable[[str], None] | None = None,
+) -> tuple[MagicMock, MagicMock]:
+    """Return (mock_channel_class, grpc_index_instance) for inspection."""
+    from pinecone.grpc import GrpcIndex
+
+    mock_channel = MagicMock()
+    mock_module = MagicMock()
+    mock_module.GrpcChannel.return_value = mock_channel
+    with patch.dict("sys.modules", {_MOCK_GRPC_MODULE_PATH: mock_module}):
+        idx = GrpcIndex(
+            host="https://test-idx-abc.svc.pinecone.io",
+            api_key="test-key",
+            on_throttle=on_throttle,
+        )
+    return mock_module.GrpcChannel, idx
+
+
+def test_grpc_index_construction_passes_on_throttle_none_by_default() -> None:
+    mock_channel_cls, _ = _make_grpc_index_with_mock(on_throttle=None)
+    _, kwargs = mock_channel_cls.call_args
+    assert kwargs.get("on_throttle") is None
+
+
+def test_grpc_index_construction_passes_on_throttle_callable() -> None:
+    sentinel: list[str] = []
+
+    def callback(host: str) -> None:
+        sentinel.append(host)
+
+    mock_channel_cls, _ = _make_grpc_index_with_mock(on_throttle=callback)
+    _, kwargs = mock_channel_cls.call_args
+    assert kwargs.get("on_throttle") is callback
+
+
+def test_pinecone_client_wires_registry_to_grpc_index() -> None:
+    """GrpcIndex constructed via Pinecone.index(grpc=True) receives the registry callback."""
+    from pinecone import Pinecone
+
+    pc = Pinecone(api_key="test-key")
+
+    mock_channel = MagicMock()
+    mock_module = MagicMock()
+    mock_module.GrpcChannel.return_value = mock_channel
+
+    with (
+        patch.dict("sys.modules", {_MOCK_GRPC_MODULE_PATH: mock_module}),
+        patch.object(pc, "_resolve_index_host", return_value="test-idx-abc.svc.pinecone.io"),
+    ):
+        pc.index(host="test-idx-abc.svc.pinecone.io", grpc=True)
+
+    _, kwargs = mock_module.GrpcChannel.call_args
+    on_throttle = kwargs.get("on_throttle")
+    assert on_throttle is not None, "on_throttle should be wired from the limiter registry"
+    assert callable(on_throttle)
+    # Bound methods create a new object on each attribute access, so compare via __func__/__self__
+    assert on_throttle.__func__ is pc._limiter_registry.report_throttled.__func__
+    assert on_throttle.__self__ is pc._limiter_registry
+
+
+def test_grpc_index_direct_construction_on_throttle_none_does_not_raise() -> None:
+    """GrpcIndex constructed directly (no parent client) works fine with on_throttle=None."""
+    from pinecone.grpc import GrpcIndex
+
+    mock_channel = MagicMock()
+    mock_module = MagicMock()
+    mock_module.GrpcChannel.return_value = mock_channel
+    with patch.dict("sys.modules", {_MOCK_GRPC_MODULE_PATH: mock_module}):
+        GrpcIndex(
+            host="https://direct-construction.svc.pinecone.io",
+            api_key="test-key",
+        )
+
+    _, kwargs = mock_module.GrpcChannel.call_args
+    assert kwargs.get("on_throttle") is None

tests/unit/test_client.py

@@ -413,11 +413,15 @@ def test_grpc_true_returns_grpc_index(self, mock_grpc_cls: MagicMock) -> None:
         result = pc.index(host="foo.svc.pinecone.io", grpc=True)
 
         assert result is mock_grpc_idx
-        mock_grpc_cls.assert_called_once_with(
-            host="foo.svc.pinecone.io",
-            api_key="test-key",
-            source_tag=None,
-        )
+        mock_grpc_cls.assert_called_once()
+        _, call_kwargs = mock_grpc_cls.call_args
+        assert call_kwargs["host"] == "foo.svc.pinecone.io"
+        assert call_kwargs["api_key"] == "test-key"
+        assert call_kwargs["source_tag"] is None
+        # on_throttle must be wired to the registry's report_throttled method
+        on_throttle = call_kwargs["on_throttle"]
+        assert callable(on_throttle)
+        assert on_throttle.__self__ is pc._limiter_registry
 
     @patch("pinecone.grpc.GrpcIndex")
     def test_grpc_true_passes_source_tag(self, mock_grpc_cls: MagicMock) -> None:
@@ -426,11 +430,12 @@ def test_grpc_true_passes_source_tag(self, mock_grpc_cls: MagicMock) -> None:
 
         pc.index(host="foo.svc.pinecone.io", grpc=True)
 
-        mock_grpc_cls.assert_called_once_with(
-            host="foo.svc.pinecone.io",
-            api_key="test-key",
-            source_tag="my_app",
-        )
+        mock_grpc_cls.assert_called_once()
+        _, call_kwargs = mock_grpc_cls.call_args
+        assert call_kwargs["host"] == "foo.svc.pinecone.io"
+        assert call_kwargs["api_key"] == "test-key"
+        assert call_kwargs["source_tag"] == "my_app"
+        assert callable(call_kwargs["on_throttle"])
 
     @patch("pinecone.grpc.GrpcIndex")
     def test_grpc_true_with_name_cached(self, mock_grpc_cls: MagicMock) -> None:
@@ -440,11 +445,12 @@ def test_grpc_true_with_name_cached(self, mock_grpc_cls: MagicMock) -> None:
 
         pc.index(name="my-index", grpc=True)
 
-        mock_grpc_cls.assert_called_once_with(
-            host="cached.host.pinecone.io",
-            api_key="test-key",
-            source_tag=None,
-        )
+        mock_grpc_cls.assert_called_once()
+        _, call_kwargs = mock_grpc_cls.call_args
+        assert call_kwargs["host"] == "cached.host.pinecone.io"
+        assert call_kwargs["api_key"] == "test-key"
+        assert call_kwargs["source_tag"] is None
+        assert callable(call_kwargs["on_throttle"])
 
     @patch("pinecone.grpc.GrpcIndex")
     def test_grpc_true_with_name_describe(self, mock_grpc_cls: MagicMock) -> None:
@@ -461,11 +467,12 @@ def test_grpc_true_with_name_describe(self, mock_grpc_cls: MagicMock) -> None:
 
         mock_indexes.describe.assert_called_once_with("my-index")
         assert pc._host_cache["my-index"] == "resolved.host.pinecone.io"
-        mock_grpc_cls.assert_called_once_with(
-            host="resolved.host.pinecone.io",
-            api_key="test-key",
-            source_tag=None,
-        )
+        mock_grpc_cls.assert_called_once()
+        _, call_kwargs = mock_grpc_cls.call_args
+        assert call_kwargs["host"] == "resolved.host.pinecone.io"
+        assert call_kwargs["api_key"] == "test-key"
+        assert call_kwargs["source_tag"] is None
+        assert callable(call_kwargs["on_throttle"])
 
     @patch("pinecone.index.Index")
     def test_grpc_false_default_returns_http_index(self, mock_index_cls: MagicMock) -> None: