Added global request throttling

src/DurableTask.AzureStorage/OrchestrationSessionManager.cs

@@ -434,7 +434,7 @@ internal void AddMessageToPendingOrchestration(
 
                         // Before the batch of messages can be processed, we need to download the latest execution state.
                         // This is done beforehand in the background as a performance optimization.
-                        this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken);
+                        Task.Run(() => this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken));
                     }
 
                     // New messages are added; duplicate messages are replaced
@@ -454,55 +454,51 @@ internal void AddMessageToPendingOrchestration(
         }
 
         // This method runs on a background task thread
-        void ScheduleOrchestrationStatePrefetch(
+        async Task ScheduleOrchestrationStatePrefetch(
             LinkedListNode<PendingMessageBatch> node,
             Guid traceActivityId,
             CancellationToken cancellationToken)
         {
             PendingMessageBatch batch = node.Value;
 
-            // Do the fetch in a background thread
-            this.fetchRuntimeStateQueue.EnqueueAndDispatch(async delegate
+            AnalyticsEventSource.SetLogicalTraceActivityId(traceActivityId);
+
+            try
             {
-                AnalyticsEventSource.SetLogicalTraceActivityId(traceActivityId);
-
-                try
+                if (batch.OrchestrationState == null)
                 {
-                    if (batch.OrchestrationState == null)
-                    {
-                        OrchestrationHistory history = await this.trackingStore.GetHistoryEventsAsync(
-                           batch.OrchestrationInstanceId,
-                           batch.OrchestrationExecutionId,
-                           cancellationToken);
-
-                        batch.OrchestrationState = new OrchestrationRuntimeState(history.Events);
-                        batch.ETag = history.ETag;
-                        batch.LastCheckpointTime = history.LastCheckpointTime;
-                    }
-
-                    this.readyForProcessingQueue.Enqueue(node);
+                    OrchestrationHistory history = await this.trackingStore.GetHistoryEventsAsync(
+                       batch.OrchestrationInstanceId,
+                       batch.OrchestrationExecutionId,
+                       cancellationToken);
+
+                    batch.OrchestrationState = new OrchestrationRuntimeState(history.Events);
+                    batch.ETag = history.ETag;
+                    batch.LastCheckpointTime = history.LastCheckpointTime;
                 }
-                catch (OperationCanceledException)
-                {
-                    // shutting down
-                }
-                catch (Exception e)
-                {
-                    this.settings.Logger.OrchestrationProcessingFailure(
-                        this.storageAccountName,
-                        this.settings.TaskHubName,
-                        batch.OrchestrationInstanceId,
-                        batch.OrchestrationExecutionId,
-                        e.ToString());
 
-                    // Sleep briefly to avoid a tight failure loop.
-                    await Task.Delay(TimeSpan.FromSeconds(5));
+                this.readyForProcessingQueue.Enqueue(node);
+            }
+            catch (OperationCanceledException)
+            {
+                // shutting down
+            }
+            catch (Exception e)
+            {
+                this.settings.Logger.OrchestrationProcessingFailure(
+                    this.storageAccountName,
+                    this.settings.TaskHubName,
+                    batch.OrchestrationInstanceId,
+                    batch.OrchestrationExecutionId,
+                    e.ToString());
 
-                    // This is a background operation so failure is not an option. All exceptions must be handled.
-                    // To avoid starvation, we need to re-enqueue this async operation instead of retrying in a loop.
-                    this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken);
-                }
-            });
+                // Sleep briefly to avoid a tight failure loop.
+                await Task.Delay(TimeSpan.FromSeconds(5));
+
+                // This is a background operation so failure is not an option. All exceptions must be handled.
+                // To avoid starvation, we need to re-enqueue this async operation instead of retrying in a loop.
+                await Task.Run(() => this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken));
+            }
         }
 
         public async Task<OrchestrationSession?> GetNextSessionAsync(CancellationToken cancellationToken)

src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs

@@ -29,6 +29,7 @@ class AzureStorageClient
         readonly CloudBlobClient blobClient;
         readonly CloudQueueClient queueClient;
         readonly CloudTableClient tableClient;
+        readonly SemaphoreSlim requestThrottleSemaphore;
 
         public AzureStorageClient(AzureStorageOrchestrationServiceSettings settings) : 
             this(settings.StorageAccountDetails == null ?
@@ -61,6 +62,8 @@ public AzureStorageClient(CloudStorageAccount account, AzureStorageOrchestration
             }
 
             this.tableClient.BufferManager = SimpleBufferManager.Shared;
+
+            this.requestThrottleSemaphore = new SemaphoreSlim(this.Settings.MaxStorageOperationConcurrency);
         }
 
         public AzureStorageOrchestrationServiceSettings Settings { get; }
@@ -101,6 +104,8 @@ public Table GetTableReference(string tableName)
 
         public async Task<T> MakeStorageRequest<T>(Func<OperationContext, CancellationToken, Task<T>> storageRequest, string operationName, string? clientRequestId = null)
         {
+            await requestThrottleSemaphore.WaitAsync();
+
             try
             {
                 return await TimeoutHandler.ExecuteWithTimeout<T>(operationName, this.StorageAccountName, this.Settings, storageRequest, this.Stats, clientRequestId);
@@ -109,6 +114,10 @@ public async Task<T> MakeStorageRequest<T>(Func<OperationContext, CancellationTo
             {
                 throw new DurableTaskStorageException(ex);
             }
+            finally
+            {
+                requestThrottleSemaphore.Release();
+            }
         }
 
         public async Task MakeStorageRequest(Func<OperationContext, CancellationToken, Task> storageRequest, string operationName, string? clientRequestId = null)