feat(batch): gate sync ThreadPoolExecutor via admission semaphore from _AdaptiveLimiter

## Purpose

Sync bulk paths submitted all batches to the thread pool up front with no
way to dynamically reduce in-flight concurrency under throttling. Without
an admission gate, a pre-sized pool of 8 threads would keep 8 batches
in-flight even when the adaptive limiter had dropped the limit to 2.

## Solution

Added `limiter_registry` and `host` parameters (both optional, SDK-internal)
to `batch_execute`. When wired, a `threading.Condition`-protected inflight
counter gates submission: the main thread acquires before each
`executor.submit()` and workers release (via `_wrapped_op` finally) on
completion. `_acquire()` re-reads `limiter.current_limit()` on every wakeup,
so a recovering limiter unlocks additional slots without any extra wiring.

When `limiter_registry is None` the gate is a no-op: `_acquire()` and
`_release()` return immediately, preserving the original submission behavior.

## Follow-ups

Transport callback wiring (DX-0159 for REST, DX-0160 for gRPC) is needed
before `report_throttled()` is called in production; until then
`current_limit()` stays at ceiling and the gate is functionally a no-op.

Author: jhamon

pinecone/_internal/batch.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import asyncio
+import threading
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from typing import TYPE_CHECKING, Any, TypeVar
 
