diff --git a/src/ConductorSharp.Client/ConductorSharp.Client.csproj b/src/ConductorSharp.Client/ConductorSharp.Client.csproj index 1a506640..b88a9d53 100644 --- a/src/ConductorSharp.Client/ConductorSharp.Client.csproj +++ b/src/ConductorSharp.Client/ConductorSharp.Client.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Client - 4.4.1 + 4.4.2 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/ConductorSharp.Engine.csproj b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj index df478cee..97235c2e 100644 --- a/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj +++ b/src/ConductorSharp.Engine/ConductorSharp.Engine.csproj @@ -6,7 +6,7 @@ Codaxy Codaxy ConductorSharp.Engine - 4.4.1 + 4.4.2 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/Service/TaskQueuePollingService.cs b/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs index f293217c..35877f8e 100644 --- a/src/ConductorSharp.Engine/Service/TaskQueuePollingService.cs +++ b/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,6 +20,7 @@ internal class TaskQueuePollingService private readonly ITaskService _taskService; private readonly ILogger _logger; + private readonly IConductorSharpHealthService _healthService; private readonly int _maxAttempts; private readonly TimeSpan _initialRetryDelay; private readonly TimeSpan _maxRetryDelay; @@ -26,10 +28,11 @@ internal class TaskQueuePollingService private readonly Func _delay; private readonly Func _jitter; - public TaskQueuePollingService(ITaskService taskService, ILogger logger) + public TaskQueuePollingService(ITaskService taskService, ILogger logger, IConductorSharpHealthService healthService) : this( taskService, logger, + healthService, DefaultMaxAttempts, DefaultInitialRetryDelay, DefaultMaxRetryDelay, @@ -41,6 +44,7 @@ public TaskQueuePollingService(ITaskService taskService, ILogger logger, + IConductorSharpHealthService healthService, int maxAttempts, TimeSpan initialRetryDelay, TimeSpan maxRetryDelay, @@ -51,6 +55,7 @@ Func jitter { _taskService = taskService; _logger = logger; + _healthService = healthService; _maxAttempts = maxAttempts; _initialRetryDelay = initialRetryDelay; _maxRetryDelay = maxRetryDelay; @@ -67,7 +72,13 @@ public async Task> 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) { diff --git a/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs b/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs index f8d2e84e..18dec395 100644 --- a/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs +++ b/src/ConductorSharp.Engine/Service/WorkflowEngineBackgroundService.cs @@ -10,22 +10,70 @@ namespace ConductorSharp.Engine.Service { - internal class WorkflowEngineBackgroundService( - IConductorSharpHealthService healthService, - ILogger logger, - IDeploymentService deploymentService, - IExecutionManager executionManager, - ModuleDeployment deployment - ) : IHostedService, IDisposable + internal class WorkflowEngineBackgroundService : IHostedService, IDisposable { - private readonly ILogger _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 _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 _delay; + private readonly Func _jitter; private Task _executingTask; private readonly CancellationTokenSource _stoppingCts = new(); + public WorkflowEngineBackgroundService( + IConductorSharpHealthService healthService, + ILogger 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 logger, + IDeploymentService deploymentService, + IExecutionManager executionManager, + ModuleDeployment deployment, + TimeSpan initialRetryDelay, + TimeSpan maxRetryDelay, + TimeSpan maxJitter, + Func delay, + Func 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; } + /// + /// 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 (); + /// only a run that got that far starts the backoff over. + /// 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(); diff --git a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj index 989e052e..7595db51 100644 --- a/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj +++ b/src/ConductorSharp.KafkaCancellationNotifier/ConductorSharp.KafkaCancellationNotifier.csproj @@ -4,7 +4,7 @@ net6.0 enable enable - 4.4.1 + 4.4.2 Codaxy Codaxy diff --git a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj index 70163e9e..c7b264b2 100644 --- a/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj +++ b/src/ConductorSharp.Patterns/ConductorSharp.Patterns.csproj @@ -7,7 +7,7 @@ False Codaxy Codaxy - 4.4.1 + 4.4.2 diff --git a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj index 0b48430b..37157bae 100644 --- a/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj +++ b/src/ConductorSharp.Toolkit/ConductorSharp.Toolkit.csproj @@ -7,7 +7,7 @@ disable true dotnet-conductorsharp - 4.4.1 + 4.4.2 diff --git a/test/ConductorSharp.Engine.Tests/Integration/EngineRecoveryTests.cs b/test/ConductorSharp.Engine.Tests/Integration/EngineRecoveryTests.cs new file mode 100644 index 00000000..741eebc4 --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Integration/EngineRecoveryTests.cs @@ -0,0 +1,335 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Net; +using System.Text; +using ConductorSharp.Client.Service; +using ConductorSharp.Engine.Builders.Metadata; +using ConductorSharp.Engine.Extensions; +using ConductorSharp.Engine.Health; +using ConductorSharp.Engine.Service; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json.Linq; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine.Tests.Integration; + +/// +/// A real engine (DI wiring, hosted service, beta execution manager, HTTP client) against a fake Conductor that can +/// fail, hang and recover. Retry delays are shortened; everything else is production code. +/// +[Collection(nameof(InMemoryHealthService))] +public class EngineRecoveryTests : IAsyncLifetime +{ + private const string ProbeTaskName = "recovery_probe_task"; + private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(10); + + private readonly FakeConductor _conductor = new(); + private readonly ProbeGate _gate = new(); + private ServiceProvider _provider; + private List _hostedServices = []; + private IConductorSharpHealthService _health; + + public async Task InitializeAsync() + { + var services = new ServiceCollection(); + services.AddLogging(); + services.AddSingleton(_gate); + services + .AddConductorSharp(baseUrl: "http://conductor") + .AddExecutionManager( + maxConcurrentWorkers: 4, + sleepInterval: 250, + longPollInterval: 250, + domain: null, + typeof(EngineRecoveryTests).Assembly + ) + .UseBetaExecutionManager() + .SetHealthCheckService(); + services.RegisterWorkerTask(); + + services.ConfigureHttpClientDefaults( + http => http.ConfigurePrimaryHttpMessageHandler(() => _conductor).ConfigureHttpClient(c => c.Timeout = TimeSpan.FromMilliseconds(300)) + ); + services.AddSingleton( + sp => + new TaskQueuePollingService( + sp.GetRequiredService(), + sp.GetRequiredService>(), + sp.GetRequiredService(), + maxAttempts: 2, + initialRetryDelay: TimeSpan.FromMilliseconds(10), + maxRetryDelay: TimeSpan.FromMilliseconds(50), + maxJitter: TimeSpan.Zero, + delay: Task.Delay, + jitter: () => 0 + ) + ); + services.RemoveAll(); + services.AddTransient( + sp => + new WorkflowEngineBackgroundService( + sp.GetRequiredService(), + sp.GetRequiredService>(), + sp.GetRequiredService(), + sp.GetRequiredService(), + sp.GetRequiredService(), + initialRetryDelay: TimeSpan.FromMilliseconds(20), + maxRetryDelay: TimeSpan.FromMilliseconds(100), + maxJitter: TimeSpan.Zero, + delay: Task.Delay, + jitter: () => 0 + ) + ); + + _provider = services.BuildServiceProvider(); + _health = _provider.GetRequiredService(); + await _health.ResetHealthData(); + } + + public async Task DisposeAsync() + { + _gate.Open(); + foreach (var hostedService in _hostedServices) + await hostedService.StopAsync(CancellationToken.None); + await _provider.DisposeAsync(); + } + + [Fact] + public async Task Outage503_EngineResumesAndPicksUpTask() + { + await StartEngineAsync(); + await WaitUntil(() => Task.FromResult(_conductor.QueueReads > 0), "the engine polls"); + + _conductor.Mode = FakeConductor.Fail503; + await WaitUntil(IsNotRunning, "the engine backs off (Unhealthy)"); + _conductor.Mode = FakeConductor.Ok; + _conductor.Enqueue("after-outage"); + + await WaitUntil(() => IsCompleted("after-outage"), "the task queued after the outage completes"); + Assert.True((await _health.GetHealthData()).IsExecutionManagerRunning); + } + + [Fact] + public async Task Hang_EngineResumesAndPicksUpTask() + { + await StartEngineAsync(); + await WaitUntil(() => Task.FromResult(_conductor.QueueReads > 0), "the engine polls"); + + _conductor.Mode = FakeConductor.Hang; + await WaitUntil(IsNotRunning, "the engine backs off (Unhealthy) after request timeouts"); + _conductor.Mode = FakeConductor.Ok; + _conductor.Enqueue("after-hang"); + + await WaitUntil(() => IsCompleted("after-hang"), "the task queued after the hang completes"); + Assert.True((await _health.GetHealthData()).IsExecutionManagerRunning); + } + + [Fact] + public async Task QueuesFailWhileMetadataWorks_EngineStaysUnhealthy() + { + _conductor.Mode = FakeConductor.FailTasks; + await StartEngineAsync(); + await WaitUntil(() => Task.FromResult(_conductor.TaskDefRegistrations >= 3), "the engine has redeployed twice"); + + var samples = new List(); + for (var i = 0; i < 50; i++) + { + samples.Add((await _health.GetHealthData()).IsExecutionManagerRunning); + await Task.Delay(10); + } + + Assert.DoesNotContain(true, samples); + + _conductor.Mode = FakeConductor.Ok; + _conductor.Enqueue("after-queue-outage"); + await WaitUntil(() => IsCompleted("after-queue-outage"), "the task queued after the queue outage completes"); + Assert.True((await _health.GetHealthData()).IsExecutionManagerRunning); + } + + [Fact] + public async Task RegistrationRace_EngineStartsAfterRetry() + { + _conductor.FailNextTaskDefRegistration = true; + await StartEngineAsync(); + _conductor.Enqueue("after-race"); + + await WaitUntil(() => IsCompleted("after-race"), "the task completes after the failed registration"); + Assert.Equal(2, _conductor.TaskDefRegistrations); + } + + [Fact] + public async Task TaskInFlight_SurvivesLoopRestart() + { + await StartEngineAsync(); + _gate.Close(); + _conductor.Enqueue("in-flight"); + await WaitUntil(() => Task.FromResult(_conductor.HandedOut.Contains("in-flight")), "the engine picks up the task"); + + _conductor.Mode = FakeConductor.Fail503; + await WaitUntil(IsNotRunning, "the engine backs off (Unhealthy)"); + _conductor.Mode = FakeConductor.Ok; + await WaitUntil(async () => !await IsNotRunning(), "the engine runs again"); + Assert.False(await IsCompleted("in-flight")); + _gate.Open(); + + await WaitUntil(() => IsCompleted("in-flight"), "the task picked up before the outage completes"); + _conductor.Enqueue("after-restart"); + await WaitUntil(() => IsCompleted("after-restart"), "a new task completes after the restart"); + } + + private async Task StartEngineAsync() + { + _hostedServices = _provider.GetServices().ToList(); + foreach (var hostedService in _hostedServices) + await hostedService.StartAsync(CancellationToken.None); + } + + private async Task IsNotRunning() => !(await _health.GetHealthData()).IsExecutionManagerRunning; + + private Task IsCompleted(string taskId) => Task.FromResult(_conductor.Updates.TryGetValue(taskId, out var status) && status == "COMPLETED"); + + private static async Task WaitUntil(Func> condition, string what) + { + var stopwatch = Stopwatch.StartNew(); + while (stopwatch.Elapsed < WaitTimeout) + { + if (await condition()) + return; + await Task.Delay(20); + } + + Assert.Fail($"Timed out after {WaitTimeout.TotalSeconds} s waiting until {what}"); + } + + public class RecoveryProbe : IRequest + { + public class Response { } + + // The gate is optional: other tests scan this assembly for handlers and validate their own containers. + [OriginalName(ProbeTaskName)] + public class Handler(ProbeGate gate = null) : TaskRequestHandler + { + public override async Task Handle(RecoveryProbe request, CancellationToken cancellationToken) + { + if (gate != null) + await gate.WaitAsync(cancellationToken); + return new Response(); + } + } + } + + /// Lets a test hold the probe handler mid-execution. + public class ProbeGate + { + private volatile TaskCompletionSource _gate = Opened(); + + public void Close() => _gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + public void Open() => _gate.TrySetResult(); + + public Task WaitAsync(CancellationToken cancellationToken) => _gate.Task.WaitAsync(cancellationToken); + + private static TaskCompletionSource Opened() + { + var gate = new TaskCompletionSource(); + gate.SetResult(); + return gate; + } + } + + /// + /// Answers the Conductor endpoints the engine uses. switches between a healthy Conductor, one + /// that answers 503 to everything, and one that accepts requests but never answers (docker pause). + /// + private sealed class FakeConductor : HttpMessageHandler + { + public const string Ok = "ok"; + public const string Fail503 = "fail503"; + public const string Hang = "hang"; + public const string FailTasks = "failTasks"; // metadata works, every /api/tasks call answers 500 + + private readonly ConcurrentQueue _queued = new(); + private int _queueReads; + private int _taskDefRegistrations; + + public volatile string Mode = Ok; + public volatile bool FailNextTaskDefRegistration; + + public int QueueReads => Volatile.Read(ref _queueReads); + public int TaskDefRegistrations => Volatile.Read(ref _taskDefRegistrations); + public ConcurrentBag HandedOut { get; } = []; + public ConcurrentDictionary Updates { get; } = new(); + + public void Enqueue(string taskId) => _queued.Enqueue(taskId); + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + if (Mode == Fail503) + return new HttpResponseMessage(HttpStatusCode.ServiceUnavailable); + if (Mode == Hang) + await Task.Delay(Timeout.Infinite, cancellationToken); + + var path = request.RequestUri!.AbsolutePath; + + if (Mode == FailTasks && path.StartsWith("/api/tasks")) + return Json(HttpStatusCode.InternalServerError, new { status = 500, message = "queue backend unavailable" }); + + if (request.Method == HttpMethod.Post && path == "/api/metadata/taskdefs") + { + Interlocked.Increment(ref _taskDefRegistrations); + if (!FailNextTaskDefRegistration) + return new HttpResponseMessage(HttpStatusCode.OK); + + FailNextTaskDefRegistration = false; + return Json( + HttpStatusCode.InternalServerError, + new { status = 500, message = "ERROR: duplicate key value violates unique constraint \"meta_task_def_pkey\"" } + ); + } + + if (request.Method == HttpMethod.Get && path == "/api/tasks/queue/all") + { + Interlocked.Increment(ref _queueReads); + return Json(HttpStatusCode.OK, new Dictionary { [ProbeTaskName] = _queued.Count }); + } + + if (request.Method == HttpMethod.Get && path == $"/api/tasks/poll/{ProbeTaskName}") + { + if (!_queued.TryDequeue(out var taskId)) + return new HttpResponseMessage(HttpStatusCode.NoContent); + + HandedOut.Add(taskId); + return Json( + HttpStatusCode.OK, + new + { + taskId, + taskDefName = ProbeTaskName, + taskType = ProbeTaskName, + referenceTaskName = "probe", + workflowInstanceId = $"workflow-{taskId}", + workflowType = "recovery_probe_workflow", + status = "IN_PROGRESS", + inputData = new { } + } + ); + } + + if (request.Method == HttpMethod.Post && path == "/api/tasks") + { + var result = JObject.Parse(await request.Content!.ReadAsStringAsync(cancellationToken)); + var taskId = (string)result["taskId"]; + Updates[taskId] = (string)result["status"]; + return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(taskId) }; + } + + return new HttpResponseMessage(HttpStatusCode.NotFound); + } + + private static HttpResponseMessage Json(HttpStatusCode statusCode, object body) => + new(statusCode) { Content = new StringContent(JsonConvert.SerializeObject(body), Encoding.UTF8, "application/json") }; + } +} diff --git a/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs b/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs index 38db4d24..af363381 100644 --- a/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs +++ b/test/ConductorSharp.Engine.Tests/Unit/TaskQueuePollingServiceTests.cs @@ -2,14 +2,40 @@ using System.Text; using ConductorSharp.Client.Generated; using ConductorSharp.Client.Service; +using ConductorSharp.Engine.Health; using ConductorSharp.Engine.Service; using Microsoft.Extensions.Logging.Abstractions; using Task = System.Threading.Tasks.Task; namespace ConductorSharp.Engine.Tests.Unit; +[Collection(nameof(InMemoryHealthService))] // InMemoryHealthService keeps its flag in a static field public class TaskQueuePollingServiceTests { + private readonly InMemoryHealthService _healthService = new(); + + public TaskQueuePollingServiceTests() => _healthService.ResetHealthData().Wait(); + + [Fact] + public async Task ListQueuesAsync_MarksEngineRunning_AfterSuccessfulRead() + { + var service = CreateService(new SequenceHandler(HttpStatusCode.InternalServerError, HttpStatusCode.OK), []); + + await service.ListQueuesAsync(CancellationToken.None); + + Assert.True((await _healthService.GetHealthData()).IsExecutionManagerRunning); + } + + [Fact] + public async Task ListQueuesAsync_DoesNotMarkEngineRunning_WhenReadsFail() + { + var service = CreateService(new SequenceHandler(HttpStatusCode.Unauthorized), []); + + await Assert.ThrowsAsync(() => service.ListQueuesAsync(CancellationToken.None)); + + Assert.False((await _healthService.GetHealthData()).IsExecutionManagerRunning); + } + [Fact] public async Task ListQueuesAsync_RetriesTransientFailuresAndReturnsQueues() { @@ -58,7 +84,7 @@ public async Task ListQueuesAsync_RethrowsAfterRetryLimit() Assert.Equal(4, delays.Count); } - private static TaskQueuePollingService CreateService(HttpMessageHandler handler, ICollection delays) + private TaskQueuePollingService CreateService(HttpMessageHandler handler, ICollection delays) { var httpClient = new HttpClient(handler) { BaseAddress = new Uri("http://conductor/") }; var taskService = new TaskService(httpClient); @@ -66,6 +92,7 @@ private static TaskQueuePollingService CreateService(HttpMessageHandler handler, return new TaskQueuePollingService( taskService, NullLogger.Instance, + _healthService, maxAttempts: 5, initialRetryDelay: TimeSpan.FromSeconds(1), maxRetryDelay: TimeSpan.FromSeconds(30), diff --git a/test/ConductorSharp.Engine.Tests/Unit/WorkflowEngineBackgroundServiceTests.cs b/test/ConductorSharp.Engine.Tests/Unit/WorkflowEngineBackgroundServiceTests.cs new file mode 100644 index 00000000..4a3f52d0 --- /dev/null +++ b/test/ConductorSharp.Engine.Tests/Unit/WorkflowEngineBackgroundServiceTests.cs @@ -0,0 +1,322 @@ +using ConductorSharp.Client.Generated; +using ConductorSharp.Engine.Health; +using ConductorSharp.Engine.Service; +using Microsoft.Extensions.Logging; +using Task = System.Threading.Tasks.Task; + +namespace ConductorSharp.Engine.Tests.Unit; + +[Collection(nameof(InMemoryHealthService))] // InMemoryHealthService keeps its flag in a static field +public class WorkflowEngineBackgroundServiceTests +{ + private static readonly TimeSpan WaitTimeout = TimeSpan.FromSeconds(5); + + private readonly InMemoryHealthService _healthService = new(); + private readonly List _delays = []; + private readonly ListLogger _logger = new(); + + public WorkflowEngineBackgroundServiceTests() => _healthService.ResetHealthData().Wait(); + + [Fact] + public async Task DeployApiError_IsRetriedAndEngineStarts() + { + var deployment = new ScriptedDeploymentService(Api(500), Api(409)); + var executionManager = new ScriptedExecutionManager(_healthService); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.Equal(3, deployment.Calls); + Assert.Equal(1, executionManager.Calls); + Assert.Equal(new[] { 1, 2 }.Select(s => TimeSpan.FromSeconds(s)), _delays); + Assert.True((await _healthService.GetHealthData()).IsExecutionManagerRunning); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task ConnectionError_WhileRunning_RestartsEngine() + { + var deployment = new ScriptedDeploymentService(); + var executionManager = new ScriptedExecutionManager(_healthService, Fails(new HttpRequestException("Connection refused"))); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.Equal(2, deployment.Calls); + Assert.Equal(2, executionManager.Calls); + Assert.True((await _healthService.GetHealthData()).IsExecutionManagerRunning); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task HttpTimeout_IsAnErrorNotAShutdown() + { + var deployment = new ScriptedDeploymentService(); + var executionManager = new ScriptedExecutionManager(_healthService, Fails(new TaskCanceledException("timeout", new TimeoutException()))); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.Equal(2, executionManager.Calls); + Assert.Single(_delays); + Assert.Contains(_logger.Entries, e => e.Level == LogLevel.Error); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task HealthIsNotRunning_WhileBackingOff() + { + var deployment = new ScriptedDeploymentService(); + var executionManager = new ScriptedExecutionManager(_healthService, PollsThenFails(new HttpRequestException("Connection refused"))); + var backingOff = new TaskCompletionSource(); + var resume = new TaskCompletionSource(); + var service = CreateService( + deployment, + executionManager, + (delay, cancellationToken) => + { + backingOff.TrySetResult(); + return resume.Task.WaitAsync(cancellationToken); + } + ); + + await service.StartAsync(CancellationToken.None); + await backingOff.Task.WaitAsync(WaitTimeout); + + Assert.False((await _healthService.GetHealthData()).IsExecutionManagerRunning); + + resume.SetResult(); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.True((await _healthService.GetHealthData()).IsExecutionManagerRunning); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task NonTransientDeployError_IsRetriedWithCappedBackoff() + { + var deployment = new ScriptedDeploymentService(Enumerable.Range(0, 8).Select(_ => Api(401)).ToArray()); + var executionManager = new ScriptedExecutionManager(_healthService); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.Equal(new[] { 1, 2, 4, 8, 16, 30, 30, 30 }.Select(s => TimeSpan.FromSeconds(s)), _delays); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task SuccessfulPoll_ResetsBackoff() + { + var deployment = new ScriptedDeploymentService(Api(500), Api(500), Api(500)); + var executionManager = new ScriptedExecutionManager(_healthService, PollsThenFails(new HttpRequestException("Connection refused"))); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.Equal(new[] { 1, 2, 4, 1 }.Select(s => TimeSpan.FromSeconds(s)), _delays); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task SuccessfulDeployWithoutPoll_KeepsBackingOff() + { + // Metadata works but the task queues fail (e.g. Conductor's queue backend is down). + var deployment = new ScriptedDeploymentService(); + var executionManager = new ScriptedExecutionManager(_healthService, Fails(Api(500)), Fails(Api(500)), Fails(Api(500))); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.Equal(new[] { 1, 2, 4 }.Select(s => TimeSpan.FromSeconds(s)), _delays); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task SuccessfulDeploy_DoesNotMarkEngineRunning() + { + var deployment = new ScriptedDeploymentService(); + var executionManager = new ScriptedExecutionManager(_healthService) { PollsWhenRunning = false }; + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + + Assert.False((await _healthService.GetHealthData()).IsExecutionManagerRunning); + + await service.StopAsync(CancellationToken.None); + } + + [Fact] + public async Task Stop_WhileDeployHangs_ReturnsPromptly() + { + var deployment = new ScriptedDeploymentService { Hangs = true }; + var executionManager = new ScriptedExecutionManager(_healthService); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await deployment.Called.Task.WaitAsync(WaitTimeout); + await service.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(1)); + + Assert.Equal(0, executionManager.Calls); + Assert.Contains(_logger.Entries, e => e.Message == "Stopping ConductorSharp background service"); + } + + [Fact] + public async Task Stop_DuringBackoff_ExitsWithoutRetry() + { + var deployment = new ScriptedDeploymentService(Api(500)); + var executionManager = new ScriptedExecutionManager(_healthService); + var backingOff = new TaskCompletionSource(); + var service = CreateService( + deployment, + executionManager, + (delay, cancellationToken) => + { + backingOff.TrySetResult(); + return Task.Delay(Timeout.Infinite, cancellationToken); + } + ); + + await service.StartAsync(CancellationToken.None); + await backingOff.Task.WaitAsync(WaitTimeout); + await service.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(1)); + + Assert.Equal(1, deployment.Calls); + Assert.Equal(0, executionManager.Calls); + Assert.Contains(_logger.Entries, e => e.Message == "Stopping ConductorSharp background service"); + } + + [Fact] + public async Task Stop_WhileRunning_DoesNotRestart() + { + var deployment = new ScriptedDeploymentService(); + var executionManager = new ScriptedExecutionManager(_healthService); + var service = CreateService(deployment, executionManager, RecordDelay); + + await service.StartAsync(CancellationToken.None); + await executionManager.Running.Task.WaitAsync(WaitTimeout); + await service.StopAsync(CancellationToken.None).WaitAsync(TimeSpan.FromSeconds(1)); + + Assert.Equal(1, executionManager.Calls); + Assert.Empty(_delays); + Assert.DoesNotContain(_logger.Entries, e => e.Level >= LogLevel.Error); + } + + private Task RecordDelay(TimeSpan delay, CancellationToken cancellationToken) + { + _delays.Add(delay); + return Task.CompletedTask; + } + + private WorkflowEngineBackgroundService CreateService( + IDeploymentService deploymentService, + IExecutionManager executionManager, + Func delay + ) => + new( + _healthService, + _logger, + deploymentService, + executionManager, + new ModuleDeployment([], [], []), + initialRetryDelay: TimeSpan.FromSeconds(1), + maxRetryDelay: TimeSpan.FromSeconds(30), + maxJitter: TimeSpan.FromMilliseconds(500), + delay: delay, + jitter: () => 0 + ); + + private static ApiException Api(int statusCode) => + new($"HTTP {statusCode}", statusCode, "{}", new Dictionary>(), null); + + private static Run Fails(Exception failure) => new(failure, PollsFirst: false); + + private static Run PollsThenFails(Exception failure) => new(failure, PollsFirst: true); + + private sealed record Run(Exception Failure, bool PollsFirst); + + private sealed class ScriptedDeploymentService(params Exception[] failures) : IDeploymentService + { + private readonly Queue _failures = new(failures); + + public int Calls { get; private set; } + + /// When set, Deploy never completes (a hung Conductor request). + public bool Hangs { get; init; } + + public TaskCompletionSource Called { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public Task Deploy(Deployment deployment) + { + Calls++; + Called.TrySetResult(); + if (Hangs) + return new TaskCompletionSource().Task; + return _failures.TryDequeue(out var failure) ? Task.FromException(failure) : Task.CompletedTask; + } + + public Task Remove(Deployment deployment) => Task.CompletedTask; + } + + /// + /// Plays the queued runs, one per call. Once they run out, it reads the queues successfully (marks the engine + /// running, as does) and blocks until cancelled. + /// + private sealed class ScriptedExecutionManager(IConductorSharpHealthService healthService, params Run[] runs) : IExecutionManager + { + private readonly Queue _runs = new(runs); + + public int Calls { get; private set; } + + /// When cleared, the final run blocks without a successful queue read. + public bool PollsWhenRunning { get; init; } = true; + + public TaskCompletionSource Running { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); + + public async Task StartAsync(CancellationToken cancellationToken) + { + Calls++; + if (_runs.TryDequeue(out var run)) + { + if (run.PollsFirst) + await healthService.SetExecutionManagerRunning(cancellationToken); + throw run.Failure; + } + + if (PollsWhenRunning) + await healthService.SetExecutionManagerRunning(cancellationToken); + Running.TrySetResult(); + await Task.Delay(Timeout.Infinite, cancellationToken); + } + } + + private sealed class ListLogger : ILogger + { + public List<(LogLevel Level, string Message)> Entries { get; } = []; + + public IDisposable BeginScope(TState state) + where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception exception, Func formatter) + { + lock (Entries) + Entries.Add((logLevel, formatter(state, exception))); + } + } +}