Fix Local RPC race condition on app startup (#1719)

* Add test

* Fix broken test

* Finally have broken test

* Fix build warnings

* Fix implementation

* Add release notes

* Fix broken test

* Respond to PR feedback

Author: Connor McMahon

release_notes.md

@@ -3,4 +3,6 @@
 - Improved concurrency defaults for the App Service Consumption plan (https://github.com/Azure/azure-functions-durable-extension/pull/1706)
 
 ## Bug Fixes:
-- Properly used update management API URLs after a successful slot swap (#1716)
+- Properly used update management API URLs after a successful slot swap (#1716)
+- Fix race condition when multiple apps start with local RPC endpoints on the same VM in parallel. (#1719)
+

src/WebJobs.Extensions.DurableTask/HttpApiHandler.cs

@@ -1141,7 +1141,7 @@ internal bool TryGetRpcBaseUrl(out Uri rpcBaseUrl)
                 return true;
             }
 
-            // The app owner explicitly disabled the local RPC endpoint.
+            // The app owner explicitly disabled the local RPC endpoint
             rpcBaseUrl = null;
             return false;
         }
@@ -1151,14 +1151,14 @@ internal async Task StartLocalHttpServerAsync()
         {
             if (!this.localHttpListener.IsListening)
             {
+                await this.localHttpListener.StartAsync();
+
                 this.traceHelper.ExtensionInformationalEvent(
                     this.durableTaskOptions.HubName,
                     instanceId: string.Empty,
                     functionName: string.Empty,
-                    message: $"Opening local RPC endpoint: {this.localHttpListener.InternalRpcUri}",
+                    message: $"Opened local RPC endpoint: {this.localHttpListener.InternalRpcUri}",
                     writeToUserLogs: true);
-
-                await this.localHttpListener.StartAsync();
             }
         }
 

src/WebJobs.Extensions.DurableTask/LocalHttpListener.cs

@@ -22,10 +22,10 @@ internal class LocalHttpListener : IDisposable
     {
         private const int DefaultPort = 17071;
 
-        private readonly IWebHost localWebHost;
         private readonly Func<HttpRequestMessage, Task<HttpResponseMessage>> handler;
         private readonly EndToEndTraceHelper traceHelper;
         private readonly DurableTaskOptions durableTaskOptions;
+        private IWebHost localWebHost;
 
         public LocalHttpListener(
             EndToEndTraceHelper traceHelper,
@@ -36,23 +36,12 @@ public LocalHttpListener(
             this.handler = handler ?? throw new ArgumentNullException(nameof(handler));
             this.durableTaskOptions = durableTaskOptions ?? throw new ArgumentNullException(nameof(durableTaskOptions));
 
-#if !FUNCTIONS_V1
-            this.InternalRpcUri = new Uri($"http://127.0.0.1:{this.GetAvailablePort()}/durabletask/");
-            var listenUri = new Uri(this.InternalRpcUri.GetLeftPart(UriPartial.Authority));
-            this.localWebHost = new WebHostBuilder()
-                .UseKestrel()
-                .UseUrls(listenUri.OriginalString)
-                .Configure(a => a.Run(this.HandleRequestAsync))
-                .Build();
-#else
-            // Just use default port for internal Uri. No need to check for port availability since
-            // we won't be listening to this endpoint.
-            this.InternalRpcUri = new Uri($"http://127.0.0.1:{DefaultPort}/durabletask/");
+            // Set to a non null value
+            this.InternalRpcUri = new Uri($"http://uninitialized");
             this.localWebHost = new NoOpWebHost();
-#endif
         }
 
-        public Uri InternalRpcUri { get; }
+        public Uri InternalRpcUri { get; private set; }
 
         public bool IsListening { get; private set; }
 
@@ -66,8 +55,44 @@ public async Task StartAsync()
             }
 
 #if !FUNCTIONS_V1
-            await this.localWebHost.StartAsync();
-            this.IsListening = true;
+            const int maxAttempts = 10;
+            int numAttempts = 1;
+            do
+            {
+                int availablePort = this.GetAvailablePort();
+                try
+                {
+                    this.InternalRpcUri = new Uri($"http://127.0.0.1:{availablePort}/durabletask/");
+                    var listenUri = new Uri(this.InternalRpcUri.GetLeftPart(UriPartial.Authority));
+                    this.localWebHost = new WebHostBuilder()
+                        .UseKestrel()
+                        .UseUrls(listenUri.OriginalString)
+                        .Configure(a => a.Run(this.HandleRequestAsync))
+                        .Build();
+
+                    await this.localWebHost.StartAsync();
+                    this.IsListening = true;
+                    break;
+                }
+                catch (IOException)
+                {
+                    this.traceHelper.ExtensionWarningEvent(
+                        this.durableTaskOptions.HubName,
+                        functionName: string.Empty,
+                        instanceId: string.Empty,
+                        message: $"Failed to open local socket {availablePort}. This was attempt #{numAttempts} to open a local port.");
+                    numAttempts++;
+                    var random = new Random();
+                    var millisecondsToWait = (int)Math.Round(random.NextDouble() * 1000);
+                    await Task.Delay(millisecondsToWait);
+                }
+            }
+            while (numAttempts <= maxAttempts);
+
+            if (!this.IsListening)
+            {
+                throw new IOException($"Unable to find a port to open an RPC endpoint on after {maxAttempts} attempts");
+            }
 #else
             // no-op: this is dummy code to make build warnings go away
             await Task.Yield();
@@ -158,7 +183,6 @@ private static async Task SetResponseAsync(HttpContext context, HttpResponseMess
             }
         }
 
-#if FUNCTIONS_V1
         private class NoOpWebHost : IWebHost
         {
             public IFeatureCollection ServerFeatures => throw new NotImplementedException();
@@ -173,6 +197,5 @@ public void Start() { }
 
             public Task StopAsync(CancellationToken cancellationToken = default(CancellationToken)) => Task.CompletedTask;
         }
-#endif
     }
 }

test/Common/DurableTaskEndToEndTests.cs

@@ -4869,35 +4869,43 @@ public async Task CallActivity_Like_From_Azure_Portal()
             }
         }
 