@@ -142,6 +143,8 @@ def batch_execute(
     show_progress: bool = True,
     desc: str = "Batches",
     executor: ThreadPoolExecutor | None = None,
+    limiter_registry: _AdaptiveLimiterRegistry | None = None,
+    host: str | None = None,
 ) -> BatchResult:
     """Execute *operation* on *items* in parallel batches.
 
@@ -163,6 +166,10 @@ def batch_execute(
             or torn down per call. Caller is responsible for ``shutdown()``.
             When ``None`` (default), a private executor is created and
             shut down at the end of this call.
+        limiter_registry (_AdaptiveLimiterRegistry | None): Optional registry
+            for adaptive concurrency. SDK-internal; not for user code.
+        host (str | None): Host key for the limiter registry lookup.
+            SDK-internal; not for user code.
 
     Returns:
         BatchResult with aggregated success/failure counts.
@@ -182,16 +189,49 @@ def batch_execute(
     lsn_reconciled_values: list[int] = []
     lsn_committed_values: list[int] = []
 
+    if limiter_registry is not None and host is not None:
+        limiter = limiter_registry.get(host, max_concurrency)
+    else:
+        limiter = None
+
+    condition = threading.Condition()
+    inflight = 0
+
+    def _acquire() -> None:
+        nonlocal inflight
+        if limiter is None:
+            return
+        with condition:
+            while inflight >= limiter.current_limit():
+                condition.wait()
+            inflight += 1
+
+    def _release() -> None:
+        nonlocal inflight
+        if limiter is None:
+            return
+        with condition:
+            inflight -= 1
+            condition.notify_all()
+
+    def _wrapped_op(batch: list[dict[str, Any]]) -> Any:
+        try:
+            return operation(batch)
+        finally:
+            _release()
+
     progress = _create_progress_bar(total_batches, desc, show_progress)
 
     own_executor = executor is None
     if executor is None:
         executor = ThreadPoolExecutor(max_workers=max_concurrency)
 
     try:
-        future_to_batch = {
-            executor.submit(operation, batch): (idx, batch) for idx, batch in enumerate(batches)
-        }
+        future_to_batch: dict[Any, tuple[int, list[dict[str, Any]]]] = {}
+        for idx, batch in enumerate(batches):
+            _acquire()
+            future = executor.submit(_wrapped_op, batch)
+            future_to_batch[future] = (idx, batch)
 
         for future in as_completed(future_to_batch):
             batch_idx, batch = future_to_batch[future]

tests/unit/_internal/test_batch.py

@@ -3,12 +3,14 @@
 from __future__ import annotations
 
 import asyncio
+import threading
+import time
 from typing import Any
 
 import pytest
 
 from pinecone._internal.adaptive import _AdaptiveLimiterRegistry
-from pinecone._internal.batch import async_batch_execute
+from pinecone._internal.batch import async_batch_execute, batch_execute
 
 
 class TestAdaptiveBatchExecute:
@@ -200,3 +202,99 @@ async def slow(batch: list[dict[str, Any]]) -> Any:
             host="host2",
         )
         assert peak[0] <= 3
+
+
+class TestAdaptiveBatchExecuteSync:
+    def test_sync_no_limiter_unchanged(self) -> None:
+        """Without a limiter_registry, behavior is identical to baseline."""
+        items = [{"id": str(i), "values": [0.1, 0.2]} for i in range(6)]
+
+        def op(batch: list[dict[str, Any]]) -> Any:
+            return {"upserted_count": len(batch)}
+
+        result = batch_execute(
+            items=items,
+            operation=op,
+            batch_size=2,
+            max_concurrency=4,
+            show_progress=False,
+        )
+        assert result.successful_item_count == 6
+        assert result.failed_item_count == 0
+        assert result.total_item_count == 6
+
+    def test_sync_uses_limiter(self) -> None:
+        """With a pre-throttled limiter (4), max observed inflight must not exceed 4."""
+        registry = _AdaptiveLimiterRegistry()
+        limiter = registry.get("test-host", 8)
+        limiter.report_throttled()  # 8 → 4
+        assert limiter.current_limit() == 4
+
+        inflight_counter = 0
+        max_observed = [0]
+        lock = threading.Lock()
+
+        def slow_op(batch: list[dict[str, Any]]) -> Any:
+            nonlocal inflight_counter
+            with lock:
+                inflight_counter += 1
+                max_observed[0] = max(max_observed[0], inflight_counter)
+            time.sleep(0.05)
+            with lock:
+                inflight_counter -= 1
+            return {"upserted_count": len(batch)}
+
+        items = [{"id": str(i), "values": [0.1, 0.2]} for i in range(20)]
+        result = batch_execute(
+            items=items,
+            operation=slow_op,
+            batch_size=2,  # 10 batches
+            max_concurrency=8,
+            show_progress=False,
+            limiter_registry=registry,
+            host="test-host",
+        )
+        assert max_observed[0] <= 4, (
+            f"observed max concurrency {max_observed[0]} exceeds limiter cap"
+        )
+        assert result.successful_item_count == 20
+        assert result.failed_item_count == 0
+
+    def test_sync_concurrency_recovers_after_signals(self) -> None:
+        """Concurrency recovers as limiter is signalled with successes mid-execution."""
+        registry = _AdaptiveLimiterRegistry()
+        limiter = registry.get("test-host", 8)
+        # Pre-throttle to 1
+        for _ in range(3):
+            limiter.report_throttled()
+        assert limiter.current_limit() == 1
+
+        call_count = 0
+        call_lock = threading.Lock()
+        limits_observed: list[int] = []
+
+        def op(batch: list[dict[str, Any]]) -> Any:
+            nonlocal call_count
+            with call_lock:
+                call_count += 1
+                current = call_count
+            # After 3 calls, recover by reporting successes
+            if current == 3:
+                for _ in range(20):
+                    limiter.report_success()
+            limits_observed.append(limiter.current_limit())
+            return {"upserted_count": len(batch)}
+
+        items = [{"id": str(i)} for i in range(20)]  # 10 batches of 2
+        result = batch_execute(
+            items=items,
+            operation=op,
+            batch_size=2,
+            max_concurrency=8,
+            show_progress=False,
+            limiter_registry=registry,
+            host="test-host",
+        )
+        assert result.successful_item_count == 20
+        assert result.failed_item_count == 0
+        assert max(limits_observed) > 1, "limiter should have recovered above 1"