retry non transient conductor failures when polling

README.md

@@ -971,7 +971,7 @@ Generate C# models from existing Conductor task/workflow definitions.
 ### Installation
 
 ```bash
-dotnet tool install --global ConductorSharp.Toolkit --version 4.0.0
+dotnet tool install --global ConductorSharp.Toolkit --version 4.0.1
 ```
 
 ### Configuration

SKILL.md

@@ -866,7 +866,7 @@ public class WorkflowController : ControllerBase
 ### Installation
 
 ```bash
-dotnet tool install --global ConductorSharp.Toolkit --version 4.0.0
+dotnet tool install --global ConductorSharp.Toolkit --version 4.0.1
 ```
 
 ### Configuration

src/ConductorSharp.Client/ConductorSharp.Client.csproj

@@ -6,7 +6,7 @@
 		<Authors>Codaxy</Authors>
 		<Company>Codaxy</Company>
 		<PackageId>ConductorSharp.Client</PackageId>
-		<Version>4.0.0</Version>
+		<Version>4.0.1</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/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("ConductorSharp.Engine.Tests")]

src/ConductorSharp.Engine/ConductorSharp.Engine.csproj

@@ -6,7 +6,7 @@
     <Authors>Codaxy</Authors>
 	<Company>Codaxy</Company>
 	<PackageId>ConductorSharp.Engine</PackageId>
-	<Version>4.0.0</Version>
+	<Version>4.0.1</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/ExecutionManager.cs

@@ -11,6 +11,7 @@
 using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Model;
 using ConductorSharp.Engine.Polling;
+using ConductorSharp.Engine.Service;
 using ConductorSharp.Engine.Util;
 using MediatR;
 using Microsoft.Extensions.DependencyInjection;
@@ -32,6 +33,7 @@ internal class ExecutionManager : IExecutionManager
         private readonly IPollTimingStrategy _pollTimingStrategy;
         private readonly IPollOrderStrategy _pollOrderStrategy;
         private readonly ICancellationNotifier _cancellationNotifier;
+        private readonly TaskQueuePollingService _taskQueuePollingService;
 
         public ExecutionManager(
             WorkerSetConfig options,
@@ -42,7 +44,8 @@ public ExecutionManager(
             IServiceScopeFactory lifetimeScope,
             IPollTimingStrategy pollTimingStrategy,
             IPollOrderStrategy pollOrderStrategy,
-            ICancellationNotifier cancellationNotifier
+            ICancellationNotifier cancellationNotifier,
+            TaskQueuePollingService taskQueuePollingService
         )
         {
             _configuration = options;
@@ -55,6 +58,7 @@ ICancellationNotifier cancellationNotifier
             _pollOrderStrategy = pollOrderStrategy;
             _cancellationNotifier = cancellationNotifier;
             _externalPayloadService = externalPayloadService;
+            _taskQueuePollingService = taskQueuePollingService;
         }
 
         public async Task StartAsync(CancellationToken cancellationToken)
@@ -63,7 +67,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
             while (!cancellationToken.IsCancellationRequested)
             {
-                var queuedTasks = (await _taskManager.ListQueuesAsync(cancellationToken))
+                var queuedTasks = (await _taskQueuePollingService.ListQueuesAsync(cancellationToken))
                     .Where(a => a.Value > 0)
                     .ToDictionary(a => a.Key, a => a.Value);
 

src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs

@@ -43,7 +43,9 @@ params Assembly[] handlerAssemblies
 
             Builder.AddTransient<ModuleDeployment>();
 
-            Builder.AddSingleton<IExecutionManager,ExecutionManager>();
+            Builder.AddSingleton<TaskQueuePollingService>();
+
+            Builder.AddSingleton<IExecutionManager, ExecutionManager>();
 
             Builder.AddScoped<ConductorSharpExecutionContext>();
 
@@ -62,10 +64,10 @@ params Assembly[] handlerAssemblies
 
         public IExecutionManagerBuilder UseBetaExecutionManager()
         {
-            Builder.AddSingleton<IExecutionManager,TypePollSpreadingExecutionManager>();
+            Builder.AddSingleton<IExecutionManager, TypePollSpreadingExecutionManager>();
             return this;
         }
-        
+
         public IExecutionManagerBuilder AddPipelines(Action<IPipelineBuilder> behaviorBuilder)
         {
             var pipelineBuilder = new PipelineBuilder(Builder);

src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Collections.Generic;
+using System.Net.Http;
+using System.Threading;
+using System.Threading.Tasks;
+using ConductorSharp.Client.Generated;
+using ConductorSharp.Client.Service;
+using Microsoft.Extensions.Logging;
+using Task = System.Threading.Tasks.Task;
+
+namespace ConductorSharp.Engine.Service
+{
+    internal class TaskQueuePollingService
+    {
+        private const int DefaultMaxAttempts = 5;
+        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 ITaskService _taskService;
+        private readonly ILogger<TaskQueuePollingService> _logger;
+        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)
+            : this(
+                taskService,
+                logger,
+                DefaultMaxAttempts,
+                DefaultInitialRetryDelay,
+                DefaultMaxRetryDelay,
+                DefaultMaxJitter,
+                Task.Delay,
+                Random.Shared.NextDouble
+            ) { }
+
+        internal TaskQueuePollingService(
+            ITaskService taskService,
+            ILogger<TaskQueuePollingService> logger,
+            int maxAttempts,
+            TimeSpan initialRetryDelay,
+            TimeSpan maxRetryDelay,
+            TimeSpan maxJitter,
+            Func<TimeSpan, CancellationToken, Task> delay,
+            Func<double> jitter
+        )
+        {
+            _taskService = taskService;
+            _logger = logger;
+            _maxAttempts = maxAttempts;
+            _initialRetryDelay = initialRetryDelay;
+            _maxRetryDelay = maxRetryDelay;
+            _maxJitter = maxJitter;
+            _delay = delay;
+            _jitter = jitter;
+        }
+
+        public async Task<IDictionary<string, long>> ListQueuesAsync(CancellationToken cancellationToken)
+        {
+            var retryDelay = _initialRetryDelay;
+
+            for (var attempt = 1; ; attempt++)
+            {
+                try
+                {
+                    return await _taskService.ListQueuesAsync(cancellationToken);
+                }
+                catch (Exception exception) when (!cancellationToken.IsCancellationRequested && IsTransient(exception) && attempt < _maxAttempts)
+                {
+                    var delay = retryDelay + TimeSpan.FromMilliseconds(_jitter() * _maxJitter.TotalMilliseconds);
+
+                    _logger.LogWarning(
+                        exception,
+                        "Failed to read Conductor task queues. Attempt {Attempt}/{MaxAttempts}; retrying in {RetryDelay}",
+                        attempt,
+                        _maxAttempts,
+                        delay
+                    );
+
+                    await _delay(delay, cancellationToken);
+                    retryDelay = TimeSpan.FromMilliseconds(Math.Min(retryDelay.TotalMilliseconds * 2, _maxRetryDelay.TotalMilliseconds));
+                }
+            }
+        }
+
+        private static bool IsTransient(Exception exception)
+        {
+            return exception is HttpRequestException
+                || exception is TaskCanceledException
+                || exception is ApiException apiException && (apiException.StatusCode is 408 or 429 || apiException.StatusCode >= 500);
+        }
+    }
+}

src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs

@@ -11,6 +11,7 @@
 using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Model;
 using ConductorSharp.Engine.Polling;
+using ConductorSharp.Engine.Service;
 using ConductorSharp.Engine.Util;
 using MediatR;
 using Microsoft.Extensions.DependencyInjection;
@@ -32,6 +33,7 @@ internal class TypePollSpreadingExecutionManager : IExecutionManager
         private readonly IPollTimingStrategy _pollTimingStrategy;
         private readonly IPollOrderStrategy _pollOrderStrategy;
         private readonly ICancellationNotifier _cancellationNotifier;
+        private readonly TaskQueuePollingService _taskQueuePollingService;
 
         public TypePollSpreadingExecutionManager(
             WorkerSetConfig options,
@@ -42,7 +44,8 @@ public TypePollSpreadingExecutionManager(
             IServiceScopeFactory lifetimeScope,
             IPollTimingStrategy pollTimingStrategy,
             IPollOrderStrategy pollOrderStrategy,
-            ICancellationNotifier cancellationNotifier
+            ICancellationNotifier cancellationNotifier,
+            TaskQueuePollingService taskQueuePollingService
         )
         {
             _configuration = options;
@@ -55,6 +58,7 @@ ICancellationNotifier cancellationNotifier
             _pollOrderStrategy = pollOrderStrategy;
             _cancellationNotifier = cancellationNotifier;
             _externalPayloadService = externalPayloadService;
+            _taskQueuePollingService = taskQueuePollingService;
         }
 
         public async Task StartAsync(CancellationToken cancellationToken)
@@ -63,7 +67,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
 
             while (!cancellationToken.IsCancellationRequested)
             {
-                var queuedTasks = (await _taskManager.ListQueuesAsync(cancellationToken))
+                var queuedTasks = (await _taskQueuePollingService.ListQueuesAsync(cancellationToken))
                     .Where(a => a.Value > 0)
                     .ToDictionary(a => a.Key, a => a.Value);
 

src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj

@@ -4,7 +4,7 @@
     <TargetFramework>net6.0</TargetFramework>
     <ImplicitUsings>enable</ImplicitUsings>
     <Nullable>enable</Nullable>
-    <Version>4.0.0</Version>
+    <Version>4.0.1</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.0.0</Version>
+    <Version>4.0.1</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.0.0</Version>
+	<Version>4.0.1</Version>
   </PropertyGroup>
 
   <ItemGroup>

test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs

@@ -0,0 +1,97 @@
+using System.Net;
+using System.Text;
+using ConductorSharp.Client.Generated;
+using ConductorSharp.Client.Service;
+using ConductorSharp.Engine.Service;
+using Microsoft.Extensions.Logging.Abstractions;
+using Task = System.Threading.Tasks.Task;
+
+namespace ConductorSharp.Engine.Tests.Unit;
+
+public class TaskQueuePollingServiceTests
+{
+    [Fact]
+    public async Task ListQueuesAsync_RetriesTransientFailuresAndReturnsQueues()
+    {
+        var handler = new SequenceHandler(HttpStatusCode.InternalServerError, HttpStatusCode.ServiceUnavailable, HttpStatusCode.OK);
+        var delays = new List<TimeSpan>();
+        var service = CreateService(handler, delays);
+
+        var queues = await service.ListQueuesAsync(CancellationToken.None);
+
+        Assert.Equal(3, handler.RequestCount);
+        Assert.Equal(2, delays.Count);
+        Assert.Equal(2, queues["test-task"]);
+    }
+
+    [Fact]
+    public async Task ListQueuesAsync_DoesNotRetryNonTransientApiErrors()
+    {
+        var handler = new SequenceHandler(HttpStatusCode.BadRequest, HttpStatusCode.OK);
+        var delays = new List<TimeSpan>();
+        var service = CreateService(handler, delays);
+
+        var exception = await Assert.ThrowsAsync<ApiException>(() => service.ListQueuesAsync(CancellationToken.None));
+
+        Assert.Equal(400, exception.StatusCode);
+        Assert.Equal(1, handler.RequestCount);
+        Assert.Empty(delays);
+    }
+
+    [Fact]
+    public async Task ListQueuesAsync_RethrowsAfterRetryLimit()
+    {
+        var handler = new SequenceHandler(
+            HttpStatusCode.InternalServerError,
+            HttpStatusCode.InternalServerError,
+            HttpStatusCode.InternalServerError,
+            HttpStatusCode.InternalServerError,
+            HttpStatusCode.InternalServerError
+        );
+        var delays = new List<TimeSpan>();
+        var service = CreateService(handler, delays);
+
+        var exception = await Assert.ThrowsAsync<ApiException>(() => service.ListQueuesAsync(CancellationToken.None));
+
+        Assert.Equal(500, exception.StatusCode);
+        Assert.Equal(5, handler.RequestCount);
+        Assert.Equal(4, delays.Count);
+    }
+
+    private static TaskQueuePollingService CreateService(HttpMessageHandler handler, ICollection<TimeSpan> delays)
+    {
+        var httpClient = new HttpClient(handler) { BaseAddress = new Uri("http://conductor/") };
+        var taskService = new TaskService(httpClient);
+
+        return new TaskQueuePollingService(
+            taskService,
+            NullLogger<TaskQueuePollingService>.Instance,
+            maxAttempts: 5,
+            initialRetryDelay: TimeSpan.FromSeconds(1),
+            maxRetryDelay: TimeSpan.FromSeconds(30),
+            maxJitter: TimeSpan.FromMilliseconds(500),
+            delay: (delay, _) =>
+            {
+                delays.Add(delay);
+                return Task.CompletedTask;
+            },
+            jitter: () => 0
+        );
+    }
+
+    private sealed class SequenceHandler(params HttpStatusCode[] statuses) : HttpMessageHandler
+    {
+        private readonly Queue<HttpStatusCode> _statuses = new(statuses);
+
+        public int RequestCount { get; private set; }
+
+        protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
+        {
+            RequestCount++;
+            var status = _statuses.Dequeue();
+            var content = status == HttpStatusCode.OK ? """{"test-task":2}""" : "{}";
+
+            return Task.FromResult(new HttpResponseMessage(status) { Content = new StringContent(content, Encoding.UTF8, "application/json"), });
+        }
+    }
+}

test/ConductorSharp.Engine.Tests/Usings.cs

@@ -6,3 +6,4 @@
 global using MediatR;
 global using Newtonsoft.Json;
 global using Xunit;
+global using EmbeddedFileHelper = ConductorSharp.Engine.Tests.Util.EmbeddedFileHelper;