initializing ErrorEventsThreshold

release_notes.md

@@ -4,5 +4,6 @@
 - My change description (#PR)
 -->
 
+- Fixed worker startup errors during metadata indexing immediately exhausting the language worker restart budget instead of retrying. (#11955)
 - Fixed Linux language worker SIGTERM exits being reported as worker failures. (#11944)
 - Prevent extension system keys from being regenerated and overwritten when the startup context cache is stale, which previously could invalidate already-published extension webhook URLs (e.g. Event Grid, Durable Task). (#11936)

src/Functions.Rpc.Server/Rpc/FunctionRegistration/RpcFunctionInvocationDispatcher.cs

@@ -26,6 +26,7 @@ namespace Microsoft.Azure.WebJobs.Script.Workers.Rpc
 {
     internal class RpcFunctionInvocationDispatcher : IFunctionInvocationDispatcher
     {
+        private const int ErrorEventsThresholdMultiplier = 3;
         private static readonly int MultiLanguageDefaultProcessCount = 1;
 
         private readonly IMetricsLogger _metricsLogger;
@@ -102,21 +103,20 @@ public RpcFunctionInvocationDispatcher(IOptions<ScriptJobHostOptions> scriptHost
             _hostingConfigOptions = hostingConfigOptions;
             _hostMetrics = hostMetrics ?? throw new ArgumentNullException(nameof(hostMetrics));
             State = FunctionInvocationDispatcherState.Default;
+            _maxProcessCount = new Lazy<Task<int>>(GetMaxProcessCount);
+            InitializeErrorEventsThreshold(_workerRuntime);
 
             _workerErrorSubscription = _eventManager.OfType<WorkerErrorEvent>().Subscribe(WorkerError);
             _workerRestartSubscription = _eventManager.OfType<WorkerRestartEvent>().Subscribe(WorkerRestart);
-
             _shutdownStandbyWorkerChannels = ShutdownWebhostLanguageWorkerChannels;
             _shutdownStandbyWorkerChannels = _shutdownStandbyWorkerChannels.Debounce(milliseconds: 5000);
-
-            _maxProcessCount = new Lazy<Task<int>>(GetMaxProcessCount);
         }
 
         internal Task<int> MaxProcessCount => _maxProcessCount.Value;
 
         public FunctionInvocationDispatcherState State { get; private set; }
 
-        public int ErrorEventsThreshold { get; private set; }
+        public int ErrorEventsThreshold { get; private set; } = ErrorEventsThresholdMultiplier;
 
         public IJobHostRpcWorkerChannelManager JobHostLanguageWorkerChannelManager => _jobHostLanguageWorkerChannelManager;
 
@@ -149,6 +149,30 @@ private async Task<int> GetMaxProcessCount()
             return (await GetAllWorkerChannelsAsync()).Count();
         }
 
+        private void InitializeErrorEventsThreshold(string workerRuntime)
+        {
+            if (_environment.IsMultiLanguageRuntimeEnvironment())
+            {
+                SetErrorEventsThreshold(MultiLanguageDefaultProcessCount);
+                return;
+            }
+
+            if (!string.IsNullOrEmpty(workerRuntime))
+            {
+                var workerConfig = _workerConfigs
+                    .FirstOrDefault(c => string.Equals(c.Description?.Language, workerRuntime, StringComparison.InvariantCultureIgnoreCase));
+                if (workerConfig?.CountOptions is not null)
+                {
+                    SetErrorEventsThreshold(workerConfig.CountOptions.ProcessCount);
+                }
+            }
+        }
+
+        private void SetErrorEventsThreshold(int maxProcessCount)
+        {
+            ErrorEventsThreshold = ErrorEventsThresholdMultiplier * Math.Max(MultiLanguageDefaultProcessCount, maxProcessCount);
+        }
+
         internal async Task InitializeJobhostLanguageWorkerChannelAsync(IEnumerable<string> languages = null)
         {
             if (languages == null)
@@ -174,8 +198,11 @@ internal async Task InitializeJobhostLanguageWorkerChannelAsync(int attemptCount
                 _logger.LogDebug("Adding jobhost language worker channel for runtime: {language}. workerId:{id}", language, rpcWorkerChannel.Id);
 
                 // if the worker is indexing, we will not have function metadata yet. So, we cannot set up invocation buffers or send load requests
-                rpcWorkerChannel.SetupFunctionInvocationBuffers(_functions);
-                rpcWorkerChannel.SendFunctionLoadRequests(_managedDependencyOptions.Value, _scriptOptions.FunctionTimeout);
+                if (_functions is not null)
+                {
+                    rpcWorkerChannel.SetupFunctionInvocationBuffers(_functions);
+                    rpcWorkerChannel.SendFunctionLoadRequests(_managedDependencyOptions.Value, _scriptOptions.FunctionTimeout);
+                }
             }
             SetFunctionDispatcherStateToInitializedAndLog();
         }
@@ -325,7 +352,9 @@ public async Task InitializeAsync(IEnumerable<FunctionMetadata> functions, Cance
                 _restartWait = workerConfig?.CountOptions.ProcessRestartInterval ?? _defaultProcessRestartInterval;
                 _shutdownTimeout = workerConfig?.CountOptions.ProcessShutdownTimeout ?? _defaultProcessShutdownInterval;
             }
-            ErrorEventsThreshold = 3 * await _maxProcessCount.Value;
+
+            // This was initialized to a default, but it's possible that the worker name was not known at that time. Re-initialize now just in case it's changed.
+            SetErrorEventsThreshold(await _maxProcessCount.Value);
 
             if (Utility.IsSupportedRuntime(_workerRuntime, _workerConfigs) || _environment.IsMultiLanguageRuntimeEnvironment())
             {

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

@@ -44,6 +44,53 @@ public RpcFunctionInvocationDispatcherTests()
             EnvironmentExtensions.ClearCache();
         }
 
+        private enum WorkerRecoveryAction
+        {
+            RestartWorker,
+            StopApplication
+        }
+
+        [Fact]
+        public void FunctionDispatcher_ErrorEventsThreshold_DefaultsToThree()
+        {
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(runtime: null);
+
+            Assert.Equal(3, functionDispatcher.ErrorEventsThreshold);
+        }
+
+        [Fact]
+        public void FunctionDispatcher_ErrorEventsThreshold_DefaultsToThree_WhenWorkerConfigIsIncomplete()
+        {
+            IList<RpcWorkerConfig> workerConfigs =
+            [
+                new(),
+                new()
+                {
+                    Description = new RpcWorkerDescription()
+                },
+                new()
+                {
+                    Description = TestHelpers.GetTestWorkerDescription(RpcWorkerConstants.NodeLanguageWorkerName, ".js"),
+                    CountOptions = null
+                }
+            ];
+
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(
+                runtime: RpcWorkerConstants.NodeLanguageWorkerName, workerConfigs: workerConfigs);
+
+            Assert.Equal(3, functionDispatcher.ErrorEventsThreshold);
+        }
+
+        [Fact]
+        public async Task FunctionDispatcher_ErrorEventsThreshold_RemainsThree_WhenNoWorkerConfigMatches()
+        {
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(runtime: RpcWorkerConstants.PythonLanguageWorkerName);
+
+            await functionDispatcher.InitializeAsync(GetTestFunctionsList(RpcWorkerConstants.PythonLanguageWorkerName));
+
+            Assert.Equal(3, functionDispatcher.ErrorEventsThreshold);
+        }
+
         [Fact]
         public async Task GetWorkerStatusesAsync_ReturnsExpectedResult()
         {
@@ -109,6 +156,80 @@ public async Task WorkerIndexing_Starting_WebhostChannel_Succeeds()
             Assert.Equal(0, finalJobhostChannelCount);
         }
 
+        [Theory]
+        [InlineData(false)]
+        [InlineData(true)]
+        public async Task FunctionDispatcher_WorkerIndexing_FirstWorkerError_RestartsWorker(bool initializeWithEmptyMetadata)
+        {
+            const int processCount = 2;
+            const string workerId = "metadata-worker";
+            IList<RpcWorkerConfig> workerConfigs =
+            [
+                new()
+                {
+                    Description = TestHelpers.GetTestWorkerDescription(
+                        RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, ".dll", workerIndexing: true),
+                    CountOptions = new WorkerProcessCountOptions
+                    {
+                        ProcessCount = processCount
+                    }
+                }
+            ];
+            var recoveryActionSource = new TaskCompletionSource<WorkerRecoveryAction>(TaskCreationOptions.RunContinuationsAsynchronously);
+            var eventManager = new ScriptEventManager();
+            var applicationLifetime = new Mock<IHostApplicationLifetime>();
+            applicationLifetime.SetupGet(m => m.ApplicationStopping).Returns(CancellationToken.None);
+            applicationLifetime.Setup(m => m.StopApplication())
+                .Callback(() => recoveryActionSource.TrySetResult(WorkerRecoveryAction.StopApplication));
+
+            var webHostChannelManager = new Mock<IWebHostRpcWorkerChannelManager>();
+            webHostChannelManager.Setup(m => m.GetChannels(RpcWorkerConstants.DotNetIsolatedLanguageWorkerName))
+                .Returns((IDictionary<string, TaskCompletionSource<IRpcWorkerChannel>>)null);
+            webHostChannelManager.Setup(m => m.ShutdownChannelIfExistsAsync(
+                    RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, workerId, It.IsAny<Exception>()))
+                .ReturnsAsync(true);
+
+            var retryChannel = new Mock<IRpcWorkerChannel>();
+            retryChannel.SetupGet(m => m.Id).Returns("retry-worker");
+            retryChannel.Setup(m => m.StartWorkerProcessAsync(CancellationToken.None))
+                .Callback(() => recoveryActionSource.TrySetResult(WorkerRecoveryAction.RestartWorker))
+                .Returns(Task.CompletedTask);
+
+            var channelFactory = new Mock<IRpcWorkerChannelFactory>();
+            channelFactory.Setup(m => m.Create(
+                    It.IsAny<string>(), RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, It.IsAny<IMetricsLogger>(), 1,
+                    It.IsAny<IEnumerable<RpcWorkerConfig>>()))
+                .Returns(retryChannel.Object);
+
+            RpcFunctionInvocationDispatcher functionDispatcher = GetTestFunctionDispatcher(
+                maxProcessCountValue: processCount,
+                runtime: RpcWorkerConstants.DotNetIsolatedLanguageWorkerName,
+                workerIndexing: true,
+                channelFactory: channelFactory.Object,
+                mockwebHostLanguageWorkerChannelManager: webHostChannelManager,
+                eventManager: eventManager,
+                applicationLifetime: applicationLifetime.Object,
+                workerConfigs: workerConfigs);
+
+            if (initializeWithEmptyMetadata)
+            {
+                await functionDispatcher.InitializeAsync(Array.Empty<FunctionMetadata>());
+            }
+
+            eventManager.Publish(new WorkerErrorEvent(
+                RpcWorkerConstants.DotNetIsolatedLanguageWorkerName, workerId, new TimeoutException("Worker did not send StartStream.")));
+
+            WorkerRecoveryAction recoveryAction = await recoveryActionSource.Task.WaitAsync(TimeSpan.FromSeconds(5));
+
+            Assert.Equal(WorkerRecoveryAction.RestartWorker, recoveryAction);
+            await TestHelpers.Await(() => functionDispatcher.State == FunctionInvocationDispatcherState.Initialized);
+            applicationLifetime.Verify(m => m.StopApplication(), Times.Never);
+            retryChannel.Verify(m => m.SetupFunctionInvocationBuffers(It.IsAny<IEnumerable<FunctionMetadata>>()), Times.Never);
+            retryChannel.Verify(m => m.SendFunctionLoadRequests(It.IsAny<ManagedDependencyOptions>(), It.IsAny<TimeSpan?>()), Times.Never);
+            Assert.Equal(3 * processCount, functionDispatcher.ErrorEventsThreshold);
+            Assert.Single(functionDispatcher.LanguageWorkerErrors);
+        }
+
         [Fact]
         public async Task Starting_MultipleJobhostChannels_Failed()
         {
@@ -765,13 +886,20 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(
             bool workerIndexing = false,
             bool placeholder = false,
             IRpcWorkerChannelFactory channelFactory = null,
-            CancellationTokenSource applicationStoppingSource = null)
+            CancellationTokenSource applicationStoppingSource = null,
+            IScriptEventManager eventManager = null,
+            IHostApplicationLifetime applicationLifetime = null,
+            IList<RpcWorkerConfig> workerConfigs = null)
         {
-            var eventManager = new ScriptEventManager();
+            eventManager ??= new ScriptEventManager();
             var metricsLogger = new Mock<IMetricsLogger>();
-            var mockApplicationLifetime = new Mock<IHostApplicationLifetime>();
             var stoppingSource = applicationStoppingSource ?? new CancellationTokenSource();
-            mockApplicationLifetime.Setup(m => m.ApplicationStopping).Returns(stoppingSource.Token);
+            if (applicationLifetime is null)
+            {
+                var mockApplicationLifetime = new Mock<IHostApplicationLifetime>();
+                mockApplicationLifetime.Setup(m => m.ApplicationStopping).Returns(stoppingSource.Token);
+                applicationLifetime = mockApplicationLifetime.Object;
+            }
             var testEnv = new TestEnvironment();
             TimeSpan intervals = startupIntervals ?? TimeSpan.FromMilliseconds(100);
 
@@ -795,8 +923,8 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(
 
             var workerConfigOptions = new LanguageWorkerOptions
             {
-                WorkerConfigs = TestHelpers.GetTestWorkerConfigs(processCountValue: maxProcessCountValue, processStartupInterval: intervals,
-                                    processRestartInterval: intervals, processShutdownTimeout: TimeSpan.FromSeconds(1), workerIndexing: workerIndexing)
+                WorkerConfigs = workerConfigs ?? TestHelpers.GetTestWorkerConfigs(processCountValue: maxProcessCountValue, processStartupInterval: intervals,
+                    processRestartInterval: intervals, processShutdownTimeout: TimeSpan.FromSeconds(1), workerIndexing: workerIndexing)
             };
 
             channelFactory ??= new TestRpcWorkerChannelFactory(eventManager, _testLogger, scriptOptions.Value.RootScriptPath, throwOnProcessStartUp);
@@ -828,7 +956,7 @@ private static RpcFunctionInvocationDispatcher GetTestFunctionDispatcher(
             return new RpcFunctionInvocationDispatcher(scriptOptions,
                 metricsLogger.Object,
                 testEnv,
-                mockApplicationLifetime.Object,
+                applicationLifetime,
                 eventManager,
                 _testLoggerFactory,
                 channelFactory,