Azure · NaluTripician · Oct 26, 2023 · Oct 26, 2023 · Nov 27, 2023 · Dec 12, 2023
@@ -13,7 +13,7 @@ namespace Microsoft.Azure.Cosmos
     using System.Net.Http;
     using System.Net.Security;
     using System.Security.Cryptography.X509Certificates;
-    using Microsoft.Azure.Cosmos.Fluent;
+    using Microsoft.Azure.Cosmos.Fluent;
     using Microsoft.Azure.Documents;
     using Microsoft.Azure.Documents.Client;
     using Newtonsoft.Json;
@@ -650,7 +650,16 @@ public Func<HttpClient> HttpClientFactory
 
                 this.httpClientFactory = value;
             }
-        }
+        }
+        /// <summary>
+        /// Availability Strategy to be used for periods of high latency
+        /// </summary>
+#if PREVIEW
+        public 
+#else
+        internal
+#endif
+        AvailabilityStrategy AvailabilityStrategy { get; set; }
 
         /// <summary>
         /// Enable partition key level failover

@@ -114,6 +114,7 @@ internal partial class DocumentClient : IDisposable, IAuthorizationTokenProvider
 
         private readonly bool IsLocalQuorumConsistency = false;
         private readonly bool isReplicaAddressValidationEnabled;
+        private readonly AvailabilityStrategy availabilityStrategy;
 
         //Fault Injection
         private readonly IChaosInterceptorFactory chaosInterceptorFactory;
@@ -439,6 +440,7 @@ internal partial class DocumentClient : IDisposable, IAuthorizationTokenProvider
         /// <param name="cosmosClientId"></param>
         /// <param name="remoteCertificateValidationCallback">This delegate responsible for validating the third party certificate. </param>
         /// <param name="cosmosClientTelemetryOptions">This is distributed tracing flag</param>
+        /// <param name="availabilityStrategy">This is the availability strategy for the client</param>"
         /// <param name="chaosInterceptorFactory">This is the chaos interceptor used for fault injection</param>
         /// <remarks>
         /// The service endpoint can be obtained from the Azure Management Portal.
@@ -468,6 +470,7 @@ internal partial class DocumentClient : IDisposable, IAuthorizationTokenProvider
                               string cosmosClientId = null,
                               RemoteCertificateValidationCallback remoteCertificateValidationCallback = null,
                               CosmosClientTelemetryOptions cosmosClientTelemetryOptions = null,
+                              AvailabilityStrategy availabilityStrategy = null,
                               IChaosInterceptorFactory chaosInterceptorFactory = null)
         {
             if (sendingRequestEventArgs != null)
@@ -491,6 +494,7 @@ internal partial class DocumentClient : IDisposable, IAuthorizationTokenProvider
             this.transportClientHandlerFactory = transportClientHandlerFactory;
             this.IsLocalQuorumConsistency = isLocalQuorumConsistency;
             this.initTaskCache = new AsyncCacheNonBlocking<string, bool>(cancellationToken: this.cancellationTokenSource.Token);
+            this.availabilityStrategy = availabilityStrategy;
             this.chaosInterceptorFactory = chaosInterceptorFactory;
             this.chaosInterceptor = chaosInterceptorFactory?.CreateInterceptor(this);
 

@@ -12,7 +12,7 @@ namespace Microsoft.Azure.Cosmos.Fluent
     using System.Threading.Tasks;
     using global::Azure;
     using global::Azure.Core;
-    using Microsoft.Azure.Cosmos.Core.Trace;
+    using Microsoft.Azure.Cosmos.Core.Trace;
     using Microsoft.Azure.Documents;
     using Microsoft.Azure.Documents.Client;
 
@@ -683,6 +683,22 @@ internal CosmosClientBuilder WithApiType(ApiType apiType)
         {
             this.clientOptions.ApiType = apiType;
             return this;
+        }
+
+        /// <summary>
+        /// Availability Stragey to be used for periods of high latency
+        /// </summary>
+        /// <param name="strategy"></param>
+        /// <returns>The CosmosClientBuilder</returns>
+#if PREVIEW
+        public
+#else
+        internal
+#endif
+        CosmosClientBuilder WithAvailibilityStrategy(AvailabilityStrategy strategy)
+        {
+            this.clientOptions.AvailabilityStrategy = strategy;
+            return this;
         }
 
         /// <summary>

@@ -12,6 +12,7 @@ namespace Microsoft.Azure.Cosmos.Handlers
     using System.Net.Http;
     using System.Threading;
     using System.Threading.Tasks;
+    using Microsoft.Azure.Cosmos.Diagnostics;
     using Microsoft.Azure.Cosmos.Routing;
     using Microsoft.Azure.Cosmos.Tracing;
     using Microsoft.Azure.Documents;
@@ -38,6 +39,7 @@ internal class RequestInvokerHandler : RequestHandler
             Cosmos.PriorityLevel? requestedClientPriorityLevel)
         {
             this.client = client;
+
             this.RequestedClientConsistencyLevel = requestedClientConsistencyLevel;       
             this.RequestedClientPriorityLevel = requestedClientPriorityLevel;
         }
