Delay invocations until worker ready on restart due to error (hotfix commit) (#5668)

* minimal changes to cover most cases

* Fix function invocation dispatcher change

* Fix comments

* No delays on placeholder mode or invalid. Added fix to http worker too

* remove dupe

* adding tests and addressing CR

* don't remove log

* log the exceptions

* update unit tests

* remove dupe test

* fix

* fix tests

(cherry picked from commit 5e452aa)

Author: mhoeger

src/WebJobs.Script/Description/Workers/WorkerFunctionInvoker.cs

@@ -90,7 +90,11 @@ protected override async Task<object> InvokeCore(object[] parameters, FunctionIn
 
         private async Task DelayUntilFunctionDispatcherInitializedOrShutdown()
         {
-            if (_functionDispatcher != null && _functionDispatcher.State == FunctionInvocationDispatcherState.Initializing)
+            // Don't delay if functionDispatcher is already initialized OR is skipping initialization for one of
+            // these reasons: started in placeholder, has no functions, functions do not match set language.
+            bool ready = _functionDispatcher.State == FunctionInvocationDispatcherState.Initialized || _functionDispatcher.State == FunctionInvocationDispatcherState.Default;
+
+            if (!ready)
             {
                 _logger.LogDebug($"functionDispatcher state: {_functionDispatcher.State}");
                 bool result = await Utility.DelayAsync((_functionDispatcher.ErrorEventsThreshold + 1) * WorkerConstants.ProcessStartTimeoutSeconds, WorkerConstants.WorkerReadyCheckPollingIntervalMilliseconds, () =>

src/WebJobs.Script/Workers/FunctionInvocationDispatcherState.cs

@@ -11,15 +11,30 @@ public enum FunctionInvocationDispatcherState
         Default,
 
         /// <summary>
-        /// The FunctionDispatcherState is starting.
+        /// The FunctionDispatcher is starting.
         /// </summary>
         Initializing,
 
         /// <summary>
-        /// The FunctionDispatcherState has been fully initialized and can accept direct function
-        /// invocations. All functions have been indexed. Listeners may not yet
-        /// be not yet running.
+        /// The FunctionDispatcher has been fully initialized and can accept function
+        /// invocations. All functions have been indexed. At least one worker process is initialized.
         /// </summary>
-        Initialized
+        Initialized,
+
+        /// <summary>
+        /// The FunctionDispatcher was previously "Initialized" but no longer has any initialized
+        /// worker processes to handle function invocations.
+        /// </summary>
+        WorkerProcessRestarting,
+
+        /// <summary>
+        /// The FunctionDispatcher is disposing
+        /// </summary>
+        Disposing,
+
+        /// <summary>
+        /// The FunctionDispatcher is disposed
+        /// </summary>
+        Disposed
     }
 }

src/WebJobs.Script/Workers/Http/HttpFunctionInvocationDispatcher.cs

@@ -103,7 +103,7 @@ public async void WorkerError(HttpWorkerErrorEvent workerError)
         {
             if (!_disposing)
             {
-                _logger.LogDebug("Handling WorkerErrorEvent for workerId:{workerId}", workerError.WorkerId);
+                _logger.LogDebug("Handling WorkerErrorEvent for workerId:{workerId}. Failed with: {exception}", workerError.WorkerId, workerError.Exception);
                 AddOrUpdateErrorBucket(workerError);
                 await DisposeAndRestartWorkerChannel(workerError.WorkerId);
             }
@@ -120,6 +120,9 @@ public async void WorkerRestart(HttpWorkerRestartEvent workerRestart)
 
         private async Task DisposeAndRestartWorkerChannel(string workerId)
         {
+            // Since we only have one HTTP worker process, as soon as we dispose it, InvokeAsync will fail. Set state to
+            // indicate we are not ready to receive new requests.
+            State = FunctionInvocationDispatcherState.WorkerProcessRestarting;
             _logger.LogDebug("Disposing channel for workerId: {channelId}", workerId);
             if (_httpWorkerChannel != null)
             {
@@ -166,13 +169,15 @@ protected virtual void Dispose(bool disposing)
                 _workerErrorSubscription.Dispose();
                 _workerRestartSubscription.Dispose();
                 (_httpWorkerChannel as IDisposable)?.Dispose();
+                State = FunctionInvocationDispatcherState.Disposed;
                 _disposed = true;
             }
         }
 
         public void Dispose()
         {
             _disposing = true;
+            State = FunctionInvocationDispatcherState.Disposing;
             Dispose(true);
         }
 

src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs

@@ -284,7 +284,7 @@ public async void WorkerError(WorkerErrorEvent workerError)
             {
                 if (string.Equals(_workerRuntime, workerError.Language))
                 {
-                    _logger.LogDebug("Handling WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}", workerError.Language, workerError.WorkerId);
+                    _logger.LogDebug("Handling WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}. Failed with: {exception}", workerError.Language, _workerRuntime, workerError.Exception);
                     AddOrUpdateErrorBucket(workerError);
                     await DisposeAndRestartWorkerChannel(workerError.Language, workerError.WorkerId);
                 }
@@ -332,8 +332,16 @@ private async Task DisposeAndRestartWorkerChannel(string runtime, string workerI
 
             if (ShouldRestartWorkerChannel(runtime, isWebHostChannel, isJobHostChannel))
             {
+                // Set state to "WorkerProcessRestarting" if there are no other workers to handle work
+                if ((await GetInitializedWorkerChannelsAsync()).Count() == 0)
+                {
+                    State = FunctionInvocationDispatcherState.WorkerProcessRestarting;
+                    _logger.LogDebug("No initialized worker channels for runtime '{runtime}'. Delaying future invocations", runtime);
+                }
+                // Restart worker channel
                 _logger.LogDebug("Restarting worker channel for runtime:{runtime}", runtime);
                 await RestartWorkerChannel(runtime, workerId);
+                // State is set back to "Initialized" when worker channel is up again
             }
             else
             {
@@ -386,13 +394,15 @@ protected virtual void Dispose(bool disposing)
                 _processStartCancellationToken.Cancel();
                 _processStartCancellationToken.Dispose();
                 _jobHostLanguageWorkerChannelManager.DisposeAndRemoveChannels();
+                State = FunctionInvocationDispatcherState.Disposed;
                 _disposed = true;
             }
         }
 
         public void Dispose()
         {
             _disposing = true;
+            State = FunctionInvocationDispatcherState.Disposing;
             Dispose(true);
         }
     }

test/WebJobs.Script.Tests/Description/Worker/WorkerFunctionInvokerTests.cs

@@ -56,6 +56,29 @@ public async Task InvokeTimeout_CallsShutdown()
             _applicationLifetime.Verify(a => a.StopApplication(), Times.Once);
         }
 
+        [Theory]
+        [InlineData(FunctionInvocationDispatcherState.Default, false)]
+        [InlineData(FunctionInvocationDispatcherState.Initializing, true)]
+        [InlineData(FunctionInvocationDispatcherState.Initialized, false)]
+        [InlineData(FunctionInvocationDispatcherState.WorkerProcessRestarting, true)]
+        [InlineData(FunctionInvocationDispatcherState.Disposing, true)]
+        [InlineData(FunctionInvocationDispatcherState.Disposed, true)]
+        public async Task FunctionDispatcher_DelaysInvoke_WhenNotReady(FunctionInvocationDispatcherState state, bool delaysExecution)
+        {
+            _mockFunctionInvocationDispatcher.Setup(a => a.State).Returns(state);
+            var timeoutTask = Task.Delay(TimeSpan.FromSeconds(5));
+            var invokeCoreTask = _testFunctionInvoker.InvokeCore(new object[] { }, null);
+            var result = await Task.WhenAny(invokeCoreTask, timeoutTask);
+            if (delaysExecution)
+            {
+                Assert.Equal(timeoutTask, result);
+            }
+            else
+            {
+                Assert.Equal(invokeCoreTask, result);
+            }
+        }
+
         [Fact]
         public async Task InvokeInitialized_DoesNotCallShutdown()
         {

test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs

@@ -200,8 +200,31 @@ public async Task FunctionDispatcherState_Transitions_From_Starting_To_Initializ
                 func1
             };
             await functionDispatcher.InitializeAsync(functions);
-            Assert.True(functionDispatcher.State == FunctionInvocationDispatcherState.Initializing || functionDispatcher.State == FunctionInvocationDispatcherState.Initialized);
+            Assert.Equal(FunctionInvocationDispatcherState.Initializing, functionDispatcher.State);
             await WaitForFunctionDispactherStateInitialized(functionDispatcher);
+            Assert.Equal(FunctionInvocationDispatcherState.Initialized, functionDispatcher.State);
+        }
+
+        [Fact]
+        public async Task FunctionDispatcherState_Transitions_From_Default_To_Initialized_To_Disposing()
+        {
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher();
+            Assert.Equal(FunctionInvocationDispatcherState.Default, functionDispatcher.State);
+            FunctionMetadata func1 = new FunctionMetadata()
+            {
+                Name = "func1",
+                Language = "node"
+            };
+            var functions = new List<FunctionMetadata>()
+            {
+                func1
+            };
+            await functionDispatcher.InitializeAsync(functions);
+            Assert.Equal(FunctionInvocationDispatcherState.Initializing, functionDispatcher.State);
+            await WaitForFunctionDispactherStateInitialized(functionDispatcher);
+            Assert.Equal(FunctionInvocationDispatcherState.Initialized, functionDispatcher.State);
+            functionDispatcher.Dispose();
+            Assert.True(functionDispatcher == null || functionDispatcher.State == FunctionInvocationDispatcherState.Disposing || functionDispatcher.State == FunctionInvocationDispatcherState.Disposed);
         }
 
         [Fact]
@@ -225,6 +248,43 @@ public async Task FunctionDispatcher_Restart_ErroredChannels_Succeeds()
             Assert.Equal(expectedProcessCount, finalChannelCount);
         }
 
+        [Fact]
+        public async Task FunctionDispatcher_Restart_ErroredChannels_And_Changes_State()
+        {
+            int expectedProcessCount = 1;
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(expectedProcessCount.ToString());
+            Assert.Equal(FunctionInvocationDispatcherState.Default, functionDispatcher.State);
+            // Add worker
+            await functionDispatcher.InitializeAsync(GetTestFunctionsList(RpcWorkerConstants.NodeLanguageWorkerName));
+            await WaitForJobhostWorkerChannelsToStartup(functionDispatcher, expectedProcessCount);
+            TestRpcWorkerChannel testWorkerChannel = (TestRpcWorkerChannel)functionDispatcher.JobHostLanguageWorkerChannelManager.GetChannels().FirstOrDefault();
+            // Restart this channel
+            testWorkerChannel.RaiseWorkerRestart();
+            await TestHelpers.Await(() =>
+            {
+                return functionDispatcher.State == FunctionInvocationDispatcherState.WorkerProcessRestarting;
+            }, 3000);
+            await WaitForJobhostWorkerChannelsToStartup(functionDispatcher, expectedProcessCount);
+            Assert.Equal(FunctionInvocationDispatcherState.Initialized, functionDispatcher.State);
+        }
+
+        [Fact]
+        public async Task FunctionDispatcher_Restart_ErroredChannels_And_DoesNot_Change_State()
+        {
+            int expectedProcessCount = 2;
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(expectedProcessCount.ToString());
+            Assert.Equal(FunctionInvocationDispatcherState.Default, functionDispatcher.State);
+            // Add worker
+            await functionDispatcher.InitializeAsync(GetTestFunctionsList(RpcWorkerConstants.NodeLanguageWorkerName));
+            await WaitForJobhostWorkerChannelsToStartup(functionDispatcher, expectedProcessCount);
+            TestRpcWorkerChannel testWorkerChannel = (TestRpcWorkerChannel)functionDispatcher.JobHostLanguageWorkerChannelManager.GetChannels().FirstOrDefault();
+            // Restart one channel
+            testWorkerChannel.RaiseWorkerRestart();
+            Assert.Equal(FunctionInvocationDispatcherState.Initialized, functionDispatcher.State);
+            await WaitForJobhostWorkerChannelsToStartup(functionDispatcher, expectedProcessCount);
+            Assert.Equal(FunctionInvocationDispatcherState.Initialized, functionDispatcher.State);
+        }
+
         [Fact]
         public async Task FunctionDispatcher_Restart_ErroredChannels_ExceedsLimit()
         {

test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannel.cs

@@ -71,7 +71,7 @@ public void SendInvocationRequest(ScriptInvocationContext context)
         public async Task StartWorkerProcessAsync()
         {
             // To verify FunctionDispatcher transistions
-            await Task.Delay(TimeSpan.FromMilliseconds(100));
+            await Task.Delay(TimeSpan.FromMilliseconds(500));
 
             if (_throwOnProcessStartUp)
             {