Fail executions when the worker goes down v2 (#5928)

mhoeger · web-flow · commit 8bdb09f4a698 · 2020-07-15T11:53:06.000-07:00
* fail executions when the worker goes down

* matching test logic to behavior

* address ahmed pr comments
diff --git a/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs b/src/WebJobs.Script/Workers/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs
@@ -201,7 +201,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                             catch (Exception ex)
                             {
                                 _logger.LogWarning(ex, "Removing errored webhost language worker channel for runtime: {workerRuntime} workerId:{workerId}", _workerRuntime, workerId);
-                                await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(_workerRuntime, workerId);
+                                await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(_workerRuntime, workerId, ex);
                                 InitializeWebhostLanguageWorkerChannel();
                             }
                         }
@@ -289,7 +289,7 @@ public async void WorkerError(WorkerErrorEvent workerError)
                 {
                     _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);
+                    await DisposeAndRestartWorkerChannel(workerError.Language, workerError.WorkerId, workerError.Exception);
                 }
                 else
                 {
@@ -308,15 +308,15 @@ public async void WorkerRestart(WorkerRestartEvent workerRestart)
             }
         }
 
-        private async Task DisposeAndRestartWorkerChannel(string runtime, string workerId)
+        private async Task DisposeAndRestartWorkerChannel(string runtime, string workerId, Exception workerException = null)
         {
             _logger.LogDebug("Attempting to dispose webhost or jobhost channel for workerId: {channelId}, runtime:{language}", workerId, runtime);
 
-            bool isWebHostChannelDisposed = await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(runtime, workerId);
+            bool isWebHostChannelDisposed = await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(runtime, workerId, workerException);
             bool isJobHostChannelDisposed = false;
             if (!isWebHostChannelDisposed)
             {
-                isJobHostChannelDisposed = await _jobHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(workerId);
+                isJobHostChannelDisposed = await _jobHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(workerId, workerException);
             }
 
             if (!isWebHostChannelDisposed && !isJobHostChannelDisposed)
diff --git a/src/WebJobs.Script/Workers/Rpc/IJobHostRpcWorkerChannelManager.cs b/src/WebJobs.Script/Workers/Rpc/IJobHostRpcWorkerChannelManager.cs
@@ -1,6 +1,7 @@
 ﻿// Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the Apache License, Version 2.0. See License.txt in the project root for license information.
 
+using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 
@@ -10,7 +11,7 @@ internal interface IJobHostRpcWorkerChannelManager
     {
         void AddChannel(IRpcWorkerChannel channel);
 
-        Task<bool> ShutdownChannelIfExistsAsync(string channelId);
+        Task<bool> ShutdownChannelIfExistsAsync(string channelId, Exception workerException);
 
         void ShutdownChannels();
 
diff --git a/src/WebJobs.Script/Workers/Rpc/IRpcWorkerChannel.cs b/src/WebJobs.Script/Workers/Rpc/IRpcWorkerChannel.cs
@@ -29,5 +29,7 @@ public interface IRpcWorkerChannel
         Task DrainInvocationsAsync();
 
         bool IsExecutingInvocation(string invocationId);
+
+        bool TryFailExecutions(Exception workerException);
     }
 }
diff --git a/src/WebJobs.Script/Workers/Rpc/IWebHostRpcWorkerChannelManager.cs b/src/WebJobs.Script/Workers/Rpc/IWebHostRpcWorkerChannelManager.cs
@@ -1,6 +1,7 @@
 ﻿// Copyright (c) .NET Foundation. All rights reserved.
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
+using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 
@@ -14,7 +15,7 @@ public interface IWebHostRpcWorkerChannelManager
 
         Task SpecializeAsync();
 
-        Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId);
+        Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId, Exception workerException);
 
         Task ShutdownChannelsAsync();
     }
diff --git a/src/WebJobs.Script/Workers/Rpc/JobHostRpcWorkerChannelManager.cs b/src/WebJobs.Script/Workers/Rpc/JobHostRpcWorkerChannelManager.cs
@@ -24,11 +24,12 @@ public void AddChannel(IRpcWorkerChannel channel)
             _channels.TryAdd(channel.Id, channel);
         }
 
-        public Task<bool> ShutdownChannelIfExistsAsync(string channelId)
+        public Task<bool> ShutdownChannelIfExistsAsync(string channelId, Exception workerException)
         {
             if (_channels.TryRemove(channelId, out IRpcWorkerChannel removedChannel))
             {
                 _logger.LogDebug("Disposing JobHost language worker channel with id:{workerId}", removedChannel.Id);
+                removedChannel.TryFailExecutions(workerException);
                 (removedChannel as IDisposable)?.Dispose();
                 return Task.FromResult(true);
             }
diff --git a/src/WebJobs.Script/Workers/Rpc/RpcWorkerChannel.cs b/src/WebJobs.Script/Workers/Rpc/RpcWorkerChannel.cs
@@ -346,7 +346,8 @@ internal void SendInvocationRequest(ScriptInvocationContext context)
         internal void InvokeResponse(InvocationResponse invokeResponse)
         {
             _workerChannelLogger.LogDebug("InvocationResponse received for invocation id: {Id}", invokeResponse.InvocationId);
-            if (_executingInvocations.TryRemove(invokeResponse.InvocationId, out ScriptInvocationContext context)
+            bool invocationExists = _executingInvocations.TryRemove(invokeResponse.InvocationId, out ScriptInvocationContext context);
+            if (invocationExists
                 && invokeResponse.Result.IsSuccess(context.ResultSource))
             {
                 try
@@ -366,6 +367,11 @@ internal void InvokeResponse(InvocationResponse invokeResponse)
                     context.ResultSource.TrySetException(responseEx);
                 }
             }
+
+            if (!invocationExists)
+            {
+                _workerChannelLogger.LogDebug("Invocation was not found as an executing invocation (id={id})", invokeResponse.InvocationId);
+            }
         }
 
         internal void Log(RpcEvent msg)
@@ -507,5 +513,22 @@ public bool IsExecutingInvocation(string invocationId)
         {
             return _executingInvocations.ContainsKey(invocationId);
         }
+
+        public bool TryFailExecutions(Exception workerException)
+        {
+            if (workerException == null)
+            {
+                return false;
+            }
+
+            foreach (ScriptInvocationContext currContext in _executingInvocations?.Values)
+            {
+                string invocationId = currContext?.ExecutionContext?.InvocationId.ToString();
+                _workerChannelLogger.LogDebug("Worker '{workerId}' encountered a fatal error. Failing invocation id: {Id}", _workerId, invocationId);
+                _executingInvocations.TryRemove(invocationId, out ScriptInvocationContext _);
+                currContext?.ResultSource?.TrySetException(workerException);
+            }
+            return true;
+        }
     }
 }
diff --git a/src/WebJobs.Script/Workers/Rpc/WebHostRpcWorkerChannelManager.cs b/src/WebJobs.Script/Workers/Rpc/WebHostRpcWorkerChannelManager.cs
@@ -146,7 +146,7 @@ private bool UsePlaceholderChannel(string workerRuntime)
             return false;
         }
 
-        public Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId)
+        public Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId, Exception workerException = null)
         {
             if (string.IsNullOrEmpty(language))
             {
@@ -168,6 +168,8 @@ public Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId)
                             if (workerChannel != null)
                             {
                                 _logger.LogDebug("Disposing WebHost channel for workerId: {channelId}, for runtime:{language}", workerId, language);
+                                // Set exception if exists
+                                workerChannel.TryFailExecutions(workerException);
                                 (channelTask.Result as IDisposable)?.Dispose();
                             }
                         }
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/RpcFunctionInvocationDispatcherTests.cs
@@ -506,7 +506,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(string
                 WorkerConfigs = TestHelpers.GetTestWorkerConfigs()
             };
             IRpcWorkerChannelFactory testLanguageWorkerChannelFactory = new TestRpcWorkerChannelFactory(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, throwOnProcessStartUp);
-            IWebHostRpcWorkerChannelManager testWebHostLanguageWorkerChannelManager = new TesRpcWorkerChannelManager(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, testLanguageWorkerChannelFactory);
+            IWebHostRpcWorkerChannelManager testWebHostLanguageWorkerChannelManager = new TestRpcWorkerChannelManager(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, testLanguageWorkerChannelFactory);
             IJobHostRpcWorkerChannelManager jobHostLanguageWorkerChannelManager = new JobHostRpcWorkerChannelManager(loggerFactory);
             if (addWebhostChannel)
             {
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/RpcWorkerChannelTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/RpcWorkerChannelTests.cs
@@ -224,6 +224,20 @@ public async Task Drain_Verify()
             Assert.Equal(result.Status, TaskStatus.RanToCompletion);
         }
 
+        [Fact]
+        public void InFlight_Functions_FailedWithException()
+        {
+            var resultSource = new TaskCompletionSource<ScriptInvocationResult>();
+            ScriptInvocationContext scriptInvocationContext = GetTestScriptInvocationContext(Guid.NewGuid(), resultSource);
+            _workerChannel.SendInvocationRequest(scriptInvocationContext);
+            Assert.True(_workerChannel.IsExecutingInvocation(scriptInvocationContext.ExecutionContext.InvocationId.ToString()));
+            Exception workerException = new Exception("worker failed");
+            _workerChannel.TryFailExecutions(workerException);
+            Assert.False(_workerChannel.IsExecutingInvocation(scriptInvocationContext.ExecutionContext.InvocationId.ToString()));
+            Assert.Equal(TaskStatus.Faulted, resultSource.Task.Status);
+            Assert.Equal(workerException, resultSource.Task.Exception.InnerException);
+        }
+
         [Fact]
         public void SendLoadRequests_PublishesOutboundEvents()
         {
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannel.cs b/test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannel.cs
@@ -130,5 +130,12 @@ public bool IsExecutingInvocation(string invocationId)
         {
             return _executingInvocations.Contains(invocationId);
         }
+
+        public bool TryFailExecutions(Exception exception)
+        {
+            // Executions are no longer executing
+            _executingInvocations = new HashSet<string>();
+            return true;
+        }
     }
 }
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannelManager.cs b/test/WebJobs.Script.Tests/Workers/Rpc/TestRpcWorkerChannelManager.cs
@@ -15,15 +15,15 @@
 
 namespace Microsoft.Azure.WebJobs.Script.Tests.Workers.Rpc
 {
-    public class TesRpcWorkerChannelManager : IWebHostRpcWorkerChannelManager
+    public class TestRpcWorkerChannelManager : IWebHostRpcWorkerChannelManager
     {
         private IScriptEventManager _eventManager;
         private ILogger _testLogger;
         private ConcurrentDictionary<string, Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>>> _workerChannels = new ConcurrentDictionary<string, Dictionary<string, TaskCompletionSource<IRpcWorkerChannel>>>();
         private string _scriptRootPath;
         private IRpcWorkerChannelFactory _testLanguageWorkerChannelFactory;
 
-        public TesRpcWorkerChannelManager(IScriptEventManager eventManager, ILogger testLogger, string scriptRootPath, IRpcWorkerChannelFactory testLanguageWorkerChannelFactory)
+        public TestRpcWorkerChannelManager(IScriptEventManager eventManager, ILogger testLogger, string scriptRootPath, IRpcWorkerChannelFactory testLanguageWorkerChannelFactory)
         {
             _eventManager = eventManager;
             _testLogger = testLogger;
@@ -82,7 +82,7 @@ public void ShutdownProcessRegistry()
         {
         }
 
-        public async Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId)
+        public async Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId, Exception workerException)
         {
             if (string.IsNullOrEmpty(language))
             {
@@ -97,6 +97,7 @@ public async Task<bool> ShutdownChannelIfExistsAsync(string language, string wor
                         IRpcWorkerChannel channel = await value?.Task;
                         if (channel != null)
                         {
+                            channel.TryFailExecutions(workerException);
                             (channel as IDisposable)?.Dispose();
                             rpcWorkerChannels.Remove(workerId);
                             return true;
diff --git a/test/WebJobs.Script.Tests/Workers/Rpc/WebHostRpcWorkerChannelManagerTests.cs b/test/WebJobs.Script.Tests/Workers/Rpc/WebHostRpcWorkerChannelManagerTests.cs
@@ -6,6 +6,7 @@
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Script.Abstractions;
+using Microsoft.Azure.WebJobs.Script.Description;
 using Microsoft.Azure.WebJobs.Script.Diagnostics;
 using Microsoft.Azure.WebJobs.Script.Eventing;
 using Microsoft.Azure.WebJobs.Script.Workers;
@@ -289,6 +290,36 @@ public async Task ShutdownChannelsIfExist_Succeeds()
             Assert.Null(initializedChannel);
         }
 
+        [Fact]
+        public async Task ShutdownChannelsIfExistsAsync_StopsWorkerInvocations()
+        {
+            IRpcWorkerChannel javaWorkerChannel = CreateTestChannel(RpcWorkerConstants.JavaLanguageWorkerName);
+            Guid invocationId = Guid.NewGuid();
+            ScriptInvocationContext scriptInvocationContext = new ScriptInvocationContext()
+            {
+                ExecutionContext = new ExecutionContext()
+                {
+                    InvocationId = invocationId
+                }
+            };
+            (javaWorkerChannel as TestRpcWorkerChannel).SendInvocationRequest(scriptInvocationContext);
+            Assert.True(javaWorkerChannel.IsExecutingInvocation(invocationId.ToString()));
+            Exception workerException = new Exception("Worker exception");
+            // Channel is removed immediately but is not failed immediately
+            await _rpcWorkerChannelManager.ShutdownChannelIfExistsAsync(RpcWorkerConstants.JavaLanguageWorkerName, javaWorkerChannel.Id, workerException);
+
+            Assert.Null(_rpcWorkerChannelManager.GetChannels(RpcWorkerConstants.JavaLanguageWorkerName));
+
+            var initializedChannel = await _rpcWorkerChannelManager.GetChannelAsync(RpcWorkerConstants.JavaLanguageWorkerName);
+            Assert.Null(initializedChannel);
+            // Execution will be terminated in the background - giving it 10 seconds
+            await TestHelpers.Await(() =>
+            {
+                return !javaWorkerChannel.IsExecutingInvocation(invocationId.ToString());
+            }, 10000);
+            Assert.False(javaWorkerChannel.IsExecutingInvocation(invocationId.ToString()));
+        }
+
         [Fact]
         public async Task InitializeLanguageWorkerChannel_ThrowsOnProcessStartup()
         {

Original file line number	Diff line number	Diff line change
`@@ -201,7 +201,7 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance`
`201`	`201`	`catch (Exception ex)`
`202`	`202`	`{`
`203`	`203`	`_logger.LogWarning(ex, "Removing errored webhost language worker channel for runtime: {workerRuntime} workerId:{workerId}", _workerRuntime, workerId);`
`204`		`- await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(_workerRuntime, workerId);`
	`204`	`+ await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(_workerRuntime, workerId, ex);`
`205`	`205`	`InitializeWebhostLanguageWorkerChannel();`
`206`	`206`	`}`
`207`	`207`	`}`
`@@ -289,7 +289,7 @@ public async void WorkerError(WorkerErrorEvent workerError)`
`289`	`289`	`{`
`290`	`290`	`_logger.LogDebug("Handling WorkerErrorEvent for runtime:{runtime}, workerId:{workerId}. Failed with: {exception}", workerError.Language, _workerRuntime, workerError.Exception);`
`291`	`291`	`AddOrUpdateErrorBucket(workerError);`
`292`		`- await DisposeAndRestartWorkerChannel(workerError.Language, workerError.WorkerId);`
	`292`	`+ await DisposeAndRestartWorkerChannel(workerError.Language, workerError.WorkerId, workerError.Exception);`
`293`	`293`	`}`
`294`	`294`	`else`
`295`	`295`	`{`
`@@ -308,15 +308,15 @@ public async void WorkerRestart(WorkerRestartEvent workerRestart)`
`308`	`308`	`}`
`309`	`309`	`}`
`310`	`310`
`311`		`- private async Task DisposeAndRestartWorkerChannel(string runtime, string workerId)`
	`311`	`+ private async Task DisposeAndRestartWorkerChannel(string runtime, string workerId, Exception workerException = null)`
`312`	`312`	`{`
`313`	`313`	`_logger.LogDebug("Attempting to dispose webhost or jobhost channel for workerId: {channelId}, runtime:{language}", workerId, runtime);`
`314`	`314`
`315`		`- bool isWebHostChannelDisposed = await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(runtime, workerId);`
	`315`	`+ bool isWebHostChannelDisposed = await _webHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(runtime, workerId, workerException);`
`316`	`316`	`bool isJobHostChannelDisposed = false;`
`317`	`317`	`if (!isWebHostChannelDisposed)`
`318`	`318`	`{`
`319`		`- isJobHostChannelDisposed = await _jobHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(workerId);`
	`319`	`+ isJobHostChannelDisposed = await _jobHostLanguageWorkerChannelManager.ShutdownChannelIfExistsAsync(workerId, workerException);`
`320`	`320`	`}`
`321`	`321`
`322`	`322`	`if (!isWebHostChannelDisposed && !isJobHostChannelDisposed)`
Original file line number	Diff line number	Diff line change
`@@ -29,5 +29,7 @@ public interface IRpcWorkerChannel`
`29`	`29`	`Task DrainInvocationsAsync();`
`30`	`30`
`31`	`31`	`bool IsExecutingInvocation(string invocationId);`
	`32`	`+`
	`33`	`+ bool TryFailExecutions(Exception workerException);`
`32`	`34`	`}`
`33`	`35`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,11 +24,12 @@ public void AddChannel(IRpcWorkerChannel channel)`
`24`	`24`	`_channels.TryAdd(channel.Id, channel);`
`25`	`25`	`}`
`26`	`26`
`27`		`- public Task<bool> ShutdownChannelIfExistsAsync(string channelId)`
	`27`	`+ public Task<bool> ShutdownChannelIfExistsAsync(string channelId, Exception workerException)`
`28`	`28`	`{`
`29`	`29`	`if (_channels.TryRemove(channelId, out IRpcWorkerChannel removedChannel))`
`30`	`30`	`{`
`31`	`31`	`_logger.LogDebug("Disposing JobHost language worker channel with id:{workerId}", removedChannel.Id);`
	`32`	`+ removedChannel.TryFailExecutions(workerException);`
`32`	`33`	`(removedChannel as IDisposable)?.Dispose();`
`33`	`34`	`return Task.FromResult(true);`
`34`	`35`	`}`
Original file line number	Diff line number	Diff line change
`@@ -346,7 +346,8 @@ internal void SendInvocationRequest(ScriptInvocationContext context)`
`346`	`346`	`internal void InvokeResponse(InvocationResponse invokeResponse)`
`347`	`347`	`{`
`348`	`348`	`_workerChannelLogger.LogDebug("InvocationResponse received for invocation id: {Id}", invokeResponse.InvocationId);`
`349`		`- if (_executingInvocations.TryRemove(invokeResponse.InvocationId, out ScriptInvocationContext context)`
	`349`	`+ bool invocationExists = _executingInvocations.TryRemove(invokeResponse.InvocationId, out ScriptInvocationContext context);`
	`350`	`+ if (invocationExists`
`350`	`351`	`&& invokeResponse.Result.IsSuccess(context.ResultSource))`
`351`	`352`	`{`
`352`	`353`	`try`
`@@ -366,6 +367,11 @@ internal void InvokeResponse(InvocationResponse invokeResponse)`
`366`	`367`	`context.ResultSource.TrySetException(responseEx);`
`367`	`368`	`}`
`368`	`369`	`}`
	`370`	`+`
	`371`	`+ if (!invocationExists)`
	`372`	`+ {`
	`373`	`+ _workerChannelLogger.LogDebug("Invocation was not found as an executing invocation (id={id})", invokeResponse.InvocationId);`
	`374`	`+ }`
`369`	`375`	`}`
`370`	`376`
`371`	`377`	`internal void Log(RpcEvent msg)`
`@@ -507,5 +513,22 @@ public bool IsExecutingInvocation(string invocationId)`
`507`	`513`	`{`
`508`	`514`	`return _executingInvocations.ContainsKey(invocationId);`
`509`	`515`	`}`
	`516`	`+`
	`517`	`+ public bool TryFailExecutions(Exception workerException)`
	`518`	`+ {`
	`519`	`+ if (workerException == null)`
	`520`	`+ {`
	`521`	`+ return false;`
	`522`	`+ }`
	`523`	`+`
	`524`	`+ foreach (ScriptInvocationContext currContext in _executingInvocations?.Values)`
	`525`	`+ {`
	`526`	`+ string invocationId = currContext?.ExecutionContext?.InvocationId.ToString();`
	`527`	`+ _workerChannelLogger.LogDebug("Worker '{workerId}' encountered a fatal error. Failing invocation id: {Id}", _workerId, invocationId);`
	`528`	`+ _executingInvocations.TryRemove(invocationId, out ScriptInvocationContext _);`
	`529`	`+ currContext?.ResultSource?.TrySetException(workerException);`
	`530`	`+ }`
	`531`	`+ return true;`
	`532`	`+ }`
`510`	`533`	`}`
`511`	`534`	`}`
Original file line number	Diff line number	Diff line change
`@@ -146,7 +146,7 @@ private bool UsePlaceholderChannel(string workerRuntime)`
`146`	`146`	`return false;`
`147`	`147`	`}`
`148`	`148`
`149`		`- public Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId)`
	`149`	`+ public Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId, Exception workerException = null)`
`150`	`150`	`{`
`151`	`151`	`if (string.IsNullOrEmpty(language))`
`152`	`152`	`{`
`@@ -168,6 +168,8 @@ public Task<bool> ShutdownChannelIfExistsAsync(string language, string workerId)`
`168`	`168`	`if (workerChannel != null)`
`169`	`169`	`{`
`170`	`170`	`_logger.LogDebug("Disposing WebHost channel for workerId: {channelId}, for runtime:{language}", workerId, language);`
	`171`	`+ // Set exception if exists`
	`172`	`+ workerChannel.TryFailExecutions(workerException);`
`171`	`173`	`(channelTask.Result as IDisposable)?.Dispose();`
`172`	`174`	`}`
`173`	`175`	`}`
Original file line number	Diff line number	Diff line change
`@@ -506,7 +506,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(string`
`506`	`506`	`WorkerConfigs = TestHelpers.GetTestWorkerConfigs()`
`507`	`507`	`};`
`508`	`508`	`IRpcWorkerChannelFactory testLanguageWorkerChannelFactory = new TestRpcWorkerChannelFactory(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, throwOnProcessStartUp);`
`509`		`- IWebHostRpcWorkerChannelManager testWebHostLanguageWorkerChannelManager = new TesRpcWorkerChannelManager(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, testLanguageWorkerChannelFactory);`
	`509`	`+ IWebHostRpcWorkerChannelManager testWebHostLanguageWorkerChannelManager = new TestRpcWorkerChannelManager(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, testLanguageWorkerChannelFactory);`
`510`	`510`	`IJobHostRpcWorkerChannelManager jobHostLanguageWorkerChannelManager = new JobHostRpcWorkerChannelManager(loggerFactory);`
`511`	`511`	`if (addWebhostChannel)`
`512`	`512`	`{`
Original file line number	Diff line number	Diff line change
`@@ -130,5 +130,12 @@ public bool IsExecutingInvocation(string invocationId)`
`130`	`130`	`{`
`131`	`131`	`return _executingInvocations.Contains(invocationId);`
`132`	`132`	`}`
	`133`	`+`
	`134`	`+ public bool TryFailExecutions(Exception exception)`
	`135`	`+ {`
	`136`	`+ // Executions are no longer executing`
	`137`	`+ _executingInvocations = new HashSet<string>();`
	`138`	`+ return true;`
	`139`	`+ }`
`133`	`140`	`}`
`134`	`141`	`}`