@@ -52,11 +54,8 @@ internal class RequestInvokerHandler : RequestHandler
             }
 
             RequestOptions promotedRequestOptions = request.RequestOptions;
-            if (promotedRequestOptions != null)
-            {
-                // Fill request options
-                promotedRequestOptions.PopulateRequestOptions(request);
-            }
+            // Fill request options
+            promotedRequestOptions?.PopulateRequestOptions(request);
 
             // Adds the NoContent header if not already added based on Client Level flag
             if (RequestInvokerHandler.ShouldSetNoContentResponseHeaders(
@@ -79,7 +78,57 @@ internal class RequestInvokerHandler : RequestHandler
 
             await request.AssertPartitioningDetailsAsync(this.client, cancellationToken, request.Trace);
             this.FillMultiMasterContext(request);
-            return await base.SendAsync(request, cancellationToken);
+
+            if (this.CanUseAvailabilityStrategy(request))
+            {
+                AvailabilityStrategy strategy = request.RequestOptions?.AvailabilityStrategy ?? this.client.ClientOptions.AvailabilityStrategy;
+
+                return await strategy.ExecuteAvailablityStrategyAsync(
+                            this.BaseSendAsync,
+                            this.client,
+                            request,
+                            cancellationToken);
+            }
+
+            return await this.BaseSendAsync(request, cancellationToken);
+        }
+
+        /// <summary>
+        /// This method determines if there is an availability strategy that the request can use.
+        /// Note that the request level availability strategy options override the client level options.
+        /// </summary>
+        /// <param name="request"></param>
+        /// <returns>whether the request should be a parallel hedging request.</returns>
+        public bool CanUseAvailabilityStrategy(RequestMessage request)
+        {
+            //No availability strategy options
+            if (request.RequestOptions?.AvailabilityStrategy == null && this.client.ClientOptions.AvailabilityStrategy == null)
+            {
+                return false;
+            }
+
+            AvailabilityStrategy strategyOptions = request.RequestOptions?.AvailabilityStrategy
+                    ?? this.client.ClientOptions.AvailabilityStrategy;
+
+            //Request level availability strategy options are disabled/override client level
+            if (!strategyOptions.Enabled())
+            {
+                return false;
+            }
+
+            return true;
+        }
+
+        public virtual async Task<ResponseMessage> BaseSendAsync(
+            RequestMessage request,
+            CancellationToken cancellationToken)
+        {
+            ResponseMessage response = await base.SendAsync(request, cancellationToken);
+            if (request.RequestOptions?.ExcludeRegions != null)
+            {
+                ((CosmosTraceDiagnostics)response.Diagnostics).Value.AddDatum("ExcludedRegions", request.RequestOptions.ExcludeRegions);
+            }
+            return response;
         }
 
         public virtual async Task<T> SendAsync<T>(

@@ -5,10 +5,11 @@
 namespace Microsoft.Azure.Cosmos
 {
     using System;
-    using System.Collections.Generic;
+    using System.Collections.Generic;
     using System.Diagnostics;
     using System.Globalization;
-    using System.IO;
+    using System.IO;
+    using System.Linq;
     using System.Net;
     using System.Net.Http;
     using System.Threading;
@@ -62,6 +63,28 @@ public RequestMessage(HttpMethod method, Uri requestUri)
             this.Method = method;
             this.RequestUriString = requestUriString;
             this.Trace = trace ?? throw new ArgumentNullException(nameof(trace));
+        }
+
+        /// <summary>
+        /// Create a <see cref="RequestMessage"/>, used for Clone() method. 
+        /// </summary>
+        /// <param name="method">The http method</param>
+        /// <param name="requestUriString">The requested URI</param>
+        /// <param name="trace">The trace node to append traces to.</param>
+        /// <param name="headers">The headers to use.</param>
+        /// <param name="properties">The properties to use.</param>
+        private RequestMessage(
+            HttpMethod method,
+            string requestUriString,
+            ITrace trace,
+            Headers headers,
+            Dictionary<string, object> properties)
+        {
+            this.Method = method;
+            this.RequestUriString = requestUriString;
+            this.Trace = trace ?? throw new ArgumentNullException(nameof(trace));
+            this.headers = new Lazy<Headers>(() => headers);
+            this.properties = new Lazy<Dictionary<string, object>>(() => properties);
         }
 
         /// <summary>
@@ -138,7 +161,8 @@ public virtual Stream Content
             this.Headers.PartitionKey == null;
 
         internal string ContainerId { get; set; }
-        internal string DatabaseId { get; set; }
+        internal string DatabaseId { get; set; }
+        internal Uri LocationEndpointToRoute { get; set; }
 
         /// <summary>
         /// Request properties Per request context available to handlers. 
@@ -274,7 +298,11 @@ internal DocumentServiceRequest ToDocumentServiceRequest()
                 serviceRequest.UseStatusCodeForFailures = true;
                 serviceRequest.UseStatusCodeFor429 = true;
                 serviceRequest.Properties = this.Properties;
-                this.DocumentServiceRequest = serviceRequest;
+                this.DocumentServiceRequest = serviceRequest;
+                if (this.LocationEndpointToRoute != null)
+                {
+                    this.DocumentServiceRequest.RequestContext.RouteToLocation(this.LocationEndpointToRoute);
+                }
             }
 
             // Routing to a particular PartitionKeyRangeId
@@ -286,6 +314,58 @@ internal DocumentServiceRequest ToDocumentServiceRequest()
             this.DocumentServiceRequest.RequestContext.ExcludeRegions = this.RequestOptions?.ExcludeRegions;
             this.OnBeforeRequestHandler(this.DocumentServiceRequest);
             return this.DocumentServiceRequest;
+        }
+
+        /// <summary>
+        /// Clone the request message
+        /// </summary>
+        /// <returns>a cloned copy of the RequestMessage</returns>
+        internal RequestMessage Clone(ITrace newTrace)
+        {
+            RequestMessage clone = new RequestMessage(
+                this.Method,
+                this.RequestUriString,
+                newTrace,
+                this.Headers.Clone(),
+                this.Properties.ToDictionary(entry => entry.Key, entry => entry.Value));
+
+            if (this.Content != null)
+            {
+                Stream cloneContent = new MemoryStream((int)this.Content.Length);
+                this.Content.Position = 0;
+                this.Content.CopyTo(cloneContent);
+                clone.Content = cloneContent;
+            }
+
+            if (this.RequestOptions != null)
+            {
+                clone.RequestOptions = this.RequestOptions.ShallowCopy();
+            }
+
+            clone.ResourceType = this.ResourceType;
+
+            clone.OperationType = this.OperationType;
+
+            if (this.PartitionKeyRangeId != null)
+            {
+                clone.PartitionKeyRangeId = string.IsNullOrEmpty(this.PartitionKeyRangeId.CollectionRid)
+                    ? new PartitionKeyRangeIdentity(this.PartitionKeyRangeId.PartitionKeyRangeId)
+                    : new PartitionKeyRangeIdentity(this.PartitionKeyRangeId.CollectionRid, this.PartitionKeyRangeId.PartitionKeyRangeId);
+            }
+
+            clone.UseGatewayMode = this.UseGatewayMode;
+
+            clone.ContainerId = this.ContainerId;
+
+            clone.DatabaseId = this.DatabaseId;
+
+            return clone;
+        }
+
+        internal void RouteToLocation(Uri location)
+        {
+            this.LocationEndpointToRoute = location;
+            this.DocumentServiceRequest?.RequestContext.RouteToLocation(location);
         }
 
         private static Dictionary<string, object> CreateDictionary()