-        [Fact]
+        [Theory]
         [Trait("Category", PlatformSpecificHelpers.TestCategory)]
-        public async Task MultipleHostsLocalRpcSameDevice()
+        [InlineData(true)]
+        [InlineData(false)]
+        public async Task MultipleHostsOnSameVM(bool enableLocalRpc)
         {
-            ITestHost host1 = TestHelpers.GetJobHost(
-                    this.loggerProvider,
-                    nameof(this.MultipleHostsLocalRpcSameDevice) + "1",
-                    false,
-                    localRpcEndpointEnabled: true);
-            await host1.StartAsync();
-            ITestHost host2 = TestHelpers.GetJobHost(
-                    this.loggerProvider,
-                    nameof(this.MultipleHostsLocalRpcSameDevice) + "2",
-                    false,
-                    localRpcEndpointEnabled: true);
-            try
-            {
-                await host2.StartAsync();
-            }
-            catch (Exception)
-            {
-                Assert.True(false, "Could not start up two hosts on the same device in parallel");
-            }
-            finally
+            // This test wants to be sure there are no race conditions while starting up multiple hosts in parallel,
+            // so attempt various times to increase the likelihood of hitting a race condition if one exists.
+            int numAttempts = 5;
+            for (int attempt = 0; attempt < numAttempts; attempt++)
             {
-                await host1.StopAsync();
-                host1.Dispose();
-                await host2.StopAsync();
-                host2.Dispose();
+                int numThreads = 10;
+                var hosts = new List<ITestHost>(numThreads);
+
+                try
+                {
+                    Parallel.For(0, numThreads, new ParallelOptions() { MaxDegreeOfParallelism = numThreads }, (i) =>
+                        hosts.Add(TestHelpers.GetJobHost(
+                                this.loggerProvider,
+                                nameof(this.MultipleHostsOnSameVM) + i,
+                                false,
+                                localRpcEndpointEnabled: enableLocalRpc)));
+
+                    await Task.WhenAll(hosts.Select(host => host.StartAsync()));
+                }
+                catch (Exception)
+                {
+                    Assert.True(false, "Could not start up two hosts on the same device in parallel");
+                }
+                finally
+                {
+                    foreach (var host in hosts)
+                    {
+                        await host.StopAsync();
+                        host.Dispose();
+                    }
+                }
             }
         }
 

test/FunctionsV2/OutOfProcTests.cs

@@ -348,7 +348,7 @@ public async Task TestLocalRcpEndpointRuntimeVersion(string runtimeVersion, bool
                 // Validate if we opened local RPC endpoint by looking at log statements.
                 var logger = this.loggerProvider.CreatedLoggers.Single(l => l.Category == TestHelpers.LogCategory);
                 var logMessages = logger.LogMessages.ToList();
-                bool enabledRpcEndpoint = logMessages.Any(msg => msg.Level == Microsoft.Extensions.Logging.LogLevel.Information && msg.FormattedMessage.StartsWith("Opening local RPC endpoint:"));
+                bool enabledRpcEndpoint = logMessages.Any(msg => msg.Level == Microsoft.Extensions.Logging.LogLevel.Information && msg.FormattedMessage.StartsWith("Opened local RPC endpoint:"));
 
                 Assert.Equal(enabledExpected, enabledRpcEndpoint);