fixing ValueError bug

Author: dibahlfi

sdk/cosmos/azure-cosmos/azure/cosmos/_routing/_routing_map_provider_common.py

@@ -28,10 +28,16 @@
 """
 
 import logging
+import random
 from typing import Any, Dict, List, Optional, Tuple
 
 from .. import _base, http_constants
-from .collection_routing_map import CollectionRoutingMap, _build_routing_map_from_ranges
+from ..exceptions import CosmosHttpResponseError
+from .collection_routing_map import (
+    CollectionRoutingMap,
+    _build_routing_map_from_ranges,
+    _OverlapDetected,  # noqa: F401  # re-exported for sync/async provider modules and tests
+)
 from . import routing_range
 from .routing_range import (
     PKRange,
@@ -44,6 +50,114 @@
 
 PAGE_SIZE_CHANGE_FEED = "-1"  # Return all available changes
 
+# Number of times the full-load path will re-fetch ``/pkranges`` when the
+# builder reports an overlap (``_OverlapDetected``). Overlap on the full-load
+# path is treated as a transient gateway inconsistency, so a small fixed
+# retry budget with backoff is preferred over surfacing immediately. After
+# this many attempts the caller surfaces a transient HTTP 503 so the
+# upstream retry policy can take over.
+#
+# Defined here (rather than in each provider module) so the sync and async
+# providers cannot drift on the retry budget — both import the same constant.
+_OVERLAP_RETRY_MAX_ATTEMPTS = 3
+# Initial backoff between overlap retries; doubles each attempt. Worst-case
+# total sleep under the budget above is ~3.5s (0.5 + 1.0 + 2.0).
+_OVERLAP_RETRY_INITIAL_BACKOFF_SECONDS = 0.5
+
+
+def _jittered_backoff(backoff_seconds: float) -> float:
+    """Return a uniformly-jittered backoff in the range ``[0, backoff_seconds]``.
+
+    Implements the "full jitter" strategy: the actual sleep is drawn uniformly
+    from zero to the full deterministic backoff. This decorrelates concurrent
+    retriers (for example, multiple Cosmos clients running inside a single
+    PySpark process that all hit the same gateway node on the same bad
+    ``/pkranges`` snapshot at the same instant) so they do not retry in
+    lockstep and re-collide on the same gateway node.
+
+    The worst-case sleep per attempt is unchanged (still bounded by the
+    deterministic backoff), so the documented retry-budget contract still
+    holds; the expected per-attempt sleep is half of it.
+    """
+    return random.uniform(0, backoff_seconds)
+
+
+def _handle_overlap_retry_decision(
+    *,
+    overlap_attempt_count: int,
+    collection_link: str,
+    logger: logging.Logger,  # pylint: disable=redefined-outer-name
+) -> float:
+    """Decide what to do after the full-load builder reported an overlap.
+
+    Centralises the sync/async-identical retry policy. Returns the number of
+    seconds the caller should sleep before the next attempt. Raises
+    :class:`CosmosHttpResponseError` (HTTP 503) when the attempt budget has
+    been exhausted; the caller's existing retry policy then handles it as
+    a transient error.
+
+    The returned sleep duration is jittered (see :func:`_jittered_backoff`)
+    so concurrent retriers do not retry in lockstep. The deterministic
+    backoff schedule (0.5s -> 1.0s -> 2.0s, doubling) defines the *upper
+    bound* of each attempt's sleep; the actual sleep is drawn uniformly
+    from ``[0, that upper bound]``.
+
+    The caller is responsible for the actual sleep (sync ``time.sleep`` or
+    ``await asyncio.sleep``). Keeping the sleep at the call site is what
+    lets this helper stay free of concurrency-runtime assumptions — the
+    only line that has to differ between the sync and async providers.
+
+    :param int overlap_attempt_count: Number of overlap attempts made so far,
+        including the one that just failed. Pass ``1`` after the first failure,
+        ``2`` after the second, etc.
+    :param str collection_link: Used in log messages and the 503 error body
+        so the caller knows which collection ran out of budget.
+    :param logging.Logger logger: Caller's module-level logger, so messages
+        appear under the right ``azure.cosmos._routing.*`` namespace.
+    :return: Jittered backoff seconds to sleep before retrying. Guaranteed
+        to be in ``[0, deterministic_backoff_for_attempt]``.
+    :rtype: float
+    :raises CosmosHttpResponseError: When ``overlap_attempt_count`` has reached
+        ``_OVERLAP_RETRY_MAX_ATTEMPTS``. Status code is 503 so the upstream
+        retry policy classifies it as transient.
+    """
+    if overlap_attempt_count >= _OVERLAP_RETRY_MAX_ATTEMPTS:
+        logger.error(
+            "Full-load routing-map fetch for collection '%s' detected "
+            "overlapping partition key ranges on every one of %d attempt(s). "
+            "Surfacing as transient HTTP 503 so the caller's retry policy "
+            "can take over.",
+            collection_link,
+            overlap_attempt_count,
+        )
+        raise CosmosHttpResponseError(
+            status_code=http_constants.StatusCodes.SERVICE_UNAVAILABLE,
+            message=(
+                "Failed to build routing map for collection '{}': "
+                "overlapping partition key ranges persisted across {} "
+                "full-load attempt(s). Surfaced as a retryable transient "
+                "error so the upstream retry policy can take over, rather "
+                "than allowing the underlying ValueError to escape as a "
+                "fatal crash."
+            ).format(collection_link, overlap_attempt_count),
+        )
+
+    deterministic_backoff = (
+        _OVERLAP_RETRY_INITIAL_BACKOFF_SECONDS * (2 ** (overlap_attempt_count - 1))
+    )
+    jittered_backoff = _jittered_backoff(deterministic_backoff)
+    logger.warning(
+        "Full-load routing-map fetch for collection '%s' detected overlapping "
+        "partition key ranges (attempt %d/%d). Sleeping %.2fs (jittered from "
+        "upper bound %.2fs) and retrying.",
+        collection_link,
+        overlap_attempt_count,
+        _OVERLAP_RETRY_MAX_ATTEMPTS,
+        jittered_backoff,
+        deterministic_backoff,
+    )
+    return jittered_backoff
+
 
 def is_cache_unchanged_since_previous(
     collection_routing_map_by_item: Dict[str, CollectionRoutingMap],
@@ -257,7 +371,32 @@ def process_fetched_ranges(
 
         unresolved = next_unresolved
 
-    result = previous_routing_map.try_combine(range_tuples, effective_etag)
+    try:
+        result = previous_routing_map.try_combine(range_tuples, effective_etag)
+    except ValueError as overlap_error:
+        # ``try_combine`` validates the merged map via
+        # ``CollectionRoutingMap.is_complete_set_of_range`` and raises
+        # ``ValueError("Ranges overlap: ...")`` if the merge produces a
+        # self-contradictory tiling. This can happen during the incremental
+        # path when the delta contains a range whose key span overlaps an
+        # existing cached range without either side declaring the other a
+        # parent.
+        #
+        # We must NOT let this ``ValueError`` escape: the cache layer above
+        # treats a ``None`` routing map as "no ranges" and would convert
+        # the bare exception into a silent empty-result return at
+        # ``get_overlapping_ranges``. Convert to ``_IncrementalMergeFailed``
+        # so the caller's existing retry loop retries the incremental fetch
+        # once and then falls back to the full-load path, which has its own
+        # ``_OverlapDetected`` handler with retry+backoff and surfaces a
+        # transient HTTP 503 if the inconsistency persists.
+        logger.warning(
+            "Incremental merge for collection '%s' produced overlapping ranges: %s. "
+            "Converting to _IncrementalMergeFailed so the caller retries / "
+            "falls back to a full refresh.",
+            collection_link, str(overlap_error),
+        )
+        raise _IncrementalMergeFailed() from overlap_error
     if not result:
         logger.warning(
             "Incremental merge resulted in incomplete routing map for "

sdk/cosmos/azure-cosmos/azure/cosmos/_routing/aio/routing_map_provider.py

@@ -38,6 +38,9 @@
     determine_refresh_action,
     get_smart_overlapping_ranges,
     _IncrementalMergeFailed,
+    _OverlapDetected,
+    _OVERLAP_RETRY_MAX_ATTEMPTS,  # noqa: F401  # re-exported for tests
+    _handle_overlap_retry_decision,
 )
 
 
@@ -103,6 +106,10 @@
 # Number of extra incremental attempts after an incomplete incremental merge
 # before falling back to a full routing-map refresh.
 _INCOMPLETE_ROUTING_MAP_MAX_RETRIES = 1
+
+# Overlap-retry budget and backoff live in ``_routing_map_provider_common`` so
+# the sync and async providers cannot drift on them. ``_OVERLAP_RETRY_MAX_ATTEMPTS``
+# is re-exported through this module for test imports.
 class PartitionKeyRangeCache(object):
     """
     PartitionKeyRangeCache provides list of effective partition key ranges for a