@@ -6,7 +6,7 @@ namespace Microsoft.Azure.Cosmos
 {
     using System;
     using System.Collections;
-    using System.Collections.Generic;
+    using System.Collections.Generic;
     using System.Globalization;
     using Microsoft.Azure.Documents;
     using Microsoft.Azure.Documents.Collections;
@@ -285,6 +285,11 @@ internal Headers(INameValueCollection nameValueCollection)
                 HttpResponseHeadersWrapper httpResponseHeaders => httpResponseHeaders,
                 _ => new NameValueResponseHeaders(nameValueCollection),
             };
+        }
+
+        internal Headers(CosmosMessageHeadersInternal cosmosMessageHeaders)
+        {
+            this.CosmosMessageHeaders = cosmosMessageHeaders;
         }
 
         /// <summary>
@@ -404,6 +409,21 @@ public virtual T GetHeaderValue<T>(string headerName)
         {
             return this.CosmosMessageHeaders.GetHeaderValue<T>(headerName);
         }
+
+        /// <summary>
+        /// Clones the current <see cref="Headers"/>.
+        /// </summary>
+        /// <returns>a cloned copy of the current <see cref="Headers"/></returns>
+        public Headers Clone()
+        {
+            StoreRequestHeaders clone = new StoreRequestHeaders();
+            foreach (string key in this.CosmosMessageHeaders.AllKeys())
+            {
+                clone.Add(key, this.CosmosMessageHeaders[key]);
+            }
+
+            return new Headers(clone);
+        }
 
         /// <summary>
         /// Enumerates all the HTTP headers names in the <see cref="Headers"/>.

