Azure · tvaron3 · Apr 29, 2026 · Apr 14, 2026 · Apr 14, 2026 · Apr 14, 2026
diff --git a/.gitignore b/.gitignore
@@ -179,4 +179,5 @@ component-detection-pip-report.json
 uv.lock
 
 # Sphinx generated documentation
-website/
+website/
+.coding-harness/
@@ -10,6 +10,7 @@
 * Fixed bug where `CosmosClient` construction with AAD credentials would crash at startup if the semantic reranking inference endpoint environment variable was not set, even when semantic reranking was not being used. The inference service is now lazily initialized on first use. See [PR 46243](https://github.com/Azure/azure-sdk-for-python/pull/46243)
 
 #### Other Changes
+* Reduced per-client memory overhead when partition-level circuit breaker (PPCB) is enabled by sharing the partition key range routing map cache across CosmosClient instances connected to the same endpoint, and stripping unused fields from cached partition key ranges using compact PKRange namedtuples. See [PR 46297](https://github.com/Azure/azure-sdk-for-python/pull/46297)
 
 ### 4.16.0b2 (2026-04-04)
 

@@ -3635,11 +3635,11 @@ def refresh_routing_map_provider(
                 )
         else:
             # Full refresh - create a new provider instance. This clears all cached routing maps.
-            self._routing_map_provider = routing_map_provider.SmartRoutingMapProvider(self)
+            self._routing_map_provider.clear_cache()
             return
 
         # Fallback to full refresh when targeted refresh fails transiently.
-        self._routing_map_provider = routing_map_provider.SmartRoutingMapProvider(self)
+        self._routing_map_provider.clear_cache()
 
     def _refresh_container_properties_cache(self, container_link: str):
         # If container properties cache is stale, refresh it by reading the container.

@@ -50,6 +50,17 @@ class _PartitionHealthInfo(object):
     """
     This internal class keeps the health and statistics for a partition.
     """
+    # __slots__ reduces per-instance memory by using a fixed-size C array
+    # instead of a per-instance __dict__. Significant when tracking many partitions.
+    __slots__ = (
+        'write_failure_count',
+        'read_failure_count',
+        'write_success_count',
+        'read_success_count',
+        'read_consecutive_failure_count',
+        'write_consecutive_failure_count',
+        'unavailability_info',
+    )
 
     def __init__(self) -> None:
         self.write_failure_count: int = 0

@@ -34,6 +34,7 @@
 from .collection_routing_map import CollectionRoutingMap, _build_routing_map_from_ranges
 from . import routing_range
 from .routing_range import (
+    PKRange,
     PartitionKeyRange,
     _is_sorted_and_non_overlapping,
     _subtract_range,
@@ -186,7 +187,7 @@ def process_fetched_ranges(
     # Incremental update -- merge deltas into the existing map.
     # Resolve parent chains transitively within this single delta so cascading
     # splits (A->B+C and B->D+E in one payload) can be merged incrementally.
-    range_tuples: List[Tuple[Dict[str, Any], Any]] = []
+    range_tuples: List[Tuple[Any, Any]] = []
     known_range_info_by_id = {
         pkr_id: pkr_tuple[1]
         for pkr_id, pkr_tuple in previous_routing_map._rangeById.items()  # pylint: disable=protected-access
@@ -209,7 +210,11 @@ def process_fetched_ranges(
                 next_unresolved.append(r)
                 continue
 
-            range_tuples.append((r, range_info))
+            range_tuples.append((PKRange(
+                id=r[PartitionKeyRange.Id],
+                minInclusive=r[PartitionKeyRange.MinInclusive],
+                maxExclusive=r[PartitionKeyRange.MaxExclusive],
+                parents=tuple(r.get(PartitionKeyRange.Parents) or ())), range_info))
             known_range_info_by_id[r[PartitionKeyRange.Id]] = range_info
             progress_made = True
 

@@ -24,6 +24,7 @@
 """
 import asyncio  # pylint: disable=do-not-import-asyncio
 import logging
+import threading
 from typing import Dict, Any, Optional, List, TYPE_CHECKING
 from azure.core.utils import CaseInsensitiveDict
 from ... import _base, http_constants
@@ -41,6 +42,39 @@
 
 if TYPE_CHECKING:
     from ...aio._cosmos_client_connection_async import CosmosClientConnection
+
+# Shared routing map cache across all clients targeting the same endpoint.
+# All four module-level dicts are keyed by endpoint and protected by
+# ``_shared_cache_lock`` for mutation. Per-collection refresh serialization is
+# handled by the per-endpoint asyncio.Locks in ``_shared_collection_locks`` so
+# that all clients sharing an endpoint single-flight refreshes through the
+# same lock.
+_shared_routing_map_cache: dict = {}
+_shared_collection_locks: Dict[str, Dict[str, asyncio.Lock]] = {}
+_shared_locks_locks: Dict[str, asyncio.Lock] = {}
+_shared_cache_refcounts: Dict[str, int] = {}
+_shared_cache_lock = threading.Lock()
+
+
+def _resolve_endpoint(client: Any) -> str:
+    """Return a cache key for ``client``'s endpoint.
+
+    Falls back to ``__unknown_<id>__`` when ``client`` has no ``url_connection``
+    so unknown/mocked clients are isolated rather than collapsed into a single
+    shared cache entry.
+
+    :param client: The CosmosClient (or compatible) instance whose endpoint
+        will be used as the shared-cache key.
+    :type client: Any
+    :returns: The endpoint URL string, or a per-instance fallback key when the
+        client does not expose ``url_connection``.
+    :rtype: str
+    """
+    try:
+        return client.url_connection
+    except AttributeError:
+        return f"__unknown_{id(client)}__"
+
 # pylint: disable=protected-access
 
 logger = logging.getLogger(__name__)
@@ -64,13 +98,68 @@ def __init__(self, client: Any):
         """
 
         self._document_client = client
+        self._endpoint = _resolve_endpoint(client)
+        self._released = False
+
+        # Share routing map cache, per-collection asyncio locks, and the
+        # per-endpoint meta-lock that guards the per-collection-lock dict
+        # across all clients with the same endpoint. Refcount lets us evict
+        # the entry when the last sharing client releases it (see ``release``).
+        with _shared_cache_lock:
+            if self._endpoint not in _shared_routing_map_cache:
+                _shared_routing_map_cache[self._endpoint] = {}
+                _shared_collection_locks[self._endpoint] = {}
+                _shared_locks_locks[self._endpoint] = asyncio.Lock()
+                _shared_cache_refcounts[self._endpoint] = 0
+            _shared_cache_refcounts[self._endpoint] += 1
+            self._collection_routing_map_by_item = _shared_routing_map_cache[self._endpoint]
+            self._collection_locks: Dict[str, asyncio.Lock] = _shared_collection_locks[self._endpoint]
+            self._locks_lock: asyncio.Lock = _shared_locks_locks[self._endpoint]
+
+    async def clear_cache(self):
+        """Clear the shared routing map cache for this endpoint.
+
+        Uses in-place ``.clear()`` to preserve all client references to the
+        same dict and the same per-collection lock dict, so concurrent clients
+        sharing the endpoint continue to single-flight through the same locks.
+        """
+        async with self._locks_lock:
+            with _shared_cache_lock:
+                if self._endpoint in _shared_routing_map_cache:
+                    _shared_routing_map_cache[self._endpoint].clear()
+            self._collection_locks.clear()
 
-        # keeps the cached collection routing map by collection id
-        self._collection_routing_map_by_item: Dict[str, CollectionRoutingMap] = {}
-        # A lock to control access to the locks dictionary itself
-        self._locks_lock = asyncio.Lock()
-        # A dictionary to hold a lock for each collection ID
-        self._collection_locks: Dict[str, asyncio.Lock] = {}
+    def release(self) -> None:
+        """Decrement the per-endpoint refcount and evict shared state at zero.
+
+        Safe to call multiple times. Best-effort: never raises.
+        """
+        if self._released:
+            return
+        self._released = True
+        endpoint = self._endpoint
+        try:
+            with _shared_cache_lock:
+                count = _shared_cache_refcounts.get(endpoint, 0) - 1
+                if count <= 0:
+                    _shared_cache_refcounts.pop(endpoint, None)
+                    _shared_routing_map_cache.pop(endpoint, None)
+                    _shared_collection_locks.pop(endpoint, None)
+                    _shared_locks_locks.pop(endpoint, None)
+                else:
+                    _shared_cache_refcounts[endpoint] = count
+        except Exception:  # pylint: disable=broad-except
+            # release() may be called from __del__ during interpreter shutdown
+            # where module globals may already be torn down.
+            pass
+
+    def __del__(self):
+        # Defensive fallback in case the owning client teardown path didn't
+        # call release(). Must never raise.
+        try:
+            self.release()
+        except Exception:  # pylint: disable=broad-except
+            pass
 
     async def _get_lock_for_collection(self, collection_id: str) -> asyncio.Lock:
         """Safely gets or creates a lock for a given collection ID.

@@ -27,7 +27,7 @@
 from typing import Optional, Union
 
 from azure.cosmos._routing import routing_range
-from azure.cosmos._routing.routing_range import PartitionKeyRange
+from azure.cosmos._routing.routing_range import PartitionKeyRange, PKRange
 
 # pylint: disable=line-too-long
 class CollectionRoutingMap(object):
@@ -288,7 +288,13 @@ def _build_routing_map_from_ranges(
         if PartitionKeyRange.Parents in r and r[PartitionKeyRange.Parents]:
             gone_range_ids.update(r[PartitionKeyRange.Parents])
 
-    filtered_ranges = [r for r in ranges if r[PartitionKeyRange.Id] not in gone_range_ids]
+    filtered_ranges = [
+        PKRange(id=r[PartitionKeyRange.Id],
+                minInclusive=r[PartitionKeyRange.MinInclusive],
+                maxExclusive=r[PartitionKeyRange.MaxExclusive],
+                parents=tuple(r.get(PartitionKeyRange.Parents) or ()))
+        for r in ranges if r[PartitionKeyRange.Id] not in gone_range_ids
+    ]
     range_tuples = [(r, True) for r in filtered_ranges]
 
     routing_map = CollectionRoutingMap.CompleteRoutingMap(

@@ -40,6 +40,38 @@
 
 if TYPE_CHECKING:
     from .._cosmos_client_connection import CosmosClientConnection
+
+# Shared routing map cache across all clients targeting the same endpoint.
+# All four module-level dicts are keyed by endpoint and protected by
+# ``_shared_cache_lock`` for mutation. Per-collection refresh serialization is
+# handled by the per-endpoint locks in ``_shared_collection_locks`` so that all
+# clients sharing an endpoint single-flight refreshes through the same lock.
+_shared_routing_map_cache: dict = {}
+_shared_collection_locks: Dict[str, Dict[str, threading.Lock]] = {}
+_shared_locks_locks: Dict[str, threading.Lock] = {}
+_shared_cache_refcounts: Dict[str, int] = {}
+_shared_cache_lock = threading.Lock()
+
+
+def _resolve_endpoint(client: Any) -> str:
+    """Return a cache key for ``client``'s endpoint.
+
+    Falls back to ``__unknown_<id>__`` when ``client`` has no ``url_connection``
+    so unknown/mocked clients are isolated rather than collapsed into a single
+    shared cache entry.
+
+    :param client: The CosmosClient (or compatible) instance whose endpoint
+        will be used as the shared-cache key.
+    :type client: Any
+    :returns: The endpoint URL string, or a per-instance fallback key when the
+        client does not expose ``url_connection``.
+    :rtype: str
+    """
+    try:
+        return client.url_connection
+    except AttributeError:
+        return f"__unknown_{id(client)}__"
+
 # pylint: disable=protected-access, line-too-long
 
 
@@ -63,13 +95,68 @@ def __init__(self, client: Any):
         """
 
         self._document_client = client
+        self._endpoint = _resolve_endpoint(client)
+        self._released = False
+
+        # Share routing map cache, per-collection locks, and the meta-lock that
+        # guards the per-collection-lock dict across all clients with the same
+        # endpoint. Refcount lets us evict the entry when the last sharing
+        # client releases it (see ``release``).
+        with _shared_cache_lock:
+            if self._endpoint not in _shared_routing_map_cache:
+                _shared_routing_map_cache[self._endpoint] = {}
+                _shared_collection_locks[self._endpoint] = {}
+                _shared_locks_locks[self._endpoint] = threading.Lock()
+                _shared_cache_refcounts[self._endpoint] = 0
+            _shared_cache_refcounts[self._endpoint] += 1
+            self._collection_routing_map_by_item = _shared_routing_map_cache[self._endpoint]
+            self._collection_locks: Dict[str, threading.Lock] = _shared_collection_locks[self._endpoint]
+            self._locks_lock: threading.Lock = _shared_locks_locks[self._endpoint]
+
+    def clear_cache(self):
+        """Clear the shared routing map cache for this endpoint.
+
+        Uses in-place ``.clear()`` to preserve all client references to the
+        same dict and the same per-collection lock dict, so concurrent clients
+        sharing the endpoint continue to single-flight through the same locks.
+        """
+        with self._locks_lock:
+            with _shared_cache_lock:
+                if self._endpoint in _shared_routing_map_cache:
+                    _shared_routing_map_cache[self._endpoint].clear()
+            self._collection_locks.clear()
 
-        # keeps the cached collection routing map by collection id
-        self._collection_routing_map_by_item: Dict[str, CollectionRoutingMap] = {}
-        # A lock to control access to the locks dictionary itself
-        self._locks_lock = threading.Lock()
-        # A dictionary to hold a lock for each collection ID
-        self._collection_locks: Dict[str, threading.Lock] = {}
+    def release(self) -> None:
+        """Decrement the per-endpoint refcount and evict shared state at zero.
+
+        Safe to call multiple times. Best-effort: never raises.
+        """
+        if self._released:
+            return
+        self._released = True
+        endpoint = self._endpoint
+        try:
+            with _shared_cache_lock:
+                count = _shared_cache_refcounts.get(endpoint, 0) - 1
+                if count <= 0:
+                    _shared_cache_refcounts.pop(endpoint, None)
+                    _shared_routing_map_cache.pop(endpoint, None)
+                    _shared_collection_locks.pop(endpoint, None)
+                    _shared_locks_locks.pop(endpoint, None)
+                else:
+                    _shared_cache_refcounts[endpoint] = count
+        except Exception:  # pylint: disable=broad-except
+            # release() may be called from __del__ during interpreter shutdown
+            # where module globals may already be torn down.
+            pass
+
+    def __del__(self):
+        # Defensive fallback in case the owning client teardown path didn't
+        # call release(). Must never raise.
+        try:
+            self.release()
+        except Exception:  # pylint: disable=broad-except
+            pass
 
     def _get_lock_for_collection(self, collection_id: str) -> threading.Lock: