CxODEV-2161: keep the engine polling after Conductor errors (4.4.2)

src/ConductorSharp.Client/ConductorSharp.Client.csproj

@@ -6,7 +6,7 @@
 		<Authors>Codaxy</Authors>
 		<Company>Codaxy</Company>
 		<PackageId>ConductorSharp.Client</PackageId>
-		<Version>4.4.1</Version>
+		<Version>4.4.2</Version>
 		<Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 		<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 		<PackageTags>netflix;conductor</PackageTags>

src/ConductorSharp.Engine/ConductorSharp.Engine.csproj

@@ -6,7 +6,7 @@
     <Authors>Codaxy</Authors>
 	<Company>Codaxy</Company>
 	<PackageId>ConductorSharp.Engine</PackageId>
-	<Version>4.4.1</Version>
+	<Version>4.4.2</Version>
     <Description>Client library for Netflix Conductor, with some additional quality of life features.</Description>
 	<RepositoryUrl>https://github.com/codaxy/conductor-sharp</RepositoryUrl>
 	<PackageTags>netflix;conductor</PackageTags>

src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs

@@ -5,6 +5,7 @@
 using System.Threading.Tasks;
 using ConductorSharp.Client.Generated;
 using ConductorSharp.Client.Service;
+using ConductorSharp.Engine.Health;
 using Microsoft.Extensions.Logging;
 using Task = System.Threading.Tasks.Task;
 
@@ -19,17 +20,19 @@ internal class TaskQueuePollingService
 
         private readonly ITaskService _taskService;
         private readonly ILogger<TaskQueuePollingService> _logger;
+        private readonly IConductorSharpHealthService _healthService;
         private readonly int _maxAttempts;
         private readonly TimeSpan _initialRetryDelay;
         private readonly TimeSpan _maxRetryDelay;
         private readonly TimeSpan _maxJitter;
         private readonly Func<TimeSpan, CancellationToken, Task> _delay;
         private readonly Func<double> _jitter;
 
-        public TaskQueuePollingService(ITaskService taskService, ILogger<TaskQueuePollingService> logger)
+        public TaskQueuePollingService(ITaskService taskService, ILogger<TaskQueuePollingService> logger, IConductorSharpHealthService healthService)
             : this(
                 taskService,
                 logger,
+                healthService,
                 DefaultMaxAttempts,
                 DefaultInitialRetryDelay,
                 DefaultMaxRetryDelay,
@@ -41,6 +44,7 @@ public TaskQueuePollingService(ITaskService taskService, ILogger<TaskQueuePollin
         internal TaskQueuePollingService(
             ITaskService taskService,
             ILogger<TaskQueuePollingService> logger,
+            IConductorSharpHealthService healthService,
             int maxAttempts,
             TimeSpan initialRetryDelay,
             TimeSpan maxRetryDelay,
@@ -51,6 +55,7 @@ Func<double> jitter
         {
             _taskService = taskService;
             _logger = logger;
+            _healthService = healthService;
             _maxAttempts = maxAttempts;
             _initialRetryDelay = initialRetryDelay;
             _maxRetryDelay = maxRetryDelay;
@@ -67,7 +72,13 @@ public async Task<IDictionary<string, long>> ListQueuesAsync(CancellationToken c
             {
                 try
                 {
-                    return await _taskService.ListQueuesAsync(cancellationToken);
+                    var queues = await _taskService.ListQueuesAsync(cancellationToken);
+
+                    // The engine counts as running once it reads the queues; the background service clears this on failure.
+                    if (!(await _healthService.GetHealthData(cancellationToken)).IsExecutionManagerRunning)
+                        await _healthService.SetExecutionManagerRunning(cancellationToken);
+
+                    return queues;
                 }
                 catch (Exception exception) when (!cancellationToken.IsCancellationRequested && IsTransient(exception) && attempt < _maxAttempts)
                 {

src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs

@@ -10,22 +10,70 @@
 
 namespace ConductorSharp.Engine.Service
 {
-    internal class WorkflowEngineBackgroundService(
-        IConductorSharpHealthService healthService,
-        ILogger<WorkflowEngineBackgroundService> logger,
-        IDeploymentService deploymentService,
-        IExecutionManager executionManager,
-        ModuleDeployment deployment
-    ) : IHostedService, IDisposable
+    internal class WorkflowEngineBackgroundService : IHostedService, IDisposable
     {
-        private readonly ILogger<WorkflowEngineBackgroundService> _logger = logger;
-        private readonly IDeploymentService _deploymentService = deploymentService;
-        private readonly IExecutionManager _executionManager = executionManager;
-        private readonly ModuleDeployment _deployment = deployment;
-        private readonly IConductorSharpHealthService _healthService = healthService;
+        private static readonly TimeSpan DefaultInitialRetryDelay = TimeSpan.FromSeconds(1);
+        private static readonly TimeSpan DefaultMaxRetryDelay = TimeSpan.FromSeconds(30);
+        private static readonly TimeSpan DefaultMaxJitter = TimeSpan.FromMilliseconds(500);
+
+        private readonly ILogger<WorkflowEngineBackgroundService> _logger;
+        private readonly IDeploymentService _deploymentService;
+        private readonly IExecutionManager _executionManager;
+        private readonly ModuleDeployment _deployment;
+        private readonly IConductorSharpHealthService _healthService;
+        private readonly TimeSpan _initialRetryDelay;
+        private readonly TimeSpan _maxRetryDelay;
+        private readonly TimeSpan _maxJitter;
+        private readonly Func<TimeSpan, CancellationToken, Task> _delay;
+        private readonly Func<double> _jitter;
         private Task _executingTask;
         private readonly CancellationTokenSource _stoppingCts = new();
 
+        public WorkflowEngineBackgroundService(
+            IConductorSharpHealthService healthService,
+            ILogger<WorkflowEngineBackgroundService> logger,
+            IDeploymentService deploymentService,
+            IExecutionManager executionManager,
+            ModuleDeployment deployment
+        )
+            : this(
+                healthService,
+                logger,
+                deploymentService,
+                executionManager,
+                deployment,
+                DefaultInitialRetryDelay,
+                DefaultMaxRetryDelay,
+                DefaultMaxJitter,
+                Task.Delay,
+                Random.Shared.NextDouble
+            ) { }
+
+        internal WorkflowEngineBackgroundService(
+            IConductorSharpHealthService healthService,
+            ILogger<WorkflowEngineBackgroundService> logger,
+            IDeploymentService deploymentService,
+            IExecutionManager executionManager,
+            ModuleDeployment deployment,
+            TimeSpan initialRetryDelay,
+            TimeSpan maxRetryDelay,
+            TimeSpan maxJitter,
+            Func<TimeSpan, CancellationToken, Task> delay,
+            Func<double> jitter
+        )
+        {
+            _healthService = healthService;
+            _logger = logger;
+            _deploymentService = deploymentService;
+            _executionManager = executionManager;
+            _deployment = deployment;
+            _initialRetryDelay = initialRetryDelay;
+            _maxRetryDelay = maxRetryDelay;
+            _maxJitter = maxJitter;
+            _delay = delay;
+            _jitter = jitter;
+        }
+
         public Task StartAsync(CancellationToken cancellationToken)
         {
             _executingTask = RunAsync(_stoppingCts.Token);
@@ -38,31 +86,67 @@ public Task StartAsync(CancellationToken cancellationToken)
             return Task.CompletedTask;
         }
 
+        /// <summary>
+        /// Deploys the definitions and runs the execution manager until the service stops. Any failure, including a
+        /// Conductor request timeout, marks the engine as not running and starts both again after a capped backoff.
+        /// The engine is marked running by its first successful queue read (<see cref="TaskQueuePollingService"/>);
+        /// only a run that got that far starts the backoff over.
+        /// </summary>
         private async Task RunAsync(CancellationToken cancellationToken)
         {
             try
             {
                 _healthService.RemoveHealthData();
-                await _deploymentService.Deploy(_deployment);
-                await _healthService.SetExecutionManagerRunning(cancellationToken);
-                await _executionManager.StartAsync(cancellationToken);
-            }
-            catch (TaskCanceledException)
-            {
+                var retryDelay = _initialRetryDelay;
+                var attempt = 0;
+
+                while (!cancellationToken.IsCancellationRequested)
+                {
+                    try
+                    {
+                        // Deploy takes no token; WaitAsync lets a shutdown leave a hung request behind.
+                        await _deploymentService.Deploy(_deployment).WaitAsync(cancellationToken);
+                        await _executionManager.StartAsync(cancellationToken);
+                    }
+                    catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                    {
+                        break;
+                    }
+                    catch (Exception exception)
+                    {
+                        if ((await _healthService.GetHealthData(CancellationToken.None)).IsExecutionManagerRunning)
+                        {
+                            retryDelay = _initialRetryDelay;
+                            attempt = 0;
+                        }
+
+                        attempt++;
+                        await _healthService.UnsetExecutionManagerRunning(CancellationToken.None);
+                        var delay = retryDelay + TimeSpan.FromMilliseconds(_jitter() * _maxJitter.TotalMilliseconds);
+
+                        _logger.LogError(
+                            exception,
+                            "ConductorSharp engine stopped polling (attempt {Attempt}); retrying in {RetryDelay}. API errors: {ApiErrors}",
+                            attempt,
+                            delay,
+                            (exception as ApiException)?.Errors
+                        );
+
+                        try
+                        {
+                            await _delay(delay, cancellationToken);
+                        }
+                        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                        {
+                            break;
+                        }
+
+                        retryDelay = TimeSpan.FromMilliseconds(Math.Min(retryDelay.TotalMilliseconds * 2, _maxRetryDelay.TotalMilliseconds));
+                    }
+                }
+
                 _logger.LogInformation("Stopping ConductorSharp background service");
             }
-            catch (ApiException exception)
-            {
-                await _healthService.UnsetExecutionManagerRunning(cancellationToken);
-                _logger.LogCritical(exception, "Workflow Engine Background Service encountered an API error(s): {apiErrors}", exception.Errors);
-                throw;
-            }
-            catch (Exception exception)
-            {
-                await _healthService.UnsetExecutionManagerRunning(cancellationToken);
-                _logger.LogCritical(exception, "Workflow Engine Background Service encountered an error");
-                throw;
-            }
             finally
             {
                 _healthService.RemoveHealthData();

src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj

@@ -4,7 +4,7 @@
     <TargetFramework>net6.0</TargetFramework>
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
-    <Version>4.4.1</Version>
+    <Version>4.4.2</Version>
     <Authors>Codaxy</Authors>
     <Company>Codaxy</Company>
   </PropertyGroup>

src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj

@@ -7,7 +7,7 @@
     <GeneratePackageOnBuild>False</GeneratePackageOnBuild>
     <Authors>Codaxy</Authors>
     <Company>Codaxy</Company>
-    <Version>4.4.1</Version>
+    <Version>4.4.2</Version>
   </PropertyGroup>
 
 	<ItemGroup>

src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj

@@ -7,7 +7,7 @@
     <Nullable>disable</Nullable>
 	<PackAsTool>true</PackAsTool>
 	<ToolCommandName>dotnet-conductorsharp</ToolCommandName>
-	<Version>4.4.1</Version>
+	<Version>4.4.2</Version>
   </PropertyGroup>
 
   <ItemGroup>