@@ -343,6 +350,7 @@ async def _fetch_routing_map(
         """
         current_previous_map = previous_routing_map
         incomplete_attempt_count = 0
+        overlap_attempt_count = 0
 
         while True:
             request_kwargs = dict(kwargs)
@@ -398,6 +406,24 @@ async def _fetch_routing_map(
                     continue
 
                 raise
+            except _OverlapDetected:
+                # The full-load builder reported overlapping ranges. Apply
+                # the retry-on-overlap policy: ``_handle_overlap_retry_decision``
+                # either returns a backoff to sleep or raises ``CosmosHttpResponseError``
+                # (503) when the attempt budget is exhausted. Reset
+                # ``current_previous_map`` to ``None`` so the next iteration
+                # runs the full-load path regardless of which path tripped
+                # the overlap — we do not want to keep retrying an incremental
+                # fetch against the same inconsistent base.
+                overlap_attempt_count += 1
+                backoff = _handle_overlap_retry_decision(
+                    overlap_attempt_count=overlap_attempt_count,
+                    collection_link=collection_link,
+                    logger=logger,
+                )
+                await asyncio.sleep(backoff)
+                current_previous_map = None
+                continue
 
     async def get_range_by_partition_key_range_id(
             self,

sdk/cosmos/azure-cosmos/azure/cosmos/_routing/collection_routing_map.py

@@ -29,6 +29,30 @@
 from azure.cosmos._routing import routing_range
 from azure.cosmos._routing.routing_range import PartitionKeyRange, PKRange
 
+
+class _OverlapDetected(Exception):
+    """Sentinel raised by :func:`_build_routing_map_from_ranges` when the
+    ``/pkranges`` response contains overlapping ranges that the SDK cannot
+    reconcile from a single snapshot.
+
+    Distinct from a plain ``ValueError`` so the caller can identify the
+    overlap case explicitly and apply the retry-on-overlap policy instead
+    of treating it as a fatal programmer error. Overlap on this path is
+    expected to be transient (a paginated ``/pkranges`` response that is
+    not snapshot-isolated across gateway nodes), so a short retry budget
+    with backoff is sufficient.
+
+    The caller (``_fetch_routing_map`` on both sync and async cache
+    providers) catches this sentinel, sleeps briefly, and retries the
+    fetch. After a small number of failed retries it surfaces a typed
+    transient HTTP error so the upstream retry policy can take over. The
+    sentinel is intentionally not allowed to escape to the query layer
+    as a bare ``ValueError`` — that path would otherwise convert into a
+    silent empty-result return at ``get_overlapping_ranges`` (which
+    treats a ``None`` routing map as "no ranges"), masking the failure
+    as a correctness bug.
+    """
+
 # pylint: disable=line-too-long
 class CollectionRoutingMap(object):
     """Stores partition key ranges in an efficient way with some additional
@@ -196,7 +220,23 @@ def is_complete_set_of_range(ordered_partition_key_range_list):
 
                 if not isComplete:
                     if previousRange[PartitionKeyRange.MaxExclusive] > currentRange[PartitionKeyRange.MinInclusive]:
-                        raise ValueError("Ranges overlap")
+                        # Include the offending pair in the message so whoever
+                        # investigates the next occurrence has actionable
+                        # diagnostics without having to reproduce the failure
+                        # under a debugger. Keep the literal substring
+                        # "Ranges overlap" for backwards compatibility with
+                        # any caller that pattern-matches on it.
+                        raise ValueError(
+                            "Ranges overlap: previous range id={!r} ({!r} -> {!r}) "
+                            "overlaps current range id={!r} ({!r} -> {!r})".format(
+                                previousRange.get(PartitionKeyRange.Id),
+                                previousRange[PartitionKeyRange.MinInclusive],
+                                previousRange[PartitionKeyRange.MaxExclusive],
+                                currentRange.get(PartitionKeyRange.Id),
+                                currentRange[PartitionKeyRange.MinInclusive],
+                                currentRange[PartitionKeyRange.MaxExclusive],
+                            )
+                        )
                     break
 
         return isComplete
@@ -270,7 +310,10 @@ def _build_routing_map_from_ranges(
 
     Filters out parent (gone) ranges and validates that the remaining ranges
     form a complete, gap-free partition key space. Returns None if the ranges
-    are incomplete.
+    are incomplete (gap), and raises ``_OverlapDetected`` if the ranges
+    overlap — the caller is expected to retry the fetch on the overlap case,
+    since overlap on the full-load path is treated as a transient gateway
+    inconsistency rather than a permanent input error.
 
     This is shared between the sync and async PartitionKeyRangeCache to avoid
     code duplication — the logic is purely synchronous.
@@ -280,9 +323,27 @@ def _build_routing_map_from_ranges(
     :param str new_etag: The ETag from the change feed response.
     :param str collection_link: The collection link, used for log messages.
     :param logging.Logger _logger: Logger instance for error reporting.
-    :return: A complete CollectionRoutingMap, or None if the ranges are incomplete.
+    :return: A complete CollectionRoutingMap, or None if the ranges are incomplete (gap).
     :rtype: Optional[CollectionRoutingMap]
+    :raises _OverlapDetected: If the ranges contain an overlap that could not
+        be resolved from this single snapshot. The caller should retry the
+        fetch; see :class:`_OverlapDetected` for the rationale.
     """
+    # Dedup the input by id BEFORE parent filtering and validation. At high
+    # partition counts the /pkranges response is paginated, and pagination
+    # is not snapshot-isolated across gateway nodes — consecutive pages can
+    # legitimately return the same range id when the page boundary falls
+    # between two nodes with one-tick-out-of-sync caches. Without this dedup
+    # the duplicate would survive into the sortedRanges list inside
+    # CompleteRoutingMap and trip the overlap check on two identical entries.
+    # Last-write-wins is safe: duplicates describe the same logical range
+    # and any later occurrence carries the same id, min/max, and (when
+    # present) the more-complete metadata such as ``parents``.
+    deduped_by_id: dict = {}
+    for r in ranges:
+        deduped_by_id[r[PartitionKeyRange.Id]] = r
+    ranges = list(deduped_by_id.values())
+
     gone_range_ids = set()
     for r in ranges:
         if PartitionKeyRange.Parents in r and r[PartitionKeyRange.Parents]:
@@ -294,11 +355,27 @@ def _build_routing_map_from_ranges(
     ]
     range_tuples = [(r, True) for r in filtered_ranges]
 
-    routing_map = CollectionRoutingMap.CompleteRoutingMap(
-        range_tuples,
-        collection_id,
-        new_etag
-    )
+    try:
+        routing_map = CollectionRoutingMap.CompleteRoutingMap(
+            range_tuples,
+            collection_id,
+            new_etag
+        )
+    except ValueError as overlap_error:
+        # ``is_complete_set_of_range`` raises ``ValueError("Ranges overlap: ...")``
+        # when the post-filter range list still contains an overlap. Convert
+        # it to ``_OverlapDetected`` so the caller can apply the retry-on-
+        # overlap policy. The bare ``ValueError`` must NOT escape to the
+        # cache layer: that path converts into a silent empty-result return
+        # at ``get_overlapping_ranges`` (which treats ``None`` from the
+        # cache as "no ranges"), masking the failure as a correctness bug.
+        _logger.warning(
+            "Full load of routing map for collection '%s' detected overlapping "
+            "partition key ranges: %s. Signalling caller to retry the "
+            "/pkranges fetch.",
+            collection_link, str(overlap_error),
+        )
+        raise _OverlapDetected() from overlap_error
 
     if not routing_map:
         _logger.error(