Enabling MaxConcurrentCalls settings for the Orchestrator (#482)

Author: agr

src/NuGet.Services.Validation.Orchestrator/Configuration/OrchestrationRunnerConfiguration.cs

@@ -17,5 +17,10 @@ public class OrchestrationRunnerConfiguration
         /// if not done by the interval end, process would be terminated forcefully with log record about the incident.
         /// </summary>
         public TimeSpan ShutdownWaitInterval { get; set; }
+
+        /// <summary>
+        /// Max number of concurrent calls to be handled by the service bus library
+        /// </summary>
+        public int MaxConcurrentCalls { get; set; }
     }
 }

src/NuGet.Services.Validation.Orchestrator/OrchestrationRunner.cs

@@ -38,7 +38,7 @@ public async Task RunOrchestrationAsync()
         {
             _logger.LogInformation("Starting up the orchestration");
 
-            _subscriptionProcessor.Start();
+            _subscriptionProcessor.Start(_configuration.MaxConcurrentCalls);
             await Task.Delay(_configuration.ProcessRecycleInterval);
 
             _logger.LogInformation("Recycling the process...");

src/NuGet.Services.Validation.Orchestrator/settings.json

@@ -92,7 +92,8 @@
   },
   "RunnerConfiguration": {
     "ProcessRecycleInterval": "1:00:00:00",
-    "ShutdownWaitInterval": "00:01:00"
+    "ShutdownWaitInterval": "00:01:00",
+    "MaxConcurrentCalls": 10
   },
   "GalleryDb": {
     "ConnectionString": "Data Source=(localdb)\\mssqllocaldb; Initial Catalog=NuGetGallery; Integrated Security=True; MultipleActiveResultSets=True"

tests/NuGet.Services.Validation.Orchestrator.Tests/OrchestrationRunnerFacts.cs

@@ -13,21 +13,23 @@ namespace NuGet.Services.Validation.Orchestrator.Tests
 {
     public class OrchestrationRunnerFacts
     {
+        private const int DefaultMaxConcurrentCalls = 24;
+
         [Fact]
         public async Task StartsMessageProcessing()
         {
             var runner = CreateRunner();
             await runner.RunOrchestrationAsync();
 
-            SubscriptionProcessorMock.Verify(o => o.Start(), Times.Once());
+            SubscriptionProcessorMock.Verify(o => o.Start(DefaultMaxConcurrentCalls), Times.Once());
         }
 
         [Fact]
         public async Task ShutsDownMessageProcessing()
         {
             var startCalled = false;
             SubscriptionProcessorMock
-                .Setup(o => o.Start())
+                .Setup(o => o.Start(DefaultMaxConcurrentCalls))
                 .Callback(() => startCalled = true);
 
             SubscriptionProcessorMock
@@ -43,7 +45,7 @@ public async Task ShutsDownMessageProcessing()
         [Fact(Skip = "Flaky test. Won't run it as part of CI.")]
         public async Task WaitsOrchestratorToShutDown()
         {
-            SetupOptionsAccessorMock(TimeSpan.Zero, TimeSpan.FromSeconds(3));
+            SetupOptionsAccessorMock(TimeSpan.Zero, TimeSpan.FromSeconds(3), 2);
 
             int numberOfRequestsInProgress = 2;
             SubscriptionProcessorMock
@@ -58,15 +60,17 @@ public async Task WaitsOrchestratorToShutDown()
 
         private Mock<IOptionsSnapshot<OrchestrationRunnerConfiguration>> SetupOptionsAccessorMock(
             TimeSpan processRecycleInterval,
-            TimeSpan shutdownWaitInterval)
+            TimeSpan shutdownWaitInterval,
+            int maxConcurrentCalls)
         {
             OrchestrationRunnerConfigurationAccessorMock = new Mock<IOptionsSnapshot<OrchestrationRunnerConfiguration>>();
             OrchestrationRunnerConfigurationAccessorMock
                 .SetupGet(o => o.Value)
                 .Returns(new OrchestrationRunnerConfiguration
                 {
                     ProcessRecycleInterval = processRecycleInterval,
-                    ShutdownWaitInterval = shutdownWaitInterval
+                    ShutdownWaitInterval = shutdownWaitInterval,
+                    MaxConcurrentCalls = maxConcurrentCalls
                 });
             return OrchestrationRunnerConfigurationAccessorMock;
         }
@@ -81,7 +85,7 @@ public OrchestrationRunnerFacts()
         {
             SubscriptionProcessorMock = new Mock<ISubscriptionProcessor<PackageValidationMessageData>>();
             LoggerMock = new Mock<ILogger<OrchestrationRunner>>();
-            SetupOptionsAccessorMock(TimeSpan.Zero, TimeSpan.Zero);
+            SetupOptionsAccessorMock(TimeSpan.Zero, TimeSpan.Zero, DefaultMaxConcurrentCalls);
         }
 
         private Mock<ISubscriptionProcessor<PackageValidationMessageData>> SubscriptionProcessorMock { get; }