@@ -5,7 +5,8 @@
 namespace Microsoft.Azure.Cosmos
 {
     using System;
-    using System.Collections.Generic;
+    using System.Collections.Generic;
+    using Microsoft.Azure.Cosmos.Routing;
     using Microsoft.Azure.Documents;
     using Telemetry;
 
@@ -69,7 +70,20 @@ public class RequestOptions
         /// This can be used to route a request to a specific region by excluding all other regions.
         /// If all regions are excluded, then the request will be routed to the primary/hub region.
         /// </summary>
-        public List<string> ExcludeRegions { get; set; }
+        public List<string> ExcludeRegions { get; set; }
+
+        /// <summary>
+        /// Cosmos availability strategy.
+        /// Availability strategy allows the SDK to send out additional cross region requests to help 
+        /// reduce latency and increase availability. Currently there is one type of availability strategy, parallel request hedging.
+        /// If there is a globally enabled availability strategy, setting one in the request options will override the global one.
+        /// </summary>
+#if PREVIEW
+        public
+#else
+        internal
+#endif
+        AvailabilityStrategy AvailabilityStrategy { get; set; }
 
         /// <summary>
         /// Gets or sets the boolean to use effective partition key routing in the cosmos db request.

@@ -83,6 +83,7 @@ internal class ClientContextCore : CosmosClientContext
                cosmosClientId: cosmosClient.Id,
                remoteCertificateValidationCallback: ClientContextCore.SslCustomValidationCallBack(clientOptions.GetServerCertificateCustomValidationCallback()),
                cosmosClientTelemetryOptions: clientOptions.CosmosClientTelemetryOptions,
+               availabilityStrategy: clientOptions.AvailabilityStrategy,
                chaosInterceptorFactory: clientOptions.ChaosInterceptorFactory);
 
             return ClientContextCore.Create(