Cosmos: Adds GlobalEndpointManager, ClientRetryPolicy and LocationCache To Support Cross-Regional Routing (#3348)

## Description:
 
This PR is introducing `GlobalEndpointManager`, `ClientRetryPolicy`,
`LocationCache` and `AccountProperties` to support cross-regional
routing in the rust SDK. Some of the key design decisions are
highlighted below:

- We used [Moka ](https://docs.rs/moka/latest/moka/)as a caching library
which supports thread safe access of the cached data and uses LRFU based
cache eviction mechanism.

- The cache refresh has been done on a TTL (time to live) model, where
each incoming request has the opportunity to refresh `AccountProperties`
and `LocationCache`, if the TTL expires or `force_refresh` is passed as
`True`.

- We haven't took the background refresh route yet, since we will need
to be a little careful and optimized for the rust bg task management.
However, our long term goal is to refresh the cache in the background,
so that we do not penalize an incoming request to refresh the cache.

The below flow diagram depicts the entire component interactions to
better understand how the request will be made.

## Flow Diagram:

```mermaid
sequenceDiagram
    participant J as ContainerClient
    participant A as CosmosPipeline
    participant B as RetryHandler
    participant C as ClientRetryPolicy
    participant D as Core Pipeline <br> send_raw <br> through delegate
    participant E as GlobalEndpointManager
    participant F as LocationCache


    J-->>A: 1. send(CosmosRequest)
    A->>B: 2. send<Sender, Fut>()
    loop Retry Iterations with force-refresh flag = false/ true
        B->>C: 3. before_send_request(request)
        C->>E: 4. refresh_location_async() <br> Refresh Cache only when TTL Expires OR <br> force_refresh is requested. <br><br> NO-OP when either of the above is false.
        E->>E: 5. get_database_account() <br> Fetch Account Properties <br> from Gateway.
        E->>F: 6. on_database_account_read() <br> Refresh Location Cache
        C->>E: 7. resolve_service_endpoint() <br> Fetch and set the next endpoint in <br> CosmosRequest to route request.
        B->>D: 8. Call sender delegate(request)
        B->>C: 9. Invoke should_retry(result)
        B->>B: 10. RetryAfter <br> (TimeSpan.Zero)
    end
        B->>J: 11. Return result if No Retry
```

## Closing issues

Fixes #3175

Author: kundadebdatta

Cargo.lock

@@ -153,6 +153,7 @@ wasm-bindgen-futures = "0.4"
 wasm-bindgen-test = "0.3"
 zerofrom = "0.1.5"
 zip = { version = "4.0.0", default-features = false, features = ["deflate"] }
+moka = { version = "0.12", features = ["future"] }
 
 [workspace.lints.clippy]
 large_futures = "deny"

Cargo.toml

@@ -72,3 +72,4 @@ url
 uuid
 zerofrom
 zip
+moka

eng/dict/crates.txt

@@ -8,6 +8,9 @@ udfs
 backoff
 pluggable
 cloneable
+moka
+substatus
+retryable
 
 # Cosmos' docs all use "Autoscale" as a single word, rather than a compound "AutoScale" or "Auto Scale"
 autoscale

sdk/cosmos/.dict.txt

@@ -4,6 +4,7 @@
 
 ### Features Added
 
+- Added GlobalEndpointManager, LocationCache to support Cross Regional Retry.
 - Added `continuation_token` to `PagerOptions` for methods that return a `Pager`.
 
 ### Breaking Changes

sdk/cosmos/azure_data_cosmos/CHANGELOG.md

@@ -21,6 +21,7 @@ serde_json.workspace = true
 serde.workspace = true
 tracing.workspace = true
 url.workspace = true
+moka.workspace = true
 
 [dev-dependencies]
 azure_identity.workspace = true

sdk/cosmos/azure_data_cosmos/Cargo.toml

@@ -9,8 +9,9 @@ use crate::{
     DeleteContainerOptions, FeedPager, ItemOptions, PartitionKey, Query, ReplaceContainerOptions,
     ThroughputOptions,
 };
+use std::sync::Arc;
 
-use crate::cosmos_request::CosmosRequestBuilder;
+use crate::cosmos_request::CosmosRequest;
 use crate::operation_context::OperationType;
 use azure_core::http::response::Response;
 use serde::{de::DeserializeOwned, Serialize};
@@ -22,19 +23,19 @@ use serde::{de::DeserializeOwned, Serialize};
 pub struct ContainerClient {
     link: ResourceLink,
     items_link: ResourceLink,
-    pipeline: CosmosPipeline,
+    pipeline: Arc<CosmosPipeline>,
 }
 
 impl ContainerClient {
     pub(crate) fn new(
-        pipeline: CosmosPipeline,
+        pipeline: Arc<CosmosPipeline>,
         database_link: &ResourceLink,
         container_id: &str,
     ) -> Self {
         let link = database_link
             .feed(ResourceType::Containers)
             .item(container_id);
-        let items_link = link.feed(ResourceType::Items);
+        let items_link = link.feed(ResourceType::Documents);
 
         Self {
             link,
@@ -65,13 +66,9 @@ impl ContainerClient {
     ) -> azure_core::Result<Response<ContainerProperties>> {
         let options = options.unwrap_or_default();
         let cosmos_request =
-            CosmosRequestBuilder::new(OperationType::Read, ResourceType::Containers).build()?;
+            CosmosRequest::builder(OperationType::Read, self.link.clone()).build()?;
         self.pipeline
-            .send(
-                cosmos_request,
-                self.link.clone(),
-                options.method_options.context,
-            )
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -111,14 +108,11 @@ impl ContainerClient {
         options: Option<ReplaceContainerOptions<'_>>,
     ) -> azure_core::Result<Response<ContainerProperties>> {
         let options = options.unwrap_or_default();
-        let builder = CosmosRequestBuilder::new(OperationType::Replace, ResourceType::Containers);
-        let cosmos_request = builder.json(&properties).build()?;
+        let cosmos_request = CosmosRequest::builder(OperationType::Replace, self.link.clone())
+            .json(&properties)
+            .build()?;
         self.pipeline
-            .send(
-                cosmos_request,
-                self.link.clone(),
-                options.method_options.context,
-            )
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -181,14 +175,10 @@ impl ContainerClient {
         options: Option<DeleteContainerOptions<'_>>,
     ) -> azure_core::Result<Response<()>> {
         let options = options.unwrap_or_default();
-        let builder = CosmosRequestBuilder::new(OperationType::Delete, ResourceType::Containers);
-        let cosmos_request = builder.build()?;
+        let cosmos_request =
+            CosmosRequest::builder(OperationType::Delete, self.link.clone()).build()?;
         self.pipeline
-            .send(
-                cosmos_request,
-                self.link.clone(),
-                options.method_options.context,
-            )
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -264,19 +254,14 @@ impl ContainerClient {
         options: Option<ItemOptions<'_>>,
     ) -> azure_core::Result<Response<()>> {
         let options = options.clone().unwrap_or_default();
-        let builder = CosmosRequestBuilder::new(OperationType::Create, ResourceType::Items);
-        let cosmos_request = builder
+        let cosmos_request = CosmosRequest::builder(OperationType::Create, self.items_link.clone())
             .headers(&options)
             .json(&item)
             .partition_key(partition_key.into())
             .build()?;
 
         self.pipeline
-            .send(
-                cosmos_request,
-                self.items_link.clone(),
-                options.method_options.context,
-            )
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -354,15 +339,14 @@ impl ContainerClient {
     ) -> azure_core::Result<Response<()>> {
         let link = self.items_link.item(item_id);
         let options = options.clone().unwrap_or_default();
-        let builder = CosmosRequestBuilder::new(OperationType::Replace, ResourceType::Items);
-        let cosmos_request = builder
+        let cosmos_request = CosmosRequest::builder(OperationType::Replace, link)
             .headers(&options)
             .json(&item)
             .partition_key(partition_key.into())
             .build()?;
 
         self.pipeline
-            .send(cosmos_request, link, options.method_options.context)
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -441,19 +425,14 @@ impl ContainerClient {
         options: Option<ItemOptions<'_>>,
     ) -> azure_core::Result<Response<()>> {
         let options = options.clone().unwrap_or_default();
-        let builder = CosmosRequestBuilder::new(OperationType::Upsert, ResourceType::Items);
-        let cosmos_request = builder
+        let cosmos_request = CosmosRequest::builder(OperationType::Upsert, self.items_link.clone())
             .headers(&options)
             .json(&item)
             .partition_key(partition_key.into())
             .build()?;
 
         self.pipeline
-            .send(
-                cosmos_request,
-                self.items_link.clone(),
-                options.method_options.context,
-            )
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -499,14 +478,13 @@ impl ContainerClient {
         options.enable_content_response_on_write = true;
 
         let link = self.items_link.item(item_id);
-        let builder = CosmosRequestBuilder::new(OperationType::Read, ResourceType::Items);
-        let cosmos_request = builder
+        let cosmos_request = CosmosRequest::builder(OperationType::Read, link)
             .partition_key(partition_key.into())
             .headers(&options)
             .build()?;
 
         self.pipeline
-            .send(cosmos_request, link, options.method_options.context)
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -538,15 +516,13 @@ impl ContainerClient {
     ) -> azure_core::Result<Response<()>> {
         let link = self.items_link.item(item_id);
         let options = options.clone().unwrap_or_default();
-
-        let builder = CosmosRequestBuilder::new(OperationType::Delete, ResourceType::Items);
-        let cosmos_request = builder
+        let cosmos_request = CosmosRequest::builder(OperationType::Delete, link)
             .partition_key(partition_key.into())
             .headers(&options)
             .build()?;
 
         self.pipeline
-            .send(cosmos_request, link, options.method_options.context)
+            .send(cosmos_request, options.method_options.context)
             .await
     }
 
@@ -614,15 +590,14 @@ impl ContainerClient {
     ) -> azure_core::Result<Response<()>> {
         let options = options.clone().unwrap_or_default();
         let link = self.items_link.item(item_id);
-        let builder = CosmosRequestBuilder::new(OperationType::Patch, ResourceType::Items);
-        let cosmos_request = builder
+        let cosmos_request = CosmosRequest::builder(OperationType::Patch, link)
             .partition_key(partition_key.into())
             .headers(&options)
             .json(&patch)
             .build()?;
 
         self.pipeline
-            .send(cosmos_request, link, options.method_options.context)
+            .send(cosmos_request, options.method_options.context)
             .await
     }