[Cosmos] Fix pkranges fetch: use container name for URL, RID for cache key (#4047)

## Summary

Fixes [#4031](#4031) —
the pkranges fetch was using the collection **RID** in a name-based URL
hierarchy (`dbs/perfdb/colls/<RID>/pkranges`), which Cosmos DB rejects
with 404 because mixed name/RID addressing is not supported.

The previous fix (#4041) corrected URL encoding (`.item()` →
`.item_by_rid()`) but did not fix the fundamental mixed-addressing
issue. This PR resolves it by passing the container **name** for URL
construction while keeping the **RID** as the cache key and request
context value.

## Root Cause

PR #4005 changed `container_connection.rs::send()` to pass
`self.container_ref.rid()` to `pk_range_cache.try_lookup()`. The cache
used this RID to build the pkranges URL:

```
dbs/perfdb/colls/pLLZAIuPigw=/pkranges
     ^^^^^^       ^^^^^^^^^^^^^
     NAME          RID  ← mixed addressing → 404
```

All other SDK and driver operations use name-based URLs. The pkranges
fetch was the only code path using a RID in a name-based link hierarchy.

## Impact (observed on continuous benchmarks)

- **1.8M 404 requests/hour** from failed pkranges fetches
- Errors silently swallowed by `try_lookup` → `Ok(routing_map.ok())`  
- Errors not cached → retried on every request (write lock contention on
AsyncCache)
- Loss of client-side partition key routing → gateway must route all
requests
- Throughput regression from ~110M to ~4.2M requests/hour

## Changes

### `partition_key_range_cache.rs`
- Added `collection_name: &str` parameter to `try_lookup`,
`get_routing_map_for_collection`, `resolve_partition_key_range_by_id`,
and `resolve_overlapping_ranges`
- Changed `.item_by_rid(collection_rid)` → `.item(collection_name)` for
pkranges URL construction
- Cache key remains the RID (`collection_rid.to_string()`)
- `resource_id` on the request remains the RID
- Updated `tracing::warn!` to include both `collection_name` and
`collection_rid`
- Replaced 3 RID-encoding unit tests with 2 tests verifying name-based
URL construction

### `container_connection.rs`
- Extracted `collection_name` from `self.container_ref.name()` alongside
existing `collection_rid`
- Passes `collection_name` to all `pk_range_cache` method calls
- `resolved_collection_rid` on request context still uses the RID
(unchanged)

### `cosmos_fault_injection.rs`
- Added `fault_injection_pkrange_readfeed_is_exercised` integration test
- Injects a transient error on `MetadataPartitionKeyRanges` ReadFeed
with hit_limit=1
- Verifies the fault rule is hit (proving pkrange fetch code path
executes)
- Verifies subsequent item operations succeed (proving end-to-end
pkrange resolution works)

## Test Results

- ✅ 31 unit tests pass (including 2 new)
- ✅ Build succeeds
- Integration test requires emulator (will run in CI)

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: 3 people

sdk/cosmos/.cspell.json

@@ -123,6 +123,7 @@
         "qself",
         "RAII",
         "reactivations",
+        "readfeed",
         "Replicaset",
         "reqs",
         "Retriable",

sdk/cosmos/azure_data_cosmos/src/fault_injection/http_client.rs

@@ -219,7 +219,10 @@ impl FaultClient {
 impl HttpClient for FaultClient {
     async fn execute_request(&self, request: &Request) -> azure_core::Result<AsyncRawResponse> {
         // Find applicable rule and clone the result if needed
-        let fault_result: Option<FaultInjectionResult> = {
+        let (fault_result, matched_rule): (
+            Option<FaultInjectionResult>,
+            Option<Arc<FaultInjectionRule>>,
+        ) = {
             let rules = self.rules.lock().unwrap();
             let mut applicable_rule_index: Option<usize> = None;
 
@@ -234,9 +237,9 @@ impl HttpClient for FaultClient {
             if let Some(index) = applicable_rule_index {
                 let rule = &rules[index];
                 rule.increment_hit_count();
-                Some(rule.result.clone())
+                (Some(rule.result.clone()), Some(Arc::clone(rule)))
             } else {
-                None
+                (None, None)
             }
         };
 
@@ -257,7 +260,23 @@ impl HttpClient for FaultClient {
                 .remove(constants::FAULT_INJECTION_OPERATION);
 
             // No fault injection or delay-only fault, proceed with actual request
-            self.inner.execute_request(&clean_request).await
+            let result = self.inner.execute_request(&clean_request).await;
+
+            // Record response status only for true spy rules: no error_type,
+            // no custom_response, and no delay. This excludes probability-skipped
+            // faults and any rule that injected a delay.
+            if let (Some(rule), Some(ref fr), Ok(ref response)) =
+                (&matched_rule, &fault_result, &result)
+            {
+                if fr.error_type.is_none()
+                    && fr.custom_response.is_none()
+                    && fr.delay == Duration::ZERO
+                {
+                    rule.record_passthrough_status(response.status());
+                }
+            }
+
+            result
         };
 
         // Apply delay after the request is sent

sdk/cosmos/azure_data_cosmos/src/fault_injection/rule.rs

@@ -4,8 +4,11 @@
 //! Defines fault injection rules that combine conditions and results.
 
 use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
+use std::sync::Mutex;
 use std::time::Instant;
 
+use azure_core::http::StatusCode;
+
 use super::condition::FaultInjectionCondition;
 use super::result::FaultInjectionResult;
 
@@ -26,8 +29,10 @@ pub struct FaultInjectionRule {
     pub id: String,
     /// Whether the rule is currently enabled.
     enabled: AtomicBool,
-    /// Number of times the rule has been matched and applied.
+    /// Number of times the rule has been matched (including matches where no fault was injected).
     hit_count: AtomicU32,
+    /// HTTP status codes of responses for matched requests that passed through without fault injection.
+    passthrough_statuses: Mutex<Vec<StatusCode>>,
 }
 
 /// Cloning snapshots the current `hit_count` and `enabled` state rather than
@@ -43,6 +48,7 @@ impl Clone for FaultInjectionRule {
             id: self.id.clone(),
             enabled: AtomicBool::new(self.enabled.load(Ordering::SeqCst)),
             hit_count: AtomicU32::new(self.hit_count.load(Ordering::SeqCst)),
+            passthrough_statuses: Mutex::new(self.passthrough_statuses.lock().unwrap().clone()),
         }
     }
 }
@@ -81,6 +87,26 @@ impl FaultInjectionRule {
     pub fn reset_hit_count(&self) {
         self.hit_count.store(0, Ordering::SeqCst);
     }
+
+    /// Records the HTTP status code of a response for a matched request that
+    /// passed through without fault injection (spy/passthrough mode).
+    pub(super) fn record_passthrough_status(&self, status: StatusCode) {
+        self.passthrough_statuses.lock().unwrap().push(status);
+    }
+
+    /// Returns the HTTP status codes of responses for matched requests that
+    /// passed through without fault injection.
+    ///
+    /// When a rule matches a request but does not inject a fault (e.g., no
+    /// `error_type` or `custom_response` is set), the real service response
+    /// status is recorded here. This enables "spy" rules that observe requests
+    /// without modifying them.
+    ///
+    /// The history grows unbounded for the lifetime of the rule. This is
+    /// designed for test scenarios with a bounded number of requests.
+    pub fn passthrough_statuses(&self) -> Vec<StatusCode> {
+        self.passthrough_statuses.lock().unwrap().clone()
+    }
 }
 
 /// Builder for creating a fault injection rule.
@@ -155,6 +181,7 @@ impl FaultInjectionRuleBuilder {
             id: self.id,
             enabled: AtomicBool::new(true),
             hit_count: AtomicU32::new(0),
+            passthrough_statuses: Mutex::new(Vec::new()),
         }
     }
 }

sdk/cosmos/azure_data_cosmos/src/handler/container_connection.rs

@@ -53,13 +53,15 @@ impl ContainerConnection {
                 || cosmos_request.resource_type == ResourceType::StoredProcedures)
         {
             let pk_def = self.container_ref.partition_key_definition();
+            let collection_name = self.container_ref.name();
             let collection_rid = self.container_ref.rid();
 
             if let Some(pk_range) = cosmos_request.partition_key_range_identity.as_ref() {
                 if !pk_range.collection_rid.is_empty() {
                     if let Some(resolved) = self
                         .pk_range_cache
                         .resolve_partition_key_range_by_id(
+                            collection_name,
                             &pk_range.collection_rid,
                             &pk_range.partition_key_range_id,
                             false,
@@ -71,7 +73,10 @@ impl ContainerConnection {
                     }
                 }
             } else if let Some(partition_key) = cosmos_request.partition_key.as_ref() {
-                let routing_map = self.pk_range_cache.try_lookup(collection_rid, None).await?;
+                let routing_map = self
+                    .pk_range_cache
+                    .try_lookup(collection_name, collection_rid, None)
+                    .await?;
 
                 if let Some(routing_map) = routing_map {
                     // Use a safe default version (2) when the service omits the version field,
@@ -94,7 +99,7 @@ impl ContainerConnection {
                             // Refresh the routing map and retry.
                             let refreshed_routing_map = self
                                 .pk_range_cache
-                                .try_lookup(collection_rid, Some(routing_map))
+                                .try_lookup(collection_name, collection_rid, Some(routing_map))
                                 .await?;
 
                             if let Some(refreshed_routing_map) = refreshed_routing_map {

sdk/cosmos/azure_data_cosmos/src/routing/partition_key_range_cache.rs

@@ -55,15 +55,20 @@ impl PartitionKeyRangeCache {
 
     pub async fn resolve_overlapping_ranges(
         &self,
+        collection_name: &str,
         collection_rid: &str,
         range: Range<String>,
         force_refresh: bool,
     ) -> Result<Option<Vec<PartitionKeyRange>>, Error> {
-        let mut routing_map = self.try_lookup(collection_rid, None).await?;
+        let mut routing_map = self
+            .try_lookup(collection_name, collection_rid, None)
+            .await?;
 
         if force_refresh {
             if let Some(previous) = routing_map.clone() {
-                routing_map = self.try_lookup(collection_rid, Some(previous)).await?;
+                routing_map = self
+                    .try_lookup(collection_name, collection_rid, Some(previous))
+                    .await?;
             }
         }
 
@@ -83,16 +88,20 @@ impl PartitionKeyRangeCache {
 
     pub async fn resolve_partition_key_range_by_id(
         &self,
-        collection_resource_id: &str,
+        collection_name: &str,
+        collection_rid: &str,
         partition_key_range_id: &str,
         force_refresh: bool,
     ) -> Option<PartitionKeyRange> {
-        let mut routing_map = self.try_lookup(collection_resource_id, None).await.ok()?;
+        let mut routing_map = self
+            .try_lookup(collection_name, collection_rid, None)
+            .await
+            .ok()?;
 
         if force_refresh {
             if let Some(previous) = routing_map.clone() {
                 routing_map = self
-                    .try_lookup(collection_resource_id, Some(previous))
+                    .try_lookup(collection_name, collection_rid, Some(previous))
                     .await
                     .ok()?;
             }
@@ -103,7 +112,7 @@ impl PartitionKeyRangeCache {
             None => {
                 info!(
                     "Routing Map Null for collection: {}, PartitionKeyRangeId: {}, forceRefresh: {}",
-                    collection_resource_id,
+                    collection_rid,
                     partition_key_range_id,
                     force_refresh
                 );
@@ -114,6 +123,7 @@ impl PartitionKeyRangeCache {
 
     pub async fn try_lookup(
         &self,
+        collection_name: &str,
         collection_rid: &str,
         previous_value: Option<CollectionRoutingMap>,
     ) -> Result<Option<CollectionRoutingMap>, Error> {
@@ -128,15 +138,19 @@ impl PartitionKeyRangeCache {
                 },
                 || async {
                     let routing_map = self
-                        .get_routing_map_for_collection(collection_rid, previous_value.clone())
+                        .get_routing_map_for_collection(
+                            collection_name,
+                            collection_rid,
+                            previous_value.clone(),
+                        )
                         .await?;
                     match routing_map {
                         Some(map) => Ok(map),
                         None => Err(Error::new(
                             azure_core::error::ErrorKind::Other,
                             format!(
-                                "Failed to get routing map for collection: {}",
-                                collection_rid
+                                "Failed to get routing map for collection: {} (rid: {})",
+                                collection_name, collection_rid
                             ),
                         )),
                     }
@@ -146,6 +160,7 @@ impl PartitionKeyRangeCache {
 
         if let Err(ref e) = routing_map {
             tracing::warn!(
+                collection_name,
                 collection_rid,
                 error = %e,
                 "Failed to fetch routing map for collection"
@@ -168,6 +183,7 @@ impl PartitionKeyRangeCache {
 
     async fn get_routing_map_for_collection(
         &self,
+        collection_name: &str,
         collection_rid: &str,
         previous_routing_map: Option<CollectionRoutingMap>,
     ) -> Result<Option<CollectionRoutingMap>, Error> {
@@ -196,7 +212,7 @@ impl PartitionKeyRangeCache {
             let pk_range_link = self
                 .database_link
                 .feed(ResourceType::Containers)
-                .item_by_rid(collection_rid)
+                .item(collection_name)
                 .feed(ResourceType::PartitionKeyRanges);
             let response = self
                 .execute_partition_key_range_read_change_feed(
@@ -681,60 +697,51 @@ mod tests {
         assert_eq!(range.target_throughput.unwrap(), 1000.0);
     }
 
-    // Tests verifying that the pkranges resource link uses item_by_rid() so that
-    // collection RIDs (which are base64-encoded and can contain '=', '+', '/') are
-    // not URL-percent-encoded. Using item() would encode '=' to '%3D', causing 404s.
-
     #[test]
-    fn pkranges_link_rid_with_equals_is_not_encoded() {
-        // RIDs like "pLLZAIuPigw=" contain '=' which item() would encode to '%3D'.
-        // item_by_rid() must preserve it as-is.
-        let collection_rid = "pLLZAIuPigw=";
+    fn pkranges_link_uses_container_name_not_rid() {
+        // The pkranges URL must use the container NAME (not RID) to match
+        // the name-based database_link. Mixed name/RID addressing causes 404s.
         let database_link = ResourceLink::root(ResourceType::Databases).item("perfdb");
         let pk_range_link = database_link
             .feed(ResourceType::Containers)
-            .item_by_rid(collection_rid)
+            .item("perfcontainer")
             .feed(ResourceType::PartitionKeyRanges);
 
-        // Correct: '=' preserved, not encoded to '%3D'
         assert_eq!(
-            "dbs/perfdb/colls/pLLZAIuPigw=/pkranges",
+            "dbs/perfdb/colls/perfcontainer/pkranges",
             pk_range_link.path()
         );
     }
 
     #[test]
-    fn pkranges_link_item_encodes_equals_incorrectly() {
-        // Demonstrates the bug: item() URL-encodes '=' to '%3D', producing a path
-        // that Cosmos DB cannot find (404).
-        let collection_rid = "pLLZAIuPigw=";
+    fn pkranges_link_with_rid_causes_mixed_addressing() {
+        // Demonstrates why using RID in a name-based link is wrong:
+        // the resulting URL mixes name-based database with RID-based container,
+        // which Cosmos DB rejects with 404.
         let database_link = ResourceLink::root(ResourceType::Databases).item("perfdb");
-        let pk_range_link_wrong = database_link
+
+        // Using item_by_rid with a RID in a name-based hierarchy produces
+        // a mixed-addressing URL that Cosmos DB cannot resolve.
+        let pk_range_link_mixed = database_link
             .feed(ResourceType::Containers)
-            .item(collection_rid)
+            .item_by_rid("pLLZAIuPigw=")
             .feed(ResourceType::PartitionKeyRanges);
 
-        // Wrong: '=' is encoded to '%3D', causing 404 from Cosmos DB
-        assert!(
-            pk_range_link_wrong.path().contains("%3D"),
-            "item() should URL-encode '=' to '%3D'"
-        );
+        // This path uses name for DB but RID for collection — Cosmos DB returns 404
         assert_eq!(
-            "dbs/perfdb/colls/pLLZAIuPigw%3D/pkranges",
-            pk_range_link_wrong.path()
+            "dbs/perfdb/colls/pLLZAIuPigw=/pkranges",
+            pk_range_link_mixed.path()
         );
-    }
 
-    #[test]
-    fn pkranges_link_rid_with_plus_is_not_encoded() {
-        // RIDs may also contain '+' (base64 char). item_by_rid() must preserve it.
-        let collection_rid = "AB+CD/EF==";
-        let database_link = ResourceLink::root(ResourceType::Databases).item("mydb");
-        let pk_range_link = database_link
+        // Using item() with a name in a name-based hierarchy produces a valid URL.
+        let pk_range_link_correct = database_link
             .feed(ResourceType::Containers)
-            .item_by_rid(collection_rid)
+            .item("perfcontainer")
             .feed(ResourceType::PartitionKeyRanges);
 
-        assert_eq!("dbs/mydb/colls/AB+CD/EF==/pkranges", pk_range_link.path());
+        assert_eq!(
+            "dbs/perfdb/colls/perfcontainer/pkranges",
+            pk_range_link_correct.path()
+        );
     }
 }