Add unit tests for worker exit

Author: jviau

src/WebJobs.Script.Grpc/Channel/GrpcWorkerChannel.cs

@@ -388,10 +388,15 @@ public async Task StartWorkerProcessAsync(CancellationToken cancellationToken)
             _workerChannelLogger.LogDebug("Initiating Worker Process start up");
             await _rpcWorkerProcess.StartProcessAsync(cancellationToken);
             _state |= RpcWorkerChannelState.Initializing;
-            Task winner = await Task.WhenAny(
-                _workerInitTask.Task, _rpcWorkerProcess.WaitForExitAsync(cancellationToken))
-                .WaitAsync(cancellationToken);
+            Task exited = _rpcWorkerProcess.WaitForExitAsync(cancellationToken);
+            Task winner = await Task.WhenAny(_workerInitTask.Task, exited).WaitAsync(cancellationToken);
             await winner;
+
+            if (winner == exited)
+            {
+                // process exited without throwing. We need to throw to indicate process is not running.
+                throw new WorkerProcessExitException("Worker process exited before initializing.");
+            }
         }
 
         public async Task<WorkerStatus> GetWorkerStatusAsync()

src/WebJobs.Script/Workers/ProcessManagement/WorkerProcess.cs

@@ -85,9 +85,9 @@ public Task StartProcessAsync(CancellationToken cancellationToken = default)
                 _processExit = new();
                 try
                 {
-                    Process.ErrorDataReceived += (sender, e) => OnErrorDataReceived(sender, e);
-                    Process.OutputDataReceived += (sender, e) => OnOutputDataReceived(sender, e);
-                    Process.Exited += (sender, e) => OnProcessExited(sender, e);
+                    Process.ErrorDataReceived += OnErrorDataReceived;
+                    Process.OutputDataReceived += OnOutputDataReceived;
+                    Process.Exited += OnProcessExited;
                     Process.EnableRaisingEvents = true;
                     string sanitizedArguments = Sanitizer.Sanitize(Process.StartInfo.Arguments);
 

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

@@ -39,9 +39,9 @@ public class GrpcWorkerChannelTests : IDisposable
         private static string _expectedSystemLogMessage = "Random system log message";
         private static string _expectedLoadMsgPartial = "Sending FunctionLoadRequest for ";
 
-        private readonly Mock<IWorkerProcess> _mockrpcWorkerProcess = new Mock<IWorkerProcess>();
+        private readonly Mock<IWorkerProcess> _mockRpcWorkerProcess = new Mock<IWorkerProcess>();
         private readonly string _workerId = "testWorkerId";
-        private readonly string _scriptRootPath = "c:\testdir";
+        private readonly string _scriptRootPath = "c:\\testdir";
         private readonly IScriptEventManager _eventManager = new ScriptEventManager();
         private readonly Mock<IScriptHostManager> _mockScriptHostManager = new Mock<IScriptHostManager>(MockBehavior.Strict);
         private readonly TestMetricsLogger _metricsLogger = new TestMetricsLogger();
@@ -57,7 +57,6 @@ public class GrpcWorkerChannelTests : IDisposable
         private readonly IOptionsMonitor<ScriptApplicationHostOptions> _hostOptionsMonitor;
         private readonly IMemoryMappedFileAccessor _mapAccessor;
         private readonly ISharedMemoryManager _sharedMemoryManager;
-        private readonly IFunctionDataCache _functionDataCache;
         private readonly IOptions<WorkerConcurrencyOptions> _workerConcurrencyOptions;
         private readonly ITestOutputHelper _testOutput;
         private readonly IOptions<FunctionsHostingConfigOptions> _hostingConfigOptions;
@@ -76,24 +75,24 @@ public GrpcWorkerChannelTests(ITestOutputHelper testOutput)
             _testWorkerConfig.CountOptions.InitializationTimeout = TimeSpan.FromSeconds(5);
             _testWorkerConfig.CountOptions.EnvironmentReloadTimeout = TimeSpan.FromSeconds(5);
 
-            _mockrpcWorkerProcess.Setup(m => m.StartProcessAsync(default)).Returns(Task.CompletedTask);
-            _mockrpcWorkerProcess.Setup(m => m.Id).Returns(910);
+            _mockRpcWorkerProcess.Setup(m => m.StartProcessAsync(It.IsAny<CancellationToken>())).Returns(Task.CompletedTask);
+            _mockRpcWorkerProcess.Setup(m => m.WaitForExitAsync(It.IsAny<CancellationToken>())).Returns(Task.Delay(Timeout.Infinite));
+            _mockRpcWorkerProcess.Setup(m => m.Id).Returns(910);
             _testEnvironment = new TestEnvironment();
             _testEnvironment.SetEnvironmentVariable(FunctionDataCacheConstants.FunctionDataCacheEnabledSettingName, "1");
             _workerConcurrencyOptions = Options.Create(new WorkerConcurrencyOptions());
             _workerConcurrencyOptions.Value.CheckInterval = TimeSpan.FromSeconds(1);
 
-            ILogger<MemoryMappedFileAccessor> mmapAccessorLogger = NullLogger<MemoryMappedFileAccessor>.Instance;
+            ILogger<MemoryMappedFileAccessor> mMapAccessorLogger = NullLogger<MemoryMappedFileAccessor>.Instance;
             if (RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
             {
-                _mapAccessor = new MemoryMappedFileAccessorWindows(mmapAccessorLogger);
+                _mapAccessor = new MemoryMappedFileAccessorWindows(mMapAccessorLogger);
             }
             else
             {
-                _mapAccessor = new MemoryMappedFileAccessorUnix(mmapAccessorLogger, _testEnvironment);
+                _mapAccessor = new MemoryMappedFileAccessorUnix(mMapAccessorLogger, _testEnvironment);
             }
             _sharedMemoryManager = new SharedMemoryManager(_loggerFactory, _mapAccessor);
-            _functionDataCache = new FunctionDataCache(_sharedMemoryManager, _loggerFactory, _testEnvironment);
 
             var hostOptions = new ScriptApplicationHostOptions
             {
@@ -127,7 +126,7 @@ private Task CreateDefaultWorkerChannel(bool autoStart = true, IDictionary<strin
                _eventManager,
                _mockScriptHostManager.Object,
                _testWorkerConfig,
-               _mockrpcWorkerProcess.Object,
+               _mockRpcWorkerProcess.Object,
                _logger,
                _metricsLogger,
                0,
@@ -182,6 +181,77 @@ await Assert.ThrowsAsync<TaskCanceledException>(async () =>
             });
         }
 
+        [Fact]
+        public async Task StartWorkerProcessAsync_ProcessExits_Throws()
+        {
+            _mockRpcWorkerProcess.Setup(m => m.WaitForExitAsync(It.IsAny<CancellationToken>()))
+                .Returns(Task.CompletedTask);
+            await CreateDefaultWorkerChannel(autoStart: false);
+
+            WorkerProcessExitException ex = await Assert.ThrowsAsync<WorkerProcessExitException>(
+                () => _workerChannel.StartWorkerProcessAsync(default))
+                    .WaitAsync(TimeSpan.FromMilliseconds(100));
+            Assert.Equal(0, ex.ExitCode);
+            Assert.Equal("Worker process exited before initializing.", ex.Message);
+        }
+
+        [Fact]
+        public async Task StartWorkerProcessAsync_ProcessFaults_Throws()
+        {
+            WorkerProcessExitException expected = new("Process has faulted.") { ExitCode = -1 };
+            _mockRpcWorkerProcess.Setup(m => m.WaitForExitAsync(It.IsAny<CancellationToken>()))
+                .ThrowsAsync(expected);
+            await CreateDefaultWorkerChannel(autoStart: false);
+
+            WorkerProcessExitException actual = await Assert.ThrowsAsync<WorkerProcessExitException>(
+                () => _workerChannel.StartWorkerProcessAsync(default))
+                    .WaitAsync(TimeSpan.FromMilliseconds(100));
+            Assert.Equal(expected, actual);
+        }
+
+        [Fact]
+        public async Task StartWorkerProcessAsync_TimesOut()
+        {
+            await CreateDefaultWorkerChannel(autoStart: false); // suppress for timeout
+            var initTask = _workerChannel.StartWorkerProcessAsync(CancellationToken.None);
+            await Assert.ThrowsAsync<TimeoutException>(async () => await initTask);
+        }
+
+        [Fact]
+        public async Task StartWorkerProcessAsync_WorkerProcess_Throws()
+        {
+            // note: uses custom worker channel
+            Mock<IWorkerProcess> mockrpcWorkerProcessThatThrows = new Mock<IWorkerProcess>();
+            mockrpcWorkerProcessThatThrows.Setup(m => m.StartProcessAsync(default)).Throws<FileNotFoundException>();
+
+            _workerChannel = new GrpcWorkerChannel(
+               _workerId,
+               _eventManager,
+               _mockScriptHostManager.Object,
+               _testWorkerConfig,
+               mockrpcWorkerProcessThatThrows.Object,
+               _logger,
+               _metricsLogger,
+               0,
+               _testEnvironment,
+               _hostOptionsMonitor,
+               _sharedMemoryManager,
+               _workerConcurrencyOptions,
+               _hostingConfigOptions,
+               _httpProxyService);
+            await Assert.ThrowsAsync<FileNotFoundException>(async () => await _workerChannel.StartWorkerProcessAsync(CancellationToken.None));
+        }
+
+        [Fact]
+        public async Task StartWorkerProcessAsync_Invoked_SetupFunctionBuffers_Verify_ReadyForInvocation()
+        {
+            await CreateDefaultWorkerChannel();
+            _mockRpcWorkerProcess.Verify(m => m.StartProcessAsync(default), Times.Once);
+            Assert.False(_workerChannel.IsChannelReadyForInvocations());
+            _workerChannel.SetupFunctionInvocationBuffers(GetTestFunctionsList("node"));
+            Assert.True(_workerChannel.IsChannelReadyForInvocations());
+        }
+
         [Fact]
         public async Task WorkerChannel_Dispose_With_WorkerTerminateCapability()
         {
@@ -221,16 +291,6 @@ public async Task WorkerChannel_Dispose_Without_WorkerTerminateCapability()
             Assert.False(traces.Any(m => string.Equals(m.FormattedMessage, expectedLogMsg)));
         }
 
-        [Fact]
-        public async Task StartWorkerProcessAsync_Invoked_SetupFunctionBuffers_Verify_ReadyForInvocation()
-        {
-            await CreateDefaultWorkerChannel();
-            _mockrpcWorkerProcess.Verify(m => m.StartProcessAsync(default), Times.Once);
-            Assert.False(_workerChannel.IsChannelReadyForInvocations());
-            _workerChannel.SetupFunctionInvocationBuffers(GetTestFunctionsList("node"));
-            Assert.True(_workerChannel.IsChannelReadyForInvocations());
-        }
-
         [Fact]
         public async Task DisposingChannel_NotReadyForInvocation()
         {
@@ -259,14 +319,6 @@ public void SetupFunctionBuffers_Verify_ReadyForInvocation_Returns_False()
             Assert.False(_workerChannel.IsChannelReadyForInvocations());
         }
 
-        [Fact]
-        public async Task StartWorkerProcessAsync_TimesOut()
-        {
-            await CreateDefaultWorkerChannel(autoStart: false); // suppress for timeout
-            var initTask = _workerChannel.StartWorkerProcessAsync(CancellationToken.None);
-            await Assert.ThrowsAsync<TimeoutException>(async () => await initTask);
-        }
-
         [Fact]
         public async Task SendEnvironmentReloadRequest_Generates_ExpectedMetrics()
         {
@@ -283,31 +335,6 @@ public async Task SendEnvironmentReloadRequest_Generates_ExpectedMetrics()
             Assert.True(_metricsLogger.EventsBegan.Contains(MetricEventNames.SpecializationEnvironmentReloadRequestResponse));
         }
 
-        [Fact]
-        public async Task StartWorkerProcessAsync_WorkerProcess_Throws()
-        {
-            // note: uses custom worker channel
-            Mock<IWorkerProcess> mockrpcWorkerProcessThatThrows = new Mock<IWorkerProcess>();
-            mockrpcWorkerProcessThatThrows.Setup(m => m.StartProcessAsync(default)).Throws<FileNotFoundException>();
-
-            _workerChannel = new GrpcWorkerChannel(
-               _workerId,
-               _eventManager,
-               _mockScriptHostManager.Object,
-               _testWorkerConfig,
-               mockrpcWorkerProcessThatThrows.Object,
-               _logger,
-               _metricsLogger,
-               0,
-               _testEnvironment,
-               _hostOptionsMonitor,
-               _sharedMemoryManager,
-               _workerConcurrencyOptions,
-               _hostingConfigOptions,
-               _httpProxyService);
-            await Assert.ThrowsAsync<FileNotFoundException>(async () => await _workerChannel.StartWorkerProcessAsync(CancellationToken.None));
-        }
-
         [Fact]
         public async Task SendWorkerInitRequest_PublishesOutboundEvent()
         {
@@ -547,7 +574,7 @@ public async Task Drain_Verify()
                _eventManager,
                _mockScriptHostManager.Object,
                _testWorkerConfig,
-               _mockrpcWorkerProcess.Object,
+               _mockRpcWorkerProcess.Object,
                _logger,
                _metricsLogger,
                0,
@@ -1256,7 +1283,7 @@ public async Task GetLatencies_StartsTimer_WhenDynamicConcurrencyEnabled()
                _eventManager,
                _mockScriptHostManager.Object,
                config,
-               _mockrpcWorkerProcess.Object,
+               _mockRpcWorkerProcess.Object,
                _logger,
                _metricsLogger,
                0,
@@ -1297,7 +1324,7 @@ public async Task GetLatencies_DoesNot_StartTimer_WhenDynamicConcurrencyDisabled
                _eventManager,
                _mockScriptHostManager.Object,
                config,
-               _mockrpcWorkerProcess.Object,
+               _mockRpcWorkerProcess.Object,
                _logger,
                _metricsLogger,
                0,

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

@@ -2,10 +2,9 @@
 // Licensed under the MIT License. See License.txt in the project root for license information.
 
 using System;
-using System.Collections;
-using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading.Tasks;
 using Microsoft.Azure.WebJobs.Host.Scale;
 using Microsoft.Azure.WebJobs.Script.Config;
 using Microsoft.Azure.WebJobs.Script.Eventing;
@@ -260,5 +259,62 @@ public void WorkerProcess_WaitForExit_AfterExit_DoesNotThrow()
             Exception ex = traces.Single().Exception;
             Assert.IsType<InvalidOperationException>(ex);
         }
+
+        [Fact]
+        public async Task WorkerProcess_WaitForExit_NotStarted_Throws()
+        {
+            using var rpcWorkerProcess = GetRpcWorkerConfigProcess(
+                TestHelpers.GetTestWorkerConfigsWithExecutableWorkingDirectory().ElementAt(0));
+            await Assert.ThrowsAsync<InvalidOperationException>(() => rpcWorkerProcess.WaitForExitAsync());
+        }
+
+        [Fact]
+        public async Task WorkerProcess_WaitForExit_Success_TaskCompletes()
+        {
+            // arrange
+            using Process process = GetProcess(exitCode: 0);
+            _hostProcessMonitorMock.Setup(m => m.RegisterChildProcess(process));
+            _workerProcessFactory.Setup(m => m.CreateWorkerProcess(It.IsNotNull<WorkerContext>())).Returns(process);
+            using var rpcWorkerProcess = GetRpcWorkerConfigProcess(
+                TestHelpers.GetTestWorkerConfigsWithExecutableWorkingDirectory().ElementAt(0));
+
+            // act
+            await rpcWorkerProcess.StartProcessAsync();
+
+            // assert
+            await rpcWorkerProcess.WaitForExitAsync();
+        }
+
+        [Fact]
+        public async Task WorkerProcess_WaitForExit_Error_Rethrows()
+        {
+            // arrange
+            using Process process = GetProcess(exitCode: -1);
+            _hostProcessMonitorMock.Setup(m => m.RegisterChildProcess(process));
+            _workerProcessFactory.Setup(m => m.CreateWorkerProcess(It.IsNotNull<WorkerContext>())).Returns(process);
+            using var rpcWorkerProcess = GetRpcWorkerConfigProcess(
+                TestHelpers.GetTestWorkerConfigsWithExecutableWorkingDirectory().ElementAt(0));
+
+            // act
+            await rpcWorkerProcess.StartProcessAsync();
+
+            // assert
+            await Assert.ThrowsAnyAsync<WorkerProcessExitException>(() => rpcWorkerProcess.WaitForExitAsync());
+        }
+
+        private static Process GetProcess(int exitCode)
+        {
+            return new()
+            {
+                StartInfo = new()
+                {
+                    WindowStyle = ProcessWindowStyle.Hidden,
+                    FileName = OperatingSystem.IsWindows() ? "cmd" : "bash",
+                    Arguments = OperatingSystem.IsWindows() ? $"/C exit {exitCode}" : $"-c \"exit {exitCode}\"",
+                    RedirectStandardError = true,
+                    RedirectStandardOutput = true,
+                }
+            };
+        }
     }
-}
+}