diff --git a/README.md b/README.md index 53ed2683..e56a0543 100644 --- a/README.md +++ b/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 diff --git a/SKILL.md b/SKILL.md index b8033ec5..31108d1a 100644 --- a/SKILL.md +++ b/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 diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 517eb589..006c6dee 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 4.0.0 + 4.0.1 Client library for Netflix Conductor, with some additional quality of life features. https://github.com/codaxy/conductor-sharp netflix;conductor diff --git a/src/ConductorSharp.Engine/AssemblyInfo.cs b/src/ConductorSharp.Engine/AssemblyInfo.cs new file mode 100644 index 00000000..f7591f45 --- /dev/null +++ b/src/ConductorSharp.Engine/AssemblyInfo.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("ConductorSharp.Engine.Tests")] diff --git a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index 06b1d54c..4463a61a 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 4.0.0 + 4.0.1 Client library for Netflix Conductor, with some additional quality of life features. https://github.com/codaxy/conductor-sharp netflix;conductor diff --git a/src/ConductorSharp.Engine/ExecutionManager.cs b/src/ConductorSharp.Engine/ExecutionManager.cs index 259b4aba..5288c5b4 100644 --- a/src/ConductorSharp.Engine/ExecutionManager.cs +++ b/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); diff --git a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs index 5c59e1e7..912acb50 100644 --- a/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs +++ b/src/ConductorSharp.Engine/Extensions/ConductorSharpBuilder.cs @@ -43,7 +43,9 @@ params Assembly[] handlerAssemblies Builder.AddTransient(); - Builder.AddSingleton(); + Builder.AddSingleton(); + + Builder.AddSingleton(); Builder.AddScoped(); @@ -62,10 +64,10 @@ params Assembly[] handlerAssemblies public IExecutionManagerBuilder UseBetaExecutionManager() { - Builder.AddSingleton(); + Builder.AddSingleton(); return this; } - + public IExecutionManagerBuilder AddPipelines(Action behaviorBuilder) { var pipelineBuilder = new PipelineBuilder(Builder); diff --git a/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs b/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs new file mode 100644 index 00000000..f293217c --- /dev/null +++ b/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 _logger; + private readonly int _maxAttempts; + private readonly TimeSpan _initialRetryDelay; + private readonly TimeSpan _maxRetryDelay; + private readonly TimeSpan _maxJitter; + private readonly Func _delay; + private readonly Func _jitter; + + public TaskQueuePollingService(ITaskService taskService, ILogger logger) + : this( + taskService, + logger, + DefaultMaxAttempts, + DefaultInitialRetryDelay, + DefaultMaxRetryDelay, + DefaultMaxJitter, + Task.Delay, + Random.Shared.NextDouble + ) { } + + internal TaskQueuePollingService( + ITaskService taskService, + ILogger logger, + int maxAttempts, + TimeSpan initialRetryDelay, + TimeSpan maxRetryDelay, + TimeSpan maxJitter, + Func delay, + Func jitter + ) + { + _taskService = taskService; + _logger = logger; + _maxAttempts = maxAttempts; + _initialRetryDelay = initialRetryDelay; + _maxRetryDelay = maxRetryDelay; + _maxJitter = maxJitter; + _delay = delay; + _jitter = jitter; + } + + public async Task> 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); + } + } +} diff --git a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs b/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs index fc12ac34..9c18eb52 100644 --- a/src/ConductorSharp.Engine/TypePollSpreadingExecutionManager.cs +++ b/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); diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index ce8e943b..7d05745d 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 4.0.0 + 4.0.1 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 4043ce42..c4e9f4bc 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 4.0.0 + 4.0.1 diff --git a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj index 973ab13f..40c7acfc 100644 --- a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj +++ b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj @@ -7,7 +7,7 @@ disable true dotnet-conductorsharp - 4.0.0 + 4.0.1 diff --git a/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs b/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs new file mode 100644 index 00000000..38db4d24 --- /dev/null +++ b/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(); + 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(); + var service = CreateService(handler, delays); + + var exception = await Assert.ThrowsAsync(() => 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(); + var service = CreateService(handler, delays); + + var exception = await Assert.ThrowsAsync(() => 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 delays) + { + var httpClient = new HttpClient(handler) { BaseAddress = new Uri("http://conductor/") }; + var taskService = new TaskService(httpClient); + + return new TaskQueuePollingService( + taskService, + NullLogger.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 _statuses = new(statuses); + + public int RequestCount { get; private set; } + + protected override Task 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"), }); + } + } +} diff --git a/test/ConductorSharp.Engine.Tests/Usings.cs b/test/ConductorSharp.Engine.Tests/Usings.cs index eb6fe6a9..12d909d5 100644 --- a/test/ConductorSharp.Engine.Tests/Usings.cs +++ b